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

Question

我是并行编程的新手.我的任务是分析数百个数据文件.这些数据中的每一个都接近300MB,并且可以切成多个切片.我的电脑是一台4芯电脑.我希望尽快得到每个数据的结果.
每个数据文件的分析包括2个程序.首先,将数据读入内存,然后将其切片为切片,这是密集型工作.然后,对此文件的切片进行大量计算,这是cpu密集型的.
所以我的策略是将这些文件分组为4组.对于这些文件的每一组,首先,将4个文件的所有数据读入内存,4个进程中有4个进程.代码就像,

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

然后对于每个data在data_list,做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\xe2\x80\xa6

\n\n

p = Pool()\nfor data in data_list:\n  result_list = pool.map(compute, data.slices)\n  analyze(result_list)\np.join()\np.close()\n

\n\n

这适用于 python 2 或 3。

\n\n

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

\n\n

>>> from pathos.pools import ProcessPool as Pool\n>>> pool = Pool()\n>>> data_list = [range(4), range(4,8), range(8,12), range(12,16)]\n>>> squared = lambda x:x**2\n>>> mean = lambda x: sum(x)/len(x)\n>>> for data in data_list:\n...   result = pool.map(squared, data)\n...   print mean(result)\n... \n3\n31\n91\n183\n

\n\n

实际上，pathos使您能够执行嵌套池，因此您还可以将for循环转换为异步映射（amap来自pathos）\xe2\x80\xa6，并且由于内部映射不需要保留顺序，因此您可以使用无序映射迭代器（imap_unordered在multiprocessing，或uimap从pathos）。有关示例，请参见此处：\n /sf/ask/1974264211/和此处：\n https://stackoverflow.com/a/ 31617653/2379433

\n\n

唯一令人遗憾的pathos是python2。但很快（待发布）将完全转换为python3.

\n