使用多进程传递大量数据

Question

我正在尝试编写一个程序，以并行方式执行计算，以便每个计算的结果可以按特定顺序写入文件。我的问题是规模；我想做我在下面示例程序中概述的内容 - 将大输出保存为字典的值，该字典在其键中存储排序系统。但我的程序一直崩溃，因为它无法存储/传递如此多的字节。

有没有固定的方法来解决这类问题？我对处理多进程和大型数据都很陌生。

from multiprocessing import Process, Manager

def eachProcess(i, d):
    LARGE_BINARY_OBJECT = #perform some computation resulting in millions of bytes
    d[i] = LARGE_BINARY_OBJECT
def main():
    manager = Manager()
    d = manager.dict()
    maxProcesses = 10
    for i in range(maxProcesses):
        process = Process(target=eachProcess, args=(i,d))
        process.start()

    counter = 0
    while counter < maxProcesses:
        file1 = open("test.txt", "wb")
        if counter in d:
            file1.write(d[counter])
            counter += 1

if __name__ == '__main__':
    main()

谢谢你。

- Zoe Christie

---

我写了这个库：https://github.com/JaggedVerge/mmap_backed_array，用于在多个进程之间传递内存的非常相似的用例。虽然它是相当底层的，但你也可以考虑类似Redis的东西。 - shuttle87

Answer 1

处理大数据通常有两种方式：

1. 如果问题足够简单，可以使用本地文件系统。
2. 如果需要更复杂的数据支持，则使用远程数据存储。

由于您的问题似乎很简单，我建议采用以下解决方案。每个进程将其部分解决方案写入本地文件中。一旦所有处理完成，主进程将所有结果文件合并在一起。

from multiprocessing import Pool
from tempfile import NamedTemporaryFile

def worker_function(partial_result_path):
    data = produce_large_binary()
    with open(partial_result_path, 'wb') as partial_result_file:
        partial_result_file.write(data)

# storing partial results in temporary files
partial_result_paths = [NamedTemporaryFile() for i in range(max_processes)]

pool = Pool(max_processes)
pool.map(worker_function, partial_result_paths)

with open('test.txt', 'wb') as result_file:
    for partial_result_path in partial_result_paths:
        with open(partial_result_path) as partial_result_file:
            result_file.write(partial_result_file.read())

- noxdafox

---

谢谢！实际上我之前写了一半类似这样的东西，但是停下来了，因为我认为使用的解决方案不够优雅。看起来我要重新回去写那个了。 - Zoe Christie

Python的经验法则是：“如果看起来简单，那么它肯定也很优雅。” - noxdafox