我有一个7GB的csv
文件,我想将它分成较小的块,这样在笔记本上使用Python进行分析时更易于阅读和更快。 我想从中获取一个小的集合,可能是250MB,那么我该如何做呢?
$ split -l 100 data.csv
将data.csv
文件分成每个包含100行的块。
我曾经需要完成一个类似的任务,使用了 pandas 库:
for i,chunk in enumerate(pd.read_csv('bigfile.csv', chunksize=500000)):
chunk.to_csv('chunk{}.csv'.format(i), index=False)
chunksize
指定的行数。 - Imad以下是我用的一个小Python脚本,用于将文件data.csv
拆分为多个CSV部分文件。可以使用chunk_size
(每个部分文件的行数)控制部分文件的数量。
原始文件的标题行(列名)将被复制到每个部分CSV文件中。
它适用于大文件,因为它使用readline()
一次读取一行,而不是一次性将整个文件加载到内存中。
#!/usr/bin/env python3
def main():
chunk_size = 9998 # lines
def write_chunk(part, lines):
with open('data_part_'+ str(part) +'.csv', 'w') as f_out:
f_out.write(header)
f_out.writelines(lines)
with open('data.csv', 'r') as f:
count = 0
header = f.readline()
lines = []
for line in f:
count += 1
lines.append(line)
if count % chunk_size == 0:
write_chunk(count // chunk_size, lines)
lines = []
# write remainder
if len(lines) > 0:
write_chunk((count // chunk_size) + 1, lines)
if __name__ == '__main__':
main()
这里有四种不同的编程方法,其中Shell方法来自Thomas Orozco,Python方法来自Roberto,Pandas方法来自Quentin Febvre,以下是Dask代码片段:
ddf = dd.read_csv("../nyc-parking-tickets/Parking_Violations_Issued_-_Fiscal_Year_2015.csv", blocksize=10000000, dtype=dtypes)
ddf.to_csv("../tmp/split_csv_dask")
#!/usr/local/cpython-3.3/bin/python
import csv
divisor = 10
outfileno = 1
outfile = None
with open('big.csv', 'r') as infile:
for index, row in enumerate(csv.reader(infile)):
if index % divisor == 0:
if outfile is not None:
outfile.close()
outfilename = 'big-{}.csv'.format(outfileno)
outfile = open(outfilename, 'w')
outfileno += 1
writer = csv.writer(outfile)
writer.writerow(row)
这是我的代码,可能会有所帮助
import os
import pandas as pd
import uuid
class FileSettings(object):
def __init__(self, file_name, row_size=100):
self.file_name = file_name
self.row_size = row_size
class FileSplitter(object):
def __init__(self, file_settings):
self.file_settings = file_settings
if type(self.file_settings).__name__ != "FileSettings":
raise Exception("Please pass correct instance ")
self.df = pd.read_csv(self.file_settings.file_name,
chunksize=self.file_settings.row_size)
def run(self, directory="temp"):
try:os.makedirs(directory)
except Exception as e:pass
counter = 0
while True:
try:
file_name = "{}/{}_{}_row_{}_{}.csv".format(
directory, self.file_settings.file_name.split(".")[0], counter, self.file_settings.row_size, uuid.uuid4().__str__()
)
df = next(self.df).to_csv(file_name)
counter = counter + 1
except StopIteration:
break
except Exception as e:
print("Error:",e)
break
return True
def main():
helper = FileSplitter(FileSettings(
file_name='sample1.csv',
row_size=10
))
helper.run()
main()
df = next(self.df).to_csv(file_name, index=False)
,以避免在第一列写入行名称(索引)。 - DFX Nguyễn我同意 @jonrsharpe 的观点,readline 应该能够逐行读取大文件。
如果你正在处理大型 csv 文件,我建议使用 pandas.read_csv。我经常为相同的目的使用它,并且总是发现它很棒(而且快)。需要一些时间来习惯 DataFrame 的概念。但是一旦你克服了这个问题,它就可以大大加速像你这样的大型操作。
希望这有所帮助。
如果想按字节的粗略边界进行分割,最新的数据点在最底部,并希望将最新的数据点放在第一个文件中:
from pathlib import Path
TEN_MB = 10000000
FIVE_MB = 5000000
def split_file_into_chunks(path, chunk_size=TEN_MB):
path = str(path)
output_prefix = path.rpartition('.')[0]
output_ext = path.rpartition('.')[-1]
with open(path, 'rb') as f:
seek_positions = []
for x, line in enumerate(f):
if not x:
header = line
seek_positions.append(f.tell())
part = 0
last_seek_pos = seek_positions[-1]
for seek_pos in reversed(seek_positions):
if last_seek_pos-seek_pos >= chunk_size:
with open(f'{output_prefix}.arch.{part}.{output_ext}', 'wb') as f_out:
f.seek(seek_pos)
f_out.write(header)
f_out.write(f.read(last_seek_pos-seek_pos))
last_seek_pos = seek_pos
part += 1
with open(f'{output_prefix}.arch.{part}.{output_ext}', 'wb') as f_out:
f.seek(0)
f_out.write(f.read(last_seek_pos))
Path(path).rename(path+'~')
Path(f'{output_prefix}.arch.0.{output_ext}').rename(path)
Path(path+'~').unlink()
split -l 100 data.csv data_split_
。 - Matt