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.

Hello Ash, thanks for reaching out. I'm not entirely sure how to enforce when the cluster cleanup occurs, but it appears you have set up the dependencies correctly to ensure

save_data()

runs after

transform()

is completed. I'll investigate a bit to see if there is a way to specify that. In the meantime, here are some resources that may be helpful. • How can I configure my flow to run with Dask?

thank you Bianca, awaiting your response, meanwhile i will go through what you shared and try experiment myself.