Hello everyone , i am using prefect and dask resource manager and running some tasks within Dask resource manager context and few outside the context manager. Flow looks like this

with Flow("Test flow") as flow:
   
    with DaskCluster(n_workers=n_workers) as client:
        data = extract()
        processed_data= transform(data)

    save_data(processed_data)

The problem is dask cluster is not shutting until save_data function completes but i expected that as soon as

transform

function completes , the

cluster cleanup

should happen, Is there any way i can initiate

save_data()

after

transform()

is done and the

cluster_cleanup

is done as well.

I have not tried this, but wouldn't the sub-flow pattern solve this?

i expected the same , but my dask cluster stays up uptil the last function also completes

Interesting, looking forward to read what the prefects are saying 🙂

What happens if you tried to get

transform(data).result()

before exiting? Also, does using the

.submit

method on each of the tasks make a difference?