在Python中并行处理大型.csv文件

Question

我正在使用Python脚本处理大型CSV文件(大约有几GB的10M行).

这些文件具有不同的行长度,无法完全加载到内存中进行分析.

每行都由我的脚本中的函数单独处理.分析一个文件大约需要20分钟,看起来磁盘访问速度不是问题,而是处理/函数调用.

代码看起来像这样(非常简单).实际代码使用Class结构,但这类似:

csvReader = csv.reader(open("file","r")
for row in csvReader:
   handleRow(row, dataStructure)

鉴于计算需要共享数据结构,使用多个内核在Python中并行运行分析的最佳方法是什么？

一般来说,如何从Python中的.csv一次读取多行以传输到线程/进程？for在行上方循环听起来效率不高.

谢谢!

Answer 1

这可能为时已晚,但对于未来的用户,无论如何我都会发布.另一张提到使用多处理的海报.我可以保证它,并可以更详细.我们每天使用Python处理数百MB /几GB的文件.所以这绝对取决于任务.我们处理的一些文件不是CSV,因此解析可能相当复杂,并且需要比磁盘访问更长的时间.但是,无论使用何种文件类型,方法都是相同的.

您可以同时处理大型文件的各个部分.这是我们如何做到的伪代码:

import os, multiprocessing as mp

# process file function
def processfile(filename, start=0, stop=0):
    if start == 0 and stop == 0:
        ... process entire file...
    else:
        with open(file, 'r') as fh:
            fh.seek(start)
            lines = fh.readlines(stop - start)
            ... process these lines ...

    return results

if __name__ == "__main__":

    # get file size and set chuck size
    filesize = os.path.getsize(filename)
    split_size = 100*1024*1024

    # determine if it needs to be split
    if filesize > split_size:

        # create pool, initialize chunk start location (cursor)
        pool = mp.Pool(cpu_count)
        cursor = 0
        results = []
        with open(file, 'r') as fh:

            # for every chunk in the file...
            for chunk in xrange(filesize // split_size):

                # determine where the chunk ends, is it the last one?
                if cursor + split_size > filesize:
                    end = filesize
                else:
                    end = cursor + split_size

                # seek to end of chunk and read next line to ensure you 
                # pass entire lines to the processfile function
                fh.seek(end)
                fh.readline()

                # get current file location
                end = fh.tell()

                # add chunk to process pool, save reference to get results
                proc = pool.apply_async(processfile, args=[filename, cursor, end])
                results.append(proc)

                # setup next chunk
                cursor = end

        # close and wait for pool to finish
        pool.close()
        pool.join()

        # iterate through results
        for proc in results:
            processfile_result = proc.get()

    else:
        ...process normally...

就像我说的那样,这只是伪代码.它应该让任何人开始需要做类似的事情.我没有在我面前的代码,只是从内存中做到这一点.

但是在第一次运行时我们的速度提高了2倍以上而没有进行微调.您可以根据您的设置微调池中的进程数以及块的大小以获得更高的速度.如果您有多个文件,请创建一个池以并行读取多个文件.小心不要用太多的进程重载盒子.

注意:您需要将其放在"if main"块中,以确保不会创建无限的进程.

Answer 2

由于GIL,Python的线程不会像IO绑定那样加速处理器绑定的计算.

相反,请查看多处理模块,该模块可以并行运行多个处理器上的代码.

Answer 3

尝试进行基准测试以读取文件并解析每个CSV行，但不执行任何操作。您排除了磁盘访问的可能性，但是仍然需要查看CSV解析是缓慢的还是您自己的代码缓慢。

如果CSV解析很慢，您可能会被卡住，因为我认为没有一种方法可以跳入CSV文件的中间而无需进行扫描。

如果是您自己的代码，则可以让一个线程读取CSV文件并将行放入队列，然后让多个线程处理该队列中的行。但是，如果CSV解析本身使速度变慢，则不必理会此解决方案。