[Python] How to nest dask.delayed functions within other dask.delayed functions

I am trying to learn dask, and have created the following toy example of a delayed pipeline.

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


So baz has a dependency on bar which in turn has a dependency on foo

I would like all 3 to be delayed tasks, so as each gets executed, they show up as individual tasks in the task pipeline on the dashboard.

My code is as follows:

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)


This doesn't work, as the output of f = foo(data) and b = bar(f) are both Delayed objects.

As such, the result returned from computing baz_task is

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


Obviously my desired output is:

baz[bar[foo[hello world]]]


Things I've tried:

removing the @dask.delayed decorator from foo and bar:

This results in there only being a single task in the task graph, and so one can't follow process on the dashboard.

calling the Delayed object inside the body of baz:

eg:

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


This returns a delayed apply, whatever that is...

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

As far as I can tell, my code follows the decorator example pretty closely in the docs.

What am I doing wrong?

Continue reading...