并行处理文件中的数据

Question

我正面临一个来自大型 csv 文件的并行计算数据问题。问题是不能并行读取文件，但可以传递来自文件的数据块以进行并行计算。我尝试使用 Multiprocessing.Pool 没有结果（Pool.imap 不接受产量生成器）。

我有一个从文件中读取数据块的生成器。它需要大约。 3 秒。从文件中获取一大块数据。这块数据被处理了大约需要 ca。 2 秒。我从文件中获得 50 块数据。等待下一个文件块我可以计算前一个块“并行”。

让我们有一些概念上的代码（但在实践中不起作用）。：

def file_data_generator(path):
    # file reading chunk by chunk 
    yield datachunk

def compute(datachunk):
    # some heavy computation 2.sec
    return partial_result

from multiprocessing import Pool
p = Pool()
result = p.imap(compute, file_data_generator(path) ) # yield is the issue?

我做错了什么？我应该使用其他任何工具吗？ 是 Python3.5

简单的代码概念/骨架赞赏:)

Answer 1

你们很亲密。 yield 的生成器位是正确的：imap 确实将生成器作为参数并在其上运行 next()，因此 yield 在此上下文中是正确的。

您缺少的是 imap 没有阻塞，这意味着即使进程尚未完成，result = p.imap 调用也会返回。你要么需要做

p.close()
p.join()

然后将results 作为一个整体做一些事情，或者您只需对结果进行迭代。这是一个工作示例：

from multiprocessing import Pool, Queue

def compute(line):
    # some heavy computation 2.sec
    return len(line)

def file_data_generator(path):
    # file reading chunk by chunk 
    with open('book.txt') as f:
        for line in f:
            yield line.strip()

if __name__ == '__main__':
    p = Pool()
    # start processes, they are still blocked because queue is empty
    # results is a generator and is empty at the start
    results = p.imap(compute, file_data_generator('book.txt'))

    # now we tell pool that we finished filling the queue
    p.close()
    for res in results:
        print(res)