为什么 pool.map() 和 map() 返回不同的结果？

Question

我有以下程序：

import string
import itertools
import multiprocessing as mp

def test(word_list):
    return list(map(lambda xy: (xy[0], len(list(xy[1]))),
        itertools.groupby(sorted(word_list))))

def f(x):
    return (x[0], len(list(x[1])))

def test_parallel(word_list):
    w = mp.cpu_count()
    pool = mp.Pool(w)
    return (pool.map(f, itertools.groupby(sorted(word_list))))

def main():
    test_list = ["test", "test", "test", "this", "this", "that"]

    print(test(test_list))
    print(test_parallel(test_list))

    return

if __name__ == "__main__":
    main()

输出是：

[('test', 3), ('that', 1), ('this', 2)]
[('test', 0), ('that', 0), ('this', 1)]

第一行是预期的正确结果。我的问题是，为什么 pool.map() 不返回与 map() 相同的结果？

另外，我知道 6 项列表并不是多处理的完美案例。这只是我在大型应用程序中实现时遇到的问题的一个演示。

我正在使用 Python 3.5.1。

Answer 1

来自https://docs.python.org/3.5/library/itertools.html#itertools.groupby：

返回的组本身就是一个迭代器，它共享底层 可使用 groupby() 进行迭代。因为源是共享的，当 groupby() 对象是高级的，以前的组不再可见。 因此，如果以后需要该数据，则应将其存储为列表：

groups = []
uniquekeys = []
data = sorted(data, key=keyfunc)
for k, g in groupby(data, keyfunc):
    groups.append(list(g))      # Store group iterator as a list
    uniquekeys.append(k)

我认为这里的问题是 Pool.map 试图截断其输入，并且在这样做时，它会遍历 groupby 的结果，这实际上跳过了除最后一组之外的所有元素。

您的代码的一个解决方法是使用 [(k, list(v)) for k, v in itertools.groupby(sorted(word_list))] 之类的东西，但我不知道这对您的实际用例有多适用。

Answer 2

groupby() 每组返回迭代器，这些不独立于传入的底层迭代器。您不能独立地并行迭代这些组；在您访问下一个组时，任何前面的组都将提前结束。

pool.map() 将尝试读取所有groupby() 迭代器结果，以将这些结果发送到单独的函数；仅仅试图获得第二组将导致第一组为空。

您可以在没有pool.map() 的情况下看到相同的结果，只需从groupby() 迭代到下一个结果：

>>> from itertools import groupby
>>> word_list = ["test", "test", "test", "this", "this", "that"]
>>> iterator = groupby(sorted(word_list))
>>> first = next(iterator)
>>> next(first[1])
'test'
>>> second = next(iterator)
>>> list(first[1])
[]

第一组的其余部分是“空的”，因为已请求第二组。

这显然是documented：

因为源是共享的，所以当groupby()对象前进时，之前的组就不再可见了。

您必须在将每个组发送到函数之前“实现”每个组：

return pool.map(lambda kg: f((k[0], list(kg[1]))), itertools.groupby(sorted(word_list)))

或

return pool.map(f, (
    (key, list(group)) for key, group in itertools.groupby(sorted(word_list))))

生成器表达式在 pool.map() 迭代时负责具体化。