Great question — assuming Prefect 3.x. Short answer - Flow retries rerun the entire flow body. Each retry attempt is a fresh run. - With

DaskTaskRunner()

that creates a temporary cluster, the cluster is torn down at the end of a failed attempt and a new one is created on the next attempt after

retry_delay_seconds

. - If you connect to an existing Dask scheduler via

address=...

, Prefect will not tear it down; retries will reconnect to the same cluster. -

persist_result

does not affect Dask infrastructure lifecycle. It’s not required to “optimize” Dask usage. For avoiding recomputation across retries, tune caching instead. Details - Cluster lifecycle - Ephemeral cluster (default):

DaskTaskRunner()

creates a temporary cluster for the duration of a flow run attempt. On failure, the attempt ends and the cluster is closed. The next retry creates a new cluster. - Persistent/external cluster:

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

connects to an existing cluster. Prefect won’t destroy that cluster between retries; it will just reconnect. - Subflows - Each flow (including subflows) has its own task runner by default. A subflow decorated with

@flow(task_runner=DaskTaskRunner(...))

will manage its own cluster as described above. - If you want a subflow to use the same Dask cluster as the parent, either: - Point both flows at the same external scheduler via

address=...

, or - Pass the same runner object to the subflow via

with_options(task_runner=parent_runner)

. - Retries vs caching vs results - Flow retries: reruns the entire flow/subflow after

retry_delay_seconds

. In-memory state is lost between attempts. - Caching: By default, task cache keys include

RUN_ID

, so you won’t hit cache across flow retries. If you want to reuse cached outputs across retries, exclude

RUN_ID

from the cache policy. - Result persistence (

persist_result

): Controls storing task/flow return values. It does not control Dask cluster reuse and is separate from the task cache. In-memory caching also doesn’t help across retries. Examples 1) Reuse an external Dask cluster across retries

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

@task
def work(x): ...

@flow(task_runner=DaskTaskRunner(address="<tcp://scheduler:8786>"), retries=3, retry_delay_seconds=60)
def my_flow():
    return work(1)

2) Share the same runner between parent and subflow

from prefect import flow
from prefect_dask.task_runners import DaskTaskRunner

runner = DaskTaskRunner(address="<tcp://scheduler:8786>")

@flow(task_runner=runner)
def subflow():
    ...

@flow(task_runner=runner)
def parent():
    subflow.with_options(task_runner=runner)()

3) Reuse cached task results across retries by excluding RUN_ID

from prefect import task
from prefect.cache_policies import INPUTS, TASK_SOURCE

@task(cache_policy=INPUTS + TASK_SOURCE)  # exclude RUN_ID to allow reuse across retries
def heavy_step(params):
    ...

Docs - DaskTaskRunner integration: Prefect + Dask integration - Retries: Retries guide - Caching concepts and policies: Caching concepts and Caching how-to A couple of clarifying questions to tailor guidance: - Are you creating ephemeral Dask clusters or connecting to an existing scheduler? - Do you want the subflow to share the same cluster as the parent, or should it be isolated?