In Prefect, when would you replace the default ThreadPoolTaskRunner with a Dask or Ray runner?

## What a task runner is A task runner is the flow-level component that executes **submitted** task runs — those created with `.submit()` or `.map()`. It is configured on the decorator: ```python from prefect import flow from prefect.task_runners import ThreadPoolTaskRunner @flow(task_runner=ThreadPoolTaskRunner(max_workers=8)) def ingest(paths: list[str]): process.map(paths) ``` It is worth separating two things that both sound like "where the work runs". The task runner governs concurrency *within* a single flow run — how its submitted task runs are executed relative to each other. Where the flow run itself is placed — which machine, container or cluster picks it up — is a deployment and infrastructure concern, configured elsewhere. A flow can be running on a Kubernetes pod and still use plain threads internally, or run on a laptop and dispatch to a Dask cluster. ## The default: threads Prefect 3's default is `ThreadPoolTaskRunner`, executing submitted runs as threads in the flow process. `max_workers` bounds how many run at once; `ThreadPoolTaskRunner(max_workers=1)` gives you effectively sequential execution when you want reproducible ordering while debugging. Threads are the right default because most orchestration work is waiting. Fifty concurrent HTTP fetches or fifty warehouse queries spend nearly all their time blocked on a socket, and threads release the interpreter lock while blocked, so concurrency is real. The cost is minimal: no serialisation, no cluster, shared memory, and tracebacks that make sense. Where threads stop paying: - **CPU-bound pure Python.** Parsing, regex-heavy cleaning, pandas transforms in Python-level loops. The interpreter lock serialises them, and eight threads finish in roughly the time one would. (Libraries that release the lock in C code — numpy, some compression and I/O paths — are an exception and do parallelise.) - **Memory-bound work.** Every task run shares one heap. Twenty runs each holding a 2 GB frame is an out-of-memory crash regardless of how many cores you have. - **Fan-out wider than one machine.** Beyond the box's cores and RAM there is nowhere else to go. ## Distributed runners `DaskTaskRunner` ships in the `prefect-dask` integration package, `RayTaskRunner` in `prefect-ray`. Both accept the same submitted task runs and dispatch them to their respective cluster, either an existing cluster you point at or a temporary one created for the flow run. ```python from prefect import flow from prefect_dask import DaskTaskRunner @flow(task_runner=DaskTaskRunner()) def heavy(partitions: list[str]): crunch.map(partitions) ``` The attraction is that the flow code does not change: the same `.map()` call fans out to a cluster. The costs are real, though, and naming them is what makes this a senior answer: - **Serialisation.** Arguments and return values must travel between processes and machines, so they must be serialisable and small enough to be worth moving. Passing a large frame per task run is a self-inflicted wound; pass a path or a key. - **Environment parity.** Cluster workers need the same code and dependencies as the flow. Version drift produces failures that look like data bugs. - **Operational surface.** A cluster is another thing to size, monitor, secure and pay for, plus a second scheduler with its own failure modes. - **Debuggability.** A traceback from a remote worker is further from you than one from a thread in your own process. ## Choosing Start with the default and change only on evidence. The decision tree in practice: 1. **Is the work waiting or computing?** Waiting → threads, and tune `max_workers`. 2. **Computing, but does the heavy lifting happen elsewhere?** If the task issues SQL to a warehouse or triggers a processing engine, the compute is not in Python at all — threads are still correct, and a cluster adds nothing. 3. **Computing in Python, and one machine is not enough?** Then a distributed runner, or — often the better answer — push the transformation into a purpose-built data-processing engine and let the orchestrator orchestrate. That third point deserves emphasis in an interview. Reaching for a distributed task runner because a pandas step is slow is frequently the wrong layer: an orchestrator distributing Python functions is not the same thing as a data-processing engine with a partitioning model, spill-to-disk and a query optimiser. Use the distributed runner when you genuinely have many independent, moderately heavy units of work — per-partition simulations, per-file model scoring — not as a substitute for the processing tier. ## Two details that trip people **Directly-called tasks ignore the runner.** `process(path)` runs inline in the flow regardless of what runner is configured. Only `.submit()` and `.map()` go through it. Someone who swaps in a cluster runner and sees no change is usually not submitting anything. **Runners are per flow, including subflows.** A subflow can use a distributed runner for one heavy stage while the parent keeps threads, which is a clean way to isolate the expensive part. ## Version note Prefect 3's default is `ThreadPoolTaskRunner`; Prefect 2 called its default `ConcurrentTaskRunner` and shipped a separate sequential runner. Say which version you mean when the name matters.