Multiprocessing.pool 具有多个 args 和 kwargs 的函数

我想使用 mutliprocessing.pool 方法并行计算。问题是我想在计算中使用的函数提供了两个参数和可选的 kwargs，第一个参数是数据帧，第二个参数是 str，任何 kwargs 是字典。

对于我尝试执行的所有计算，我想要使用的数据帧和字典都是相同的，只有第二个参数不断变化。因此，我希望能够使用 map 方法将其作为不同字符串的列表传递给带有 df 和 dict 的已打包函数。

from utils import *
import multiprocessing
from functools import partial



def sumifs(df, result_col, **kwargs):

    compare_cols = list(kwargs.keys())
    operators = {}
    for col in compare_cols:
        if type(kwargs[col]) == tuple:
            operators[col] = kwargs[col][0]
            kwargs[col] = list(kwargs[col][1])
        else:
            operators[col] = operator.eq
            kwargs[col] = list(kwargs[col])
    result = []
    cache = {}
    # Go through each value
    for i in range(len(kwargs[compare_cols[0]])):
        compare_values = [kwargs[col][i] for col in compare_cols]
        cache_key = ','.join([str(s) for s in compare_values])
        if (cache_key in cache):
            entry = cache[cache_key]
        else:
            df_copy = df.copy()
            for compare_col, compare_value in zip(compare_cols, compare_values):
                df_copy = df_copy.loc[operators[compare_col](df_copy[compare_col], compare_value)]
            entry = df_copy[result_col].sum()
            cache[cache_key] = entry
        result.append(entry)
    return pd.Series(result)

if __name__ == '__main__':

    ca = read_in_table('Tab1')
    total_consumer_ids = len(ca)

    base = pd.DataFrame()
    base['ID'] = range(1, total_consumer_ids + 1)


    result_col= ['A', 'B', 'C']
    keywords = {'Z': base['Consumer archetype ID']}

    max_number_processes = multiprocessing.cpu_count()
    with multiprocessing.Pool(processes=max_number_processes) as pool:
        results = pool.map(partial(sumifs, a=ca, kwargs=keywords), result_col)
    print(results)

但是，当我运行上面的代码时，出现以下错误：TypeError: sumifs() missing 1 required positional argument: 'result_col'。如何为函数提供第一个 arg 和 kwargs，同时提供第二个参数作为 str 列表，以便我可以并行计算？我在论坛上读过几个类似的问题，但似乎没有一个解决方案适用于这种情况......

谢谢您，如果有不清楚的地方，我深表歉意，我今天刚刚了解到多处理包！

让我们看一下代码的两部分。

首先是sumifs函数声明：

def sumifs(df, result_col, **kwargs):

其次，调用该函数时带上相关参数。

# Those are the params
ca = read_in_table('Tab1')
keywords = {'Z': base['Consumer archetype ID']}

# This is the function call
results = pool.map(partial(sumifs, a=ca, kwargs=keywords), tasks)

更新1：

编辑原始代码后。看起来问题是位置参数分配，尝试丢弃它。

替换行：

results = pool.map(partial(sumifs, a=ca, kwargs=keywords), result_col)

with:

results = pool.map(partial(sumifs, ca, **keywords), result_col)

示例代码：

import multiprocessing
from functools import partial

def test_func(arg1, arg2, **kwargs):
    print(arg1)
    print(arg2)
    print(kwargs)
    return arg2

if __name__ == '__main__':
    list_of_args2 = [1, 2, 3]
    just_a_dict = {'key1': 'Some value'}
    with multiprocessing.Pool(processes=3) as pool:
        results = pool.map(partial(test_func, 'This is arg1', **just_a_dict), list_of_args2)
    print(results)

将输出：

This is arg1
1
{'key1': 'Some value'}
This is arg1
2
{'key1': 'Some value'}
This is arg1
2
{'key1': 'Some value'}
['1', '2', '3']

有关如何操作的更多示例Multiprocessing.pool 具有多个 args 和 kwargs 的函数 https://stackoverflow.com/questions/59756754/multiprocessing-pool-with-a-function-that-has-multiple-args-and-kwargs/59775559?noredirect=1#comment105779457_59775559

更新2：

扩展示例（由于评论）：

然而，我想知道，以同样的方式，如果我的函数有三个 args 和 kwargs，并且我想保持 arg1、arg3 和 kwargs 恒定，我如何将 arg2 作为多处理列表传递？本质上，我将如何指示映射（partial（test_func，'这是arg1'，'这将是arg3'，** just_a_dict），arg2）的多重处理，partial中的第二个值对应于arg3而不是arg2？

The Update 1代码将更改如下：

# The function signature
def test_func(arg1, arg2, arg3, **kwargs):

# The map call
pool.map(partial(test_func, 'This is arg1', arg3='This is arg3', **just_a_dict), list_of_args2)

这可以使用 python 来完成位置和关键字分配。 请注意，kwargs被放在一边并且没有使用 a 进行分配keyword尽管事实上它位于keyword指定值。

有关参数分配差异的更多信息可以找到here https://stackoverflow.com/questions/9450656/positional-argument-v-s-keyword-argument.