如何在 Python 3 中重用进程池进行并行编程

Question

我是并行编程的新手。我的任务是分析数百个数据文件。这些数据中的每一个都接近 300MB，并且可以被分割成许多片。我的电脑是四核电脑。我想尽快得到每个数据的结果。
每个数据文件的分析包括 2 个过程。首先，将数据读入内存，然后将其切片，这是io密集型工作。然后，对该文件的切片进行大量计算，这是 cpu 密集型的。
所以我的策略是将这些文件分组为 4 个一组。对于这些文件的每一组，首先将 4 个文件的所有数据读取到内存中，其中包含 4 个内核中的 4 个进程。代码是这样的，

with Pool(processes=4) as pool:
    data_list = pool.map(read_and_slice, files)  # len(files)==4

然后对于data_list 中的每个data，使用4 个进程进行计算。

for data in data_list:  # I want to get the result of each data asap
    with Pool(processes=4) as pool:
        result_list = pool.map(compute, data.slices)  # anaylyze each slice of data
    analyze(result_list)  # analyze the results of previous procedure, for example, get the average.

然后去另一个小组。
所以问题是在数百个文件的整个计算过程中，池被重新创建了很多次。如何避免重新创建池和进程的开销？我的代码中是否存在大量内存开销？有没有更好的方法让我尽可能减少所需的时间？

谢谢！

Answer 1

一种选择是将with Pool 语句移到for 循环之外……

p = Pool()
for data in data_list:
  result_list = pool.map(compute, data.slices)
  analyze(result_list)
p.join()
p.close()

这适用于 python 2 或 3。

如果您安装（我的模块）pathos，然后执行from pathos.pools import ProcessPool as Pool，并保持其余代码与您所拥有的完全相同——您将只创建一个Pool。这是因为pathos 缓存了Pool，当创建具有相同配置的新Pool 实例时，它只是重用现有实例。您可以通过pool.terminate() 关闭它。

>>> from pathos.pools import ProcessPool as Pool
>>> pool = Pool()
>>> data_list = [range(4), range(4,8), range(8,12), range(12,16)]
>>> squared = lambda x:x**2
>>> mean = lambda x: sum(x)/len(x)
>>> for data in data_list:
...   result = pool.map(squared, data)
...   print mean(result)
... 
3
31
91
183

实际上，pathos 使您能够执行嵌套池，因此您还可以将您的 for 循环转换为异步映射（amap 来自 pathos）......并且由于内部映射不需要保留为了您可以使用无序映射迭代器（multiprocessing 中的imap_unordered，或pathos 中的uimap）。例如，请参见此处： https://stackoverflow.com/questions/28203774/how-to-do-hierarchical-parallelism-in-ipython-parallel 在这里： https://stackoverflow.com/a/31617653/2379433

只有 pathos 是 python2。但很快（待发布）将完全转换为python3。