将 CSV 文件分成相等的部分？

Question

我有一个大的 CSV 文件，我想将其拆分为一个等于系统中 CPU 内核数的数字。然后我想使用多进程让所有核心一起处理文件。但是，我什至无法将文件分成几部分。我查看了整个谷歌，发现一些示例代码似乎可以满足我的需求。这是我目前所拥有的：

def split(infilename, num_cpus=multiprocessing.cpu_count()):
    READ_BUFFER = 2**13
    total_file_size = os.path.getsize(infilename)
    print total_file_size
    files = list()
    with open(infilename, 'rb') as infile:
        for i in xrange(num_cpus):
            files.append(tempfile.TemporaryFile())
            this_file_size = 0
            while this_file_size < 1.0 * total_file_size / num_cpus:
                files[-1].write(infile.read(READ_BUFFER))
                this_file_size += READ_BUFFER
        files[-1].write(infile.readline()) # get the possible remainder
        files[-1].seek(0, 0)
    return files

files = split("sample_simple.csv")
print len(files)

for ifile in files:
    reader = csv.reader(ifile)
    for row in reader:
        print row

这两个打印显示了正确的文件大小，并且它被分成了 4 个部分（我的系统有 4 个 CPU 内核）。

但是，打印每个片段中所有行的代码的最后一部分给出了错误：

for row in reader:
_csv.Error: line contains NULL byte

我尝试在不运行拆分功能的情况下打印行，它正确打印了所有值。我怀疑 split 函数在生成的 4 个文件片段中添加了一些 NULL 字节，但我不知道为什么。

有谁知道这是否是一种正确且快速的文件分割方法？我只想要 csv.reader 可以成功读取的结果片段。

Answer 1

正如我在评论中所说，csv 文件需要在行（或行）边界上拆分。您的代码没有这样做，并且可能会在其中的某个地方将它们分解 - 我怀疑这是您的 _csv.Error 的原因。

下面通过将输入文件处理为一系列行来避免这样做。我已经对其进行了测试，它似乎可以独立工作，因为它将示例文件分成 大约 大小相等的块，因为整个行数不太可能完全适合一个块。

更新

这是一个大大比我最初发布的代码更快的版本。改进是因为它现在使用临时文件自己的tell() 方法来确定正在写入的文件的不断变化的长度，而不是调用os.path.getsize()，这消除了flush() 文件和调用os.fsync() 的需要写完每一行后在上面。

import csv
import multiprocessing
import os
import tempfile

def split(infilename, num_chunks=multiprocessing.cpu_count()):
    READ_BUFFER = 2**13
    in_file_size = os.path.getsize(infilename)
    print 'in_file_size:', in_file_size
    chunk_size = in_file_size // num_chunks
    print 'target chunk_size:', chunk_size
    files = []
    with open(infilename, 'rb', READ_BUFFER) as infile:
        for _ in xrange(num_chunks):
            temp_file = tempfile.TemporaryFile()
            while temp_file.tell() < chunk_size:
                try:
                    temp_file.write(infile.next())
                except StopIteration:  # end of infile
                    break
            temp_file.seek(0)  # rewind
            files.append(temp_file)
    return files

files = split("sample_simple.csv", num_chunks=4)
print 'number of files created: {}'.format(len(files))

for i, ifile in enumerate(files, start=1):
    print 'size of temp file {}: {}'.format(i, os.path.getsize(ifile.name))
    print 'contents of file {}:'.format(i)
    reader = csv.reader(ifile)
    for row in reader:
        print row
    print ''