python – 用于嵌套列表理解的joblib.Parallel

>>> nested = [[1,2],[3,4,5]]
>>> [[sqrt(i) for i in j] for j in nested]
[[1.0,1.4142135623730951],[1.7320508075688772,2.0,2.23606797749979]]

>>> #this syntax fails:
>>> Parallel(n_jobs = 2) (delayed(sqrt)(i for i in j) for j in nested)
Traceback (most recent call last):
  File "<stdin>",line 1,in <module>
  File "C:Python27libsite-packagesjoblibparallel.py",line 660,in __call__
    self.retrieve()
  File "C:Python27libsite-packagesjoblibparallel.py",line 512,in retrieve
    self._output.append(job.get())
  File "C:Python27libmultiprocessingpool.py",line 558,in get
    raise self._value
pickle.PicklingError: Can't pickle <type 'generator'>: it's not found as __builtin__.generator
>>> #this syntax doesn't fail,but gives the wrong output:
>>> Parallel(n_jobs = 2) (delayed(sqrt)(i) for i in j for j in nested)
[1.7320508075688772,1.7320508075688772,2.23606797749979,2.23606797749979]

我不太确定你的第二次尝试会发生什么,但第一次对我来说很清楚：
sqrt后面的括号中的表达式(i for j in j)会生成一个 “generator”对象,该对象将传递给并行处理管道.不幸的是,发电机的输出可能取决于先前的呼叫.在你的情况下,它将在每次被调用时提供j的下一个元素,但它也可能会进行一些内部计算,这意味着不同的进程相互依赖,并且结果可能依赖于顺序其中执行并行处理.因此,多处理库拒绝继续.

正如我所说,我不太确定第二个例子中发生了什么,但可能只是你设法意外地将多处理技巧转化为它在第一种情况下完全避免的做法.

潜在解决方案

1：分离迭代级别

…例如,正如j_n建议的那样,通过定义将在低级列表上迭代的函数.这很容易实现,但可能无法从并行化中获得多少好处,具体取决于单个列表的长度.
对于外部循环使用非并行列表理解也可以是一个选项,但是并行化内部循环,甚至两者兼并 – 这是否有用取决于数据的结构.

2：迭代嵌套列表的线性化版本

这样,每次执行都是并行完成的,但这意味着您需要首先“展平”列表并稍后重新构建它.

如果您的嵌套列表是有规律的结构(即如果它包含n个列表,每个列表包含m个元素),这很容易：

从嵌套列表中创建一个numpy array,如下所示：

import numpy as np
# convert to array -- only works well if you have a regular structure!
nested_arr = np.array(nested)
# the shape of the array,for later
shape = nested_arr.shape

# generate an (n*m) linear array from an (n,m) 2D one
linear = nested_arr.ravel()

# run the parallel calculation
results_lin = Parallel(n_jobs = 2) (delayed(sqrt)(e) for e in linear)

# get everything back into shape:
results = results_lin.reshape(shape)

实际上,这可能更简单,因为np.nditer()在多维数组上以元素方式迭代.不过,我不确定它是否会与joblib和多处理协同工作.
如果你有常规数据(并且你真的只想做更复杂的事情而不是获得平方根),你还应该考虑使用np.sqrt(nested_arr) – 这比迭代数字列表要快得多按顺序分别对它们进行平方！

如果嵌套列表不规则,则线性化变得更加复杂：

# store lengths of the sub-lists
structure = [len(e) for e in nested]

# make one linear list
linlist = []
for l in nested:
    linlist.extend(l)

# finally run the parallel computation:
results_lin = Parallel(n_jobs = 2) (delayed(sqrt)(e) for e in linlist)

# ...and bring it all back into shape:
results = []
i = 0

for n in structure:
    results.append(results_lin[i:i+n])

这一切是否有意义取决于您正在处理的数据量以及列表的复杂程度.使用您的简单示例,显然排序将比计算平方根需要更长的时间.

你真的需要并行化吗？

如果您所做的只是对大量数字的简单数学运算,请考虑使用np.array.您可以将数组放入大多数方程式中,就像它们是数字一样,并且计算运行得更快：

In [14]: time resl = [sqrt(e) for e in range(1000000)]
CPU times: user 2.1 s,sys: 194 ms,total: 2.29 s
Wall time: 2.19 s

In [15]: time res = np.sqrt(np.arange(1000000))
CPU times: user 10.4 ms,sys: 0 ns,total: 10.4 ms
Wall time: 10.1 ms

这比列表上的操作可以加速到更快,甚至可以在24个内核上并行运行. (实际上,你需要大约216个并行进程才能跟上numpy,而且我确信mutliprocessing将负载分配给那么多进程的计算工作无论如何都会使尝试失败.