在 for 循环上使用 python 多重处理将结果附加到字典中

Question

因此，我查看了多处理模块的文档，也查看了此处提出的其他问题，但似乎没有一个与我的情况相似，因此我开始了一个新问题。

为了简单起见，我有一段如下形式的代码：

# simple dataframe of some users and their properties.
data = {'userId': [1, 2, 3, 4],
        'property': [12, 11, 13, 43]}
df = pd.DataFrame.from_dict(data)

# a function that generates permutations of the above users, in the form of a list of lists
# such as [[1,2,3,4], [2,1,3,4], [2,3,4,1], [2,4,1,3]]
user_perm = generate_permutations(nr_perm=4)

# a function that computes some relation between users
def comp_rel(df, permutation, user_dict):
    df1 = df.userId.isin(permutation[0])
    df2 = df.userId.isin(permutation[1])
    user_dict[permutation[0]] += permutation[1]
    return user_dict


# and finally a loop: 
user_dict = defaultdict(int)
for permutation in user_perm:
    user_dict = comp_rel(df, permutation, user_dict)    

我知道这段代码现在没有什么意义（如果有的话），但我只是写了一个小例子，它接近我正在处理的实际代码的结构。这user_dict最终应该包含userIds一些价值。

我有实际的代码，它工作正常，给出了正确的字典和所有内容，但是......它在单个线程上运行。而且速度慢得令人痛苦，记住我还有另外 15 个完全空闲的线程。

我的问题是，如何使用multiprocessingpython 模块来更改最后一个 for 循环，并能够在所有可用的线程/核心上运行？我看了文档，不是很容易理解。

编辑：我尝试将池用作：

p = multiprocessing.Pool(multiprocessing.cpu_count())
p.map(comp_rel(df, permutation, user_dict), user_perm)
p.close()
p.join()

但是这会中断，因为我正在使用该行：

user_dict = comp_rel(df, permutation, user_dict) 

在最初的代码中，我不知道池完成后应该如何合并这些字典。

Answer 1

经过评论中的简短讨论后，我决定使用以下方法发布解决方案ProcessPoolExecutor：

import concurrent.futures
from collections import defaultdict

def comp_rel(df, perm):
    ...
    return perm[0], perm[1]

user_dict = defaultdict(int)
with concurrent.futures.ProcessPoolExecutor() as executor:
    futures = {executor.submit(comp_rel, df, perm): perm for perm in user_perm}
    for future in concurrent.futures.as_completed(futures):
        try:
            k, v = future.result()
        except Exception as e:
            print(f"{futures[future]} throws {e}")
        else:
            user_dict[k] += v

它的工作方式与 @tzaman 相同，但它使您可以处理异常。该模块还有更多有趣的功能，请查看文档。