Hi folks question about the `ResourceManager` and using `set Prefect Community #ask-community

Hi folks, question about the `ResourceManager` an...

Marwan Sarieddine

01/13/2022, 7:44 PM

Hi folks, question about the

ResourceManager

and using

set_upstream

How can I ensure that a task is run after the resource manager cleanup ? (Please see the thread for more details)

Marwan Sarieddine

01/13/2022, 7:45 PM

this is the canonical example provided in the docs

Copy code

with Flow("dask-example") as flow:
    n_workers = Parameter("n_workers", default=None)
    out_path = Parameter("out_path", default="summary.csv")

    with DaskCluster(n_workers=n_workers) as client:
        # These tasks rely on a dask cluster to run, so we create them inside
        # the `DaskCluster` resource manager
        df = load_data()
        summary = summarize(df, client)

    # This task doesn't rely on the dask cluster to run, so it doesn't need to
    # be under the `DaskCluster` context
    write_csv(summary, out_path)

Marwan Sarieddine

01/13/2022, 7:45 PM

I would like

write_csv

to only run after

summarize

is run and more importantly after the

DaskCluster.cleanup

is run ….

Marwan Sarieddine

01/13/2022, 7:47 PM

using set_upstream against summary is not sufficent:

Copy code

with Flow("dask-example") as flow:
    n_workers = Parameter("n_workers", default=None)
    out_path = Parameter("out_path", default="summary.csv")

    with DaskCluster(n_workers=n_workers) as client:
        # These tasks rely on a dask cluster to run, so we create them inside
        # the `DaskCluster` resource manager
        df = load_data()
        summary = summarize(df, client)

    # This task doesn't rely on the dask cluster to run, so it doesn't need to
    # be under the `DaskCluster` context
    written = write_csv(summary, out_path)
    written.set_upstream(summary)

Given this won’t set

cleanup_task

as an upstream to

written

…

Marwan Sarieddine

01/13/2022, 7:52 PM

looks to me that one solution is:

Copy code

with Flow("dask-example") as flow:
    n_workers = Parameter("n_workers", default=None)
    out_path = Parameter("out_path", default="summary.csv")
    
    client = DaskCluster(n_workers=n_workers)
    with client:
        # These tasks rely on a dask cluster to run, so we create them inside
        # the `DaskCluster` resource manager
        df = load_data()
        summary = summarize(df, client)

    # This task doesn't rely on the dask cluster to run, so it doesn't need to
    # be under the `DaskCluster` context
    written = write_csv(summary, out_path)
    written.set_upstream([summary, client.cleanup_task])

is this the recommended approach ?

Kevin Kho

01/13/2022, 8:06 PM

Did that work for you? I think that looks good to me

Marwan Sarieddine

01/17/2022, 1:52 PM

Apologies for the very late reply - yes it worked, thank you

10 Views

Open in Slack

Previous Next