Dask Executor in Prefect

Preliminaries

from prefect.executors import LocalDaskExecutor,  DaskExecutor
from prefect import task, Flow

from typing import Iterable

Create a Prefect Task

# create sample prefect task to run in a flow
@task
def add(list_of_int: Iterable[int]):
    return list_of_int[0] + list_of_int[1]

Create a Prefect Flow

# list of integers to add
list_of_ints = [[1, 2], [-3, 5], [4, 5], [9, 0], [3, -8]]

# create prefect Flow to run with Dask Executors
with Flow("sample-flow") as flow:
    result = add.map(list_of_ints)

With LocalDaskExecutor

if __name__ == "__main__":
    flow.run(executor=LocalDaskExecutor(scheduler="processes", num_workers=4))

With DaskExecutor

if __name__ == "__main__":
    flow.run(executor=DaskExecutor(cluster_kwargs={"n_workers": 4}))

Note: LocalDaskExecutor uses num_workers and DaskExecutor uses n_workers to specify the number of workers.