python - 为什么 dask 不并行化这个工作流程？

Question

我有两个非常简单的函数:

import time

def sleepy(a=1):
    time.sleep(a)
    print(a)

def ending(*args):
    print(args)
    print('finished')

我还有一个使用这些函数的 dask 工作流程:

workflow = {'task_0': (sleepy, 1), 
            'task_1': (sleepy, 2), 
            'task_2': (sleepy, 3), 
            'ending': (ending, 'task_0', 'task_1', 'task_2')}

这个工作流程可以这样可视化:

sleepy、sleepy、sleepy 应该并行运行，但事实并非如此。

我等待 1 秒，它从 sleepy() 打印 1，然后我等待 2 秒，它打印 2，然后我再等待 3 秒，它打印 3:

1
2
3
(None, None, None)
finished

我做错了什么？

Answer 1

这就是我对您的工作流程进行编码的方式，并且 sleep 操作确实是并行发生的

import dask.delayed
import time

@dask.delayed
def sleepy(a=1):
    time.sleep(a)
    print(a)

@dask.delayed
def ending(*args):
    print(args)
    print('finished')

d = ending(*[sleepy(i) for i in [1, 2, 3]])
d.compute()

请注意，@装饰器只是语法上的好处，你也可以做dask.delayed(sleepy)等。