Dynamic Workflows: Just Write Python

Here’s the moment Prefect earns its place in your toolbox. You have a task that returns today’s order files — one per region, and you don’t know how many until the extract runs. In Airflow, that unknown count is a design problem: the graph is fixed at parse time, so a task-per-file needs dynamic task mapping — .expand(), a mappable operator, and the discipline to keep the expanded input serializable and small. In Prefect the same requirement is a for loop, or one call to .map(), and there was never a graph to negotiate with in the first place.

There’s one trap in that sentence, though, and it catches almost everyone coming from a language runtime rather than a scheduler. So before the fan-out patterns, the gotcha.

The `for` loop that doesn’t parallelize

Write the obvious thing and it will run, produce correct output, and quietly do it one file at a time:

@flow(log_prints=True)
def daily_load():
    files = list_order_files()
    counts = []
    for path in files:
        counts.append(load_file(path))   # BLOCKS here, every iteration
    print(f"loaded {sum(counts)} rows")

Calling load_file(path) directly — the way you’d call any Python function — runs the task to completion and returns its value before the loop advances. Three files that each take twenty seconds take a minute, in order. Nothing is concurrent. This surprises Airflow engineers because in Airflow a loop that builds tasks is the parallel construct: the loop runs at parse time, emits N task instances, and the scheduler runs them across your workers. Here the loop runs at execution time and each call is a blocking function call. The loop is control flow, not a fan-out.

Concurrency is opt-in, and the opt-in is one of two method calls on the task. .submit(path) schedules the task on the flow’s task runner and hands back a future immediately, without waiting:

@flow(log_prints=True)
def daily_load():
    files = list_order_files()
    futures = [load_file.submit(path) for path in files]   # all scheduled, none awaited
    counts = [f.result() for f in futures]                  # now collect
    print(f"loaded {sum(counts)} rows")

The list comprehension fires off every task run without blocking; the three loads overlap on the thread pool; only the second comprehension, where you call .result(), waits. Or skip the manual loop entirely and let .map() do the fan-out for you — which is where most dynamic Prefect code actually lives.

Fan-out is a method call

The bookshop extract lands a variable number of regional order files each morning. You want to load each one, then consolidate. The whole thing:

from prefect import flow, task

@task
def list_order_files() -> list[str]:
    # today's per-region drops; count varies day to day
    return ["orders-emea.csv", "orders-amer.csv", "orders-apac.csv"]

@task
def load_file(path: str) -> int:
    rows = read_csv(path)
    write_to_warehouse(rows)
    return len(rows)

@task
def consolidate(counts: list[int]) -> int:
    return sum(counts)

@flow(log_prints=True)
def daily_load():
    files = list_order_files()
    counts = load_file.map(files)
    total = consolidate(counts)
    print(f"loaded {total} rows")

load_file.map(files) submits one task run per element (the fan-out) and returns a PrefectFutureList — a list of futures standing in for the results. Hand that list straight to consolidate and Prefect resolves every future before the reduce runs; you didn’t await anything by hand. The count of load_file runs is decided the instant list_order_files returns. Nothing was declared up front, because there’s no up front — the shape is the execution.

Compare the Airflow version in your head: a mapped task, an XCom carrying the file list, the parse-time constraint that the thing you map over stay modest, and a UI that renders the mapped instances as a group. The Prefect version is the same idea with the scaffolding removed. If you want the map without a separate reduce task, a comprehension over the futures works too:

counts = load_file.map(files)
total = sum(f.result() for f in counts)

One thing to internalize: .map() runs concurrently and .submit()-in-a-loop runs concurrently, but a bare for path in files: load_file(path) runs sequentially. Same three lines of intent, three very different execution profiles. When you want fan-out, reach for .map() or .submit() deliberately.

`.map()` past the happy path

The one-iterable case is the demo. Real fan-out mixes per-element inputs with constants, zips several lists together, and occasionally maps a map. Prefect handles all of it, but the rules are specific.

Constants need unmapped(). By default .map() treats every argument as an iterable to spread over, one element per task run. Pass a plain value you want shared across all runs and Prefect will try to iterate it — a string gets mapped character by character, a dict by its keys, and a bare object raises. Wrap anything that should be held constant:

from prefect import unmapped

@task
def load_file(path: str, target_schema: str, run_date: str) -> int:
    ...

@flow
def daily_load(run_date: str):
    files = list_order_files()
    counts = load_file.map(
        files,
        target_schema=unmapped("analytics"),   # same for every run
        run_date=unmapped(run_date),            # same for every run
    )

Without unmapped, "analytics" would be spread letter by letter and you’d get one task run per character — usually a crash when the lengths don’t line up, occasionally a silent nonsense fan-out. unmapped is also how you pass a list you mean as a single value: unmapped(["a", "b"]) gives every run the whole two-element list, where the bare list would map over it.

Multiple iterables zip. Give .map() two lists of equal length and it pairs them positionally — element i of each goes to task run i:

counts = load_file.map(files, schemas)   # (files[0], schemas[0]), (files[1], schemas[1]), ...

This is zip, not a cross product. And unlike Python’s zip, mismatched lengths do not silently truncate to the shortest — Prefect refuses outright with MappingLengthMismatch: Received iterable parameters with different lengths. That’s the right call, and it means you don’t have to defensively assert the lengths match: if the pairing is wrong, you find out immediately and loudly rather than by noticing that three of your five regions never loaded. Keyword arguments map the same way; you can mix mapped and unmapped freely:

counts = load_file.map(
    path=files,                       # mapped
    target_schema=schemas,            # mapped, zipped with files
    run_date=unmapped(run_date),      # constant
)

Nested maps compose because .map() returns futures and .map() accepts futures. To fan a second stage out over the first stage’s results, map the downstream task over the upstream future list — no intermediate collection, no .result() call to force a barrier:

raw = load_file.map(files)            # PrefectFutureList
validated = validate.map(raw)         # one validate run per load, chained through futures
report = summarize(validated)

Prefect wires each validate run to wait on its matching load_file future, so the two stages pipeline: a file that finishes loading early can start validating while a slower file is still loading. You didn’t build a dependency graph; you passed futures to a function.

Ordering dependencies with no data link use wait_for=, the same knob you’d use on a normal task call. When a mapped stage must run after some upstream that it doesn’t take as input — a schema migration, a truncate, a permission grant — pass the upstream future(s):

prep = ensure_target_tables()
counts = load_file.map(files, wait_for=[prep])   # every load waits on prep, no data passed

Finally, a cost worth knowing before you map over something huge: to fan out, Prefect has to introspect the mapped inputs — walk each iterable, resolve any futures inside it, and pair everything up. For a few hundred elements that’s nothing. For a map over tens of thousands of large dicts it’s real overhead, both in the introspection and in the per-run bookkeeping. The fix is the same as Airflow’s advice, for the same reason: map over small identifiers — file paths, IDs, partition keys — and let each task run fetch its own heavy payload, rather than mapping over the payloads themselves.

The Futures API

.submit() and .map() both trade in PrefectFuture objects, and knowing the handful of methods on them is what lets you move past “map then reduce” into arbitrary orchestration.

A future is a handle to a task run that may not have finished. The methods you’ll reach for:

.result() blocks until the run finishes and returns its value — or re-raises the task’s exception, so a failed upstream surfaces where you call .result(), not silently.
.wait() blocks until the run finishes but does not raise or return the value; it just resolves the future. Use it when you care that the work is done, not what it produced.
.state gives you the run’s State object without blocking on a not-yet-finished run in the way .result() does — useful for inspecting whether a run is Completed or Failed and branching on it.

future = load_file.submit("orders-emea.csv")
# ... do unrelated work while it runs ...
if future.state.is_completed():
    rows = future.result()

When you’ve submitted a batch, you rarely want to .result() them in list order, because that makes you wait on the slowest run before touching the ones that already finished. Prefect gives you two collective helpers from prefect.futures:

from prefect.futures import wait, as_completed

futures = [load_file.submit(p) for p in files]

# Barrier: block until every run is done, whatever its state.
done = wait(futures)
print(f"{len(done.done)} done, {len(done.not_done)} still running")

# Stream: handle each result the moment it lands, fastest first.
for future in as_completed(futures):
    print("finished:", future.result())

wait(futures) is a barrier that returns after all (or, with a timeout=, as many as finished in time) have resolved, sorting them into done and not_done sets. as_completed(futures) is an iterator that yields each future the instant it completes, so you process results in completion order rather than submission order — the right tool when downstream work per result is expensive and you don’t want the tail of the batch to hold up the head. Both are the Prefect-3.x equivalents of the concurrent.futures functions of the same name, and they behave the way that experience trained you to expect.

Passing a whole list of futures where a task wants a list of values also just works — Prefect resolves futures inside arguments before the task body runs, which is exactly why consolidate(counts) at the top of this post never had to call .result().

Choosing a task runner

.map() and .submit() describe what runs concurrently; the task runner decides how. It’s a per-flow argument, which is the sharpest contrast with Airflow: there, the executor is a cluster-wide deployment decision made by whoever runs the scheduler; here, two flows in the same file can use two different runners, and you change one by editing a keyword.

The default is ThreadPoolTaskRunner — submit tasks and they run across a thread pool with no configuration from you. Set the pool size where the default isn’t enough:

from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def daily_load():
    ...

Threads are the right default because they cover I/O-bound work — reading files, calling a warehouse, hitting an API — which is most of what a load pipeline does. While one thread waits on the network, another runs. But threads share one Python process and one GIL, so genuinely CPU-bound work (parsing, hashing, compression, in-Python number crunching) doesn’t actually parallelize on a thread pool — the threads take turns holding the GIL. For that, Prefect 3.x adds ProcessPoolTaskRunner, which runs each task in a separate process with its own interpreter:

from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def crunch_reports():
    reports = build_report.map(partitions)   # each runs in its own process, real parallelism

The tradeoff is the one you’d expect from multiprocessing: task inputs and return values have to be picklable because they cross a process boundary, and there’s spawn overhead per run, so it pays off for coarse CPU-heavy tasks and loses to threads for short I/O-bound ones. Reach for it specifically when profiling shows the GIL is your ceiling.

When one machine isn’t enough — CPU work that has to spread across a cluster, or fan-out too wide for a single host — the runner swaps for a distributed one and the flow code doesn’t change. DaskTaskRunner and RayTaskRunner ship as separate integration packages (uv pip install prefect-dask / prefect-ray) and take the cluster wiring as constructor kwargs:

from prefect_dask import DaskTaskRunner

@flow(task_runner=DaskTaskRunner(
    cluster_kwargs={"n_workers": 4, "threads_per_worker": 2},
))
def daily_load():
    counts = load_file.map(files)   # identical call, now scheduled onto a Dask cluster

Point it at an existing cluster with address="tcp://..." instead of having it spin up an ephemeral one. RayTaskRunner is the same shape against Ray. The point worth holding onto: .map(), .submit(), and every future method above are runner-agnostic. You develop against threads locally and move to Dask for a heavy run by changing one decorator argument — no rewrite of the orchestration.

One alternative that isn’t a task runner at all: if your tasks are async def, a flow can await them and use asyncio.gather for concurrency directly, no pool involved. That’s the natural fit when your whole stack is already async (an httpx.AsyncClient, an async database driver). For synchronous code — most data work — the task runners above are the path, and ThreadPoolTaskRunner is the one you’ll use most.

When one mapped run fails

A fan-out’s default behavior is strict, and that’s usually correct: if one load_file in the map fails, the futures it feeds into propagate that failure, and consolidate — which depends on all of them — is marked Failed without running. One bad regional file sinks the whole reduce.

Sometimes that’s exactly wrong. If a single corrupt file shouldn’t cost you the other nine loads, wrap the upstream futures in allow_failure so the downstream task receives the failures as values instead of having them short-circuit its run:

from prefect import allow_failure

@task
def consolidate(counts: list) -> int:
    # counts may contain State objects for the runs that failed
    good = [c for c in counts if isinstance(c, int)]
    return sum(good)

@flow
def daily_load():
    files = list_order_files()
    counts = load_file.map(files)
    total = consolidate(allow_failure(counts))   # runs even if some loads failed

allow_failure tells Prefect: don’t block consolidate on the success of these upstreams; let it start, and pass through the failed runs’ states so the reduce can decide what to do. The failed load_file runs are still Failed in the UI — you don’t lose the signal — but they no longer veto the aggregation. Inside consolidate you filter out the non-values and sum what survived, which is the “process the files that loaded, alert on the ones that didn’t” pattern that partial-failure fan-outs actually want.

Keeping a wide fan-out from becoming an attack

.map() over a list makes it trivially easy to launch a thousand task runs, and a thousand task runs hitting one Postgres or one rate-limited API at once is a self-inflicted outage. The task runner’s max_workers caps local concurrency, but the ceiling that matters is often the shared external resource, and that’s not the runner’s job.

Prefect’s answer is a concurrency limit keyed to the resource, not the flow — so every task touching that warehouse shares a ceiling even across unrelated flows. In brief, you guard the sensitive section with the concurrency context manager:

from prefect.concurrency.sync import concurrency

@task
def load_file(path: str) -> int:
    rows = read_csv(path)
    with concurrency("warehouse-writes", occupy=1):   # at most N of these run at once, globally
        write_to_warehouse(rows)
    return len(rows)

Now load_file.map(thousand_files) can submit a thousand runs, but only N are inside the warehouse-write section at any moment; the rest wait their turn. This is the load-bearing companion to easy fan-out, and it has its own chapter — global limits, tag-based limits, deployment limits, collision strategies, and rate_limit for per-second ceilings — in concurrency and rate limiting. The one-line version: dynamic fan-out and concurrency limits are two halves of the same feature. Ship them together.

Fanning out over subflows and other infrastructure

Everything above maps a task. You can just as well fan out over subflows when each unit of work is itself a multi-step pipeline — call a subflow inside a loop or a comprehension and each becomes a nested flow run with its own tasks and state:

@flow
def load_region(region: str) -> int:
    files = list_files_for(region)
    return sum(load_file.map(files).result())

@flow
def daily_load():
    regions = ["emea", "amer", "apac"]
    results = [load_region(r) for r in regions]   # subflows; see the note on concurrency below

Subflows called this way run in the parent’s process, and a plain loop over them is sequential for the same reason the task loop was — to overlap subflow runs you generally push each onto its own infrastructure. That’s what run_deployment is for: it triggers a registered deployment as its own flow run, on whatever work pool that deployment targets, and hands you back the run to await. Fan out across it and each shard runs on separate infrastructure — the cross-infra version of .map():

from prefect.deployments import run_deployment

@flow
def orchestrate_regions():
    regions = ["emea", "amer", "apac"]
    runs = [
        run_deployment(
            name="load-region/prod",
            parameters={"region": r},
            timeout=0,          # return immediately, don't block on completion
        )
        for r in regions
    ]
    # each triggered run executes on its deployment's work pool

timeout=0 makes each call return as soon as the run is scheduled rather than blocking until it finishes, so the three regional loads launch in parallel across your infrastructure instead of one machine. This is how a Prefect fan-out scales past a single host without a distributed task runner: the parent flow is a thin dispatcher, and the heavy work runs as independent deployment runs. For very large fan-outs — thousands of shards — this pattern plus a deployment-level concurrency limit is far kinder to your control plane than a single flow trying to hold thousands of futures in one process.

Branching is an `if`

Airflow gives dynamic routing its own machinery — BranchPythonOperator, and the skipped-downstream bookkeeping that follows from a graph where every possible path had to exist at parse time. Prefect has no equivalent because it needs none. A branch is an if:

@flow
def daily_load():
    files = list_order_files()
    if not files:
        print("no drops today, nothing to load")
        return
    counts = load_file.map(files)
    if sum(f.result() for f in counts) == 0:
        alert_empty_load()

No path exists unless the code runs it. There’s no skipped state to reason about, no trigger rule to pick so the reduce still fires when a branch was pruned — the untaken branch simply never happened. The same goes for loops with early exits, try/except around a flaky source, and building the work list from a comprehension. It’s all just Python, because the flow is just Python.

The honest tradeoff

This flexibility has a real cost, and pretending otherwise would be the marketing version. In Airflow, the static graph buys you a complete picture of the workflow before it runs — every task, every dependency, visible in the DAG view, lintable in CI, diffable in review. You can look at a DAG and know its shape without executing it.

A Prefect flow gives that up. Because the graph is discovered at runtime, there’s no full task inventory until the flow has actually run — the UI shows you the real run after the fact, richly, but it can’t show you a run that hasn’t happened. For a pipeline whose shape depends on the data, that’s the correct trade: the static picture was always a fiction you maintained by hand anyway. For a fixed, well-known batch graph that never changes, Airflow’s parse-time certainty is a genuine feature you’re walking away from. Know which kind of pipeline you have before you decide the loop is strictly better.

Final thoughts

The reason dynamic work is easy here isn’t a cleverer mapping API — it’s that Prefect never built the wall that made mapping necessary. Airflow’s dynamic features are all workarounds for a graph fixed too early; .expand(), branch operators, and trigger rules are the tax on a parse step that had to commit to a shape before it saw the data. Remove the parse step and the workarounds have nothing to work around.

What’s left is Python you already know how to write, with exactly one thing to remember that isn’t obvious: the loop doesn’t parallelize on its own. Fan-out is .map() or .submit(); the results come back as futures you .result(), wait(), or stream with as_completed; the runner underneath is a one-line choice between threads, processes, and a cluster; and a wide fan-out wants a concurrency limit the same way it wants the fan-out. Get those pieces right and you have real concurrency that reads as ordinary code — which is what you wanted the dynamic features to give you in the first place, minus the negotiation.

Next: Deployments: From .serve() to Production

The for loop that doesn’t parallelize