python - 触发一系列并行任务

Question

对于这个 dask 代码：

def inc(x): 
  return x + 1

for x in range(5):
  array[x] = delay(inc)(x)

我想array通过执行延迟任务来访问所有元素。但我不能打电话array.compute()，因为array它不是一个函数。如果我做

for x in range(5):
  array[x].compute()

那么每个任务是并行执行还是仅在终止a[1]后才被触发？a[0]有没有更好的方法来编写这段代码？

Answer 1

您可以使用该dask.compute函数一次计算许多延迟值

from dask import delayed, compute

array = [delayed(inc)(i) for i in range(5)]
result = compute(*array)

Answer 2

如果你强迫它们花费很长时间，很容易判断它们是否并行执行。如果您运行此代码：

from time import sleep, time
from dask import delayed

start = time()

def inc(x):
    sleep(1)
    print('[inc(%s): %s]' % (x, time() - start))
    return x + 1

array = [0] * 5
for x in range(5):
    array[x] = delayed(inc)(x)

for x in range(5):
    array[x].compute()

很明显，调用是按顺序发生的。但是，如果你用这个替换最后一个循环：

delayed(array).compute()

然后你可以看到它们是并行的。在我的机器上，输出如下所示：

[inc(1): 1.00373506546]
[inc(4): 1.00429320335]
[inc(2): 1.00471806526]
[inc(3): 1.00475406647]
[inc(0): 2.00795912743]

显然，它执行的前四个任务是并行的。大概默认并行度设置为机器上的内核数，因为对于 CPU 密集型任务，拥有更多内核通常没有用处。