Short answer: either is fine. In Prefect 3.x the engine calls

duplicate()

on the flow’s task runner at run time, so each flow run gets its own fresh DaskTaskRunner instance even if you “share” a module-level runner. Sharing a module-level runner is just sharing configuration; no runtime state is shared across flows. Guidance: - If you want to avoid repeating config (e.g., the same

address

), a single module-level DaskTaskRunner instance is perfectly safe to reference in multiple

@flow

decorators. - If you prefer clarity/encapsulation, instantiate the runner per flow. Behavior is identical at runtime. - Concurrency: multiple flows can run concurrently against the same external Dask cluster; tasks will interleave on the cluster. If you need isolation, use separate clusters or queues. Examples - Shared module-level instance (safe; shares config only):

from prefect import flow
from prefect_dask.task_runners import DaskTaskRunner

RUNNER = DaskTaskRunner(address="<tcp://dask-scheduler:8786>")

@flow(task_runner=RUNNER)
def flow_a():
    ...

@flow(task_runner=RUNNER)
def flow_b():
    ...

- Instantiate per flow (equivalent behavior at runtime):

from prefect import flow
from prefect_dask.task_runners import DaskTaskRunner

@flow(task_runner=DaskTaskRunner(address="<tcp://dask-scheduler:8786>"))
def flow_a():
    ...

@flow(task_runner=DaskTaskRunner(address="<tcp://dask-scheduler:8786>"))
def flow_b():
    ...

Why this works: on each flow run, Prefect creates a run-scoped copy via

task_runner.duplicate()

, so clients, exit stacks, and any runtime state are isolated per run. References - Task runners (Prefect 3.x) - DaskTaskRunner source (duplicate and lifecycle) - Flow engine (uses task_runner.duplicate() per run)