为何Python多进程Pool的map与普通map行为不一致？

问题场景 #

尝试并行化字母列表的分组操作，编写了逻辑看似一致的串行和多进程函数，但执行结果不同。

代码实现 #

from itertools import groupby, count
import random
from multiprocess import Pool

random.seed(0)

input_letter = ["a", "b", "c"]

a = [random.choice(input_letter) for _ in range(10)]
print("input a", a)

def functional_map_approach(a:list) -> list:
    def reducer(x): 
        return [v for v, _ in x[1]]  
    
    return list(map(reducer, groupby(enumerate(a), key=lambda x: x[1])))

def functional_multiprocess_approach(a:list) -> list:        
    def reducer(x): 
        return [v for v, _ in x[1]]
    
    with Pool(5) as p:
        res= p.map(reducer, groupby(enumerate(a), key=lambda x: x[1]))
    return res
        

assert functional_map_approach(a) == functional_multiprocess_approach(a), (functional_map_approach(a), functional_multiprocess_approach(a))

运行输出 #

input a ['b', 'b', 'a', 'b', 'c', 'b', 'b', 'b', 'b', 'b']

AssertionError: ([[0, 1], [2], [3], [4], [5, 6, 7, 8, 9]], [[], [], [], [], []])

原因分析 #

核心问题在于itertools.groupby的特性和多进程机制冲突：

• itertools.groupby返回的是迭代器对象，每个分组的第二个元素（该分组的元素集合）同样是一次性迭代器，只能被遍历一次。
• 多进程的Pool.map在分发任务前，会先完整遍历groupby的迭代器以收集所有任务项，这个过程会耗尽每个分组内部的元素迭代器。当子进程中的reducer尝试遍历这些迭代器时，已经没有元素可获取，因此返回空列表。
• 迭代器本身无法被序列化传递，多进程间传递的分组对象实际上是已经被耗尽的迭代器外壳，自然得不到数据。

正确的并行化实现方案 #

由于groupby的分组结果依赖序列的连续性，无法直接对groupby的迭代器并行处理。正确的做法是先将groupby的分组结果转换为可序列化的列表结构，再将每个分组作为任务分发到子进程：

from itertools import groupby, count
import random
from multiprocess import Pool

random.seed(0)

input_letter = ["a", "b", "c"]

a = [random.choice(input_letter) for _ in range(10)]
print("input a", a)

def functional_map_approach(a:list) -> list:
    def reducer(x): 
        return [v for v, _ in x[1]]  
    
    return list(map(reducer, groupby(enumerate(a), key=lambda x: x[1])))

def functional_multiprocess_approach(a:list) -> list:        
    def reducer(x): 
        # x已经是固化的分组列表，直接处理
        return [v for v, _ in x]
    
    # 先把groupby的分组转换为(key, list(items))的结构，确保数据可序列化且迭代器被固化
    groups = [(key, list(items)) for key, items in groupby(enumerate(a), key=lambda x: x[1])]
    with Pool(5) as p:
        res = p.map(reducer, [group[1] for group in groups])
    return res
        

assert functional_map_approach(a) == functional_multiprocess_approach(a), (functional_map_approach(a), functional_multiprocess_approach(a))
print("断言成功，串行与并行结果一致")

实现说明 #

1. 先将groupby生成的每个分组迭代器转换为列表，把动态的迭代器固化为静态的可序列化结构，确保能安全传递给子进程。
2. 调整reducer的输入逻辑，直接处理固化后的分组列表，避免迭代器耗尽的问题。

内容的提问来源于stack exchange，提问作者Nikolay Zakirov