Как вложить функции dask.delayed в другие функции dask.delayed ⇐ Python

[Anonymous]

Я пытаюсь изучить dask и создал следующий игрушечный пример конвейера с задержкой.

Код: Выделить всё

+-----+  +-----+  +-----+
| baz +--+ bar +--+ foo |
+-----+  +-----+  +-----+

Итак, baz зависит от bar, который, в свою очередь, зависит от foo
I хотелось бы, чтобы все три задачи были отложены, чтобы каждая из них отображалась как отдельные задачи в конвейере задач на информационной панели.
Мой код выглядит следующим образом:

Код: Выделить всё

import dask
import distributed

@dask.delayed
def foo(data:str) -> str:
return f'foo[{data}]'

@dask.delayed
def bar(data:str) -> str:
return f'bar[{data}]'

@dask.delayed
def baz(data:str) -> str:
f = foo(data)
b = bar(f)
return f'baz[{b}]'

baz_task = baz('hello world')

client = distributed.Client()
future = client.compute(baz_task)

result = future.result()
print(result)

Это не работает, поскольку выходные данные f = foo(data) и b = bar(f) являются объектами Delayed .
Таким образом, результат, возвращаемый при вычислении baz_task, равен

Код: Выделить всё

baz[Delayed('bar-4c8a0ec0-1d87-43e7-a8af-fee33a9fae3d')]

Очевидно, мой желаемый результат:

Код: Выделить всё

baz[bar[foo[hello world]]]

Что я пробовал:
удаление декоратора @dask.delayed из foo и bar :
В результате на графике задач появляется только одна задача, и поэтому невозможно следить за процессом на информационной панели.
вызов объекта Delayed внутри тела baz:
например:

Код: Выделить всё

    f = foo(data)()
^^---- # calling the Delayed object

Это возвращает отложенное применение, что бы это ни было...

Код: Выделить всё

Delayed('apply-145901ce-1271-4387-8f12-4640c9fcd23a')

Насколько я могу судить, мой код довольно точно соответствует примеру декоратора, приведенному в документации.
Что я делаю не так? ?

Подробнее здесь: https://stackoverflow.com/questions/792 ... -functions