Python:使用多个核心的流程文件

Question

我目前正在尝试读取一个大文件(8000万行),我需要为每个条目进行计算密集型矩阵乘法.计算完之后,我想将结果插入数据库.由于此过程采用时间密集的方式,我希望将文件拆分为多个核心以加快进程.

在研究之后,我发现了这个有希望的尝试,它将文件分成n个部分.

def file_block(fp, number_of_blocks, block):
    '''
    A generator that splits a file into blocks and iterates
    over the lines of one of the blocks.

    '''

    assert 0 <= block and block < number_of_blocks
    assert 0 < number_of_blocks

    fp.seek(0,2)
    file_size = fp.tell()

    ini = file_size * block / number_of_blocks
    end = file_size * (1 + block) / number_of_blocks

    if ini <= 0:
        fp.seek(0)
    else:
        fp.seek(ini-1)
        fp.readline()

    while fp.tell() < end:
        yield fp.readline()

迭代地,您可以像这样调用函数:

if __name__ == '__main__':
    fp = open(filename)
    number_of_chunks = 4
    for chunk_number in range(number_of_chunks):
        print chunk_number, 100 * '='
        for line in file_block(fp, number_of_chunks, chunk_number):
            process(line)

虽然这有效,但我遇到了问题,使用多处理并行化:

fp = open(filename)
number_of_chunks = 4
li = [file_block(fp, number_of_chunks, chunk_number) for chunk_number in range(number_of_chunks)]

p = Pool(cpu_count() - 1)
p.map(processChunk,li)

由于错误,生成器无法被腌制.

虽然我理解这个错误,但是首先迭代整个文件以将所有行放入列表中是太昂贵了.

此外,我希望每次迭代使用每个核心的行块,因为一次将多行插入数据库更有效(如果使用典型的映射方法,则不是1乘1)

谢谢你的帮助.

Answer 1

不要预先创建生成器并将它们传递到每个线程中，而是将其留给线程代码。

def processChunk(params):
    filename, chunk_number, number_of_chunks = params
    with open(filename, 'r') as fp:
        for line in file_block(fp, number_of_chunks, chunk_number):
            process(line)

li = [(filename, i, number_of_chunks) for i in range(number_of_chunks)]
p.map(processChunk, li)