@Marvin show a flow sample with dask runner.

Here's an example flow that demonstrates using a Dask Runner with Prefect:

python
from prefect import flow, task
from prefect.task_runners import SequentialTaskRunner
from prefect_dask.task_runners import DaskTaskRunner

@task
def hello_local():
    print("Hello!")

@task
def hello_dask():
    print("Hello from Dask!")

@flow(task_runner=SequentialTaskRunner())
def sequential_flow():
    hello_local.submit()
    dask_subflow()
    hello_local.submit()

@flow(task_runner=DaskTaskRunner())
def dask_subflow():
    hello_dask.submit()

if __name__ == "__main__":
    sequential_flow()

In this example,

sequential_flow

uses the

SequentialTaskRunner

to run tasks one at a time, while the

dask_subflow

uses the

DaskTaskRunner

to potentially execute tasks in parallel on a Dask cluster. Please note that if you have an existing Dask cluster, you can configure the

DaskTaskRunner

with the address of your cluster. Otherwise, a temporary Dask cluster will be created for the duration of the flow run. Remember to use the

if __name__ == "__main__":

guard to avoid issues when running tasks in parallel. For more details, visit the Prefect documentation on task runners.