multiprocessing.pool.imap是否有一个允许多个参数的变体(如starmap)？

Question

我正在对大量字节进行一些计算.该进程在大块字节上运行.我试图使用多处理并行处理来提高性能.最初我尝试使用pool.map,但只允许单个参数,然后我找到了pool.starmap.但是pool.starmap仅在所有进程完成后才给出结果.我想要结果(有点).我正在尝试使用pool.imap,它在进程完成时提供结果,但不允许多个参数(我的函数需要2个参数).而且,结果的顺序很重要.

下面的一些示例代码:

pool = mp.Pool(processes=4)
y = []
for x in pool.starmap(f, zip(da, repeat(db))):
    y.append(x)

上面的代码有效,但只在完成所有进程后才给出结果.我看不到任何进展.这就是为什么我尝试使用pool.imap,效果很好但只有一个参数:

pool = mp.Pool(processes=4)
y = []
for x in pool.imap(f, da)):
    y.append(x)

在多个参数上引发以下异常:

TypeError: f() missing 1 required positional argument: 'd'

寻找实现所有3个要求的简单方法:

1. 使用多个参数/参数进行并行处理
2. 设法在流程运行时查看进度
3. 有序的结果.

谢谢!

Answer 1

您可以通过以下函数模拟starmap使用：imapfunctools.partial()

import functools
import multiprocessing as mp

def my_function(constant, my_list, optional_param=None):
    print(locals())

with mp.Pool() as pool:
    list(
        pool.imap(
            functools.partial(
                my_function, 2, optional_param=3
            ),
            [1,2,3,4,5]
        )
    )

输出：

$ python3 foo.py
{'optional_param': 3, 'my_list': 1, 'constant': 2}
{'optional_param': 3, 'my_list': 3, 'constant': 2}
{'optional_param': 3, 'my_list': 2, 'constant': 2}
{'optional_param': 3, 'my_list': 4, 'constant': 2}
{'optional_param': 3, 'my_list': 5, 'constant': 2}

Answer 2

我可以很快回答前两个问题。我想你在理解了前两个问题之后应该能够处理第三个问题。

1. 多个参数的并行处理

我不确定整个“星图”的等价物，但这里有一个替代方案。我过去所做的是将我的参数压缩成单个数据对象，例如列表。例如，如果您想将三个参数传递给map_function，您可以将这些参数附加到一个列表中，然后将该列表与.map()or.imap()函数一起使用。

def map_function(combo):
    a = combo[0]
    b = combo[1]
    c = combo[2]
    return a + b + c

if '__name__' == '__main__':
    combo = []
    combo[0] = arg_1
    combo[1] = arg_2
    combo[2] = arg_3

    pool = Pool(processes=4)
    pool.map(map_function, combo)

2. 跟踪进度

一个好方法是使用 的multiprocessing共享值。实际上，大约一个月前我问了这个（几乎）完全相同的问题。这允许您从函数创建的不同进程中操作相同的变量map。为了学习的目的，我将让您自己阅读并找出共享状态解决方案。如果您在尝试几次后仍然遇到问题，我将非常乐意帮助您，但我相信自学如何理解某些事物比我给您答案更有价值。

希望这可以帮助！！