Dynamic Task Mapping and TaskGroups

Here is a problem you hit the moment a pipeline meets the real world. The bookshop doesn’t hand you one tidy orders file every morning. It hands you however many the overnight export produced — one per storefront, and the storefront count drifts as branches open and close. Some mornings it’s four files. Some mornings it’s nine. You want a load task for each one, running in parallel, each with its own logs and its own retry.

Your first instinct is a Python loop:

for path in list_todays_files():   # this runs at PARSE time, not run time
    load(path)

That instinct is wrong, and it’s worth understanding exactly why, because the reason shapes everything in this chapter. Airflow parses your DAG file to discover its shape — the tasks and the edges between them — before any run exists. At parse time there is no “today.” list_todays_files() would run against whatever the scheduler happens to see when it re-reads the file, not against the run you care about, and it would bake a fixed number of tasks into the graph until the next parse. A loop can only build a structure that’s known in advance. You need a structure whose width is decided when the run actually happens.

That is dynamic task mapping. One task returns a list at run time; another task expands into one mapped instance per element. Same mechanism your old loop wanted, except Airflow does the fanning-out during the run, when the data is real.

`expand`: one task, many instances

The core move is .expand(). You take a task and, instead of calling it once, you tell it to run once per item in a list:

from airflow.sdk import dag, task

@task
def list_order_files() -> list[str]:
    # discovered at RUN time — could be 4 paths, could be 9
    return find_csvs("/data/incoming/orders/")

@task
def load_file(path: str) -> int:
    rows = copy_csv_to_warehouse(path)
    return rows          # a small number — fine to return

@dag(schedule="@daily", catchup=False)
def load_orders():
    files = list_order_files()
    load_file.expand(path=files)

load_orders()

Read the last line as a sentence: for each path in files, run one load_file. If the run finds four files, you get four load_file instances; nine files, nine instances. They run in parallel, up to your concurrency limits, and — this is the part that pays off at 3 a.m. — in the UI they collapse into a single node you can expand to see every instance separately. Each mapped instance has its own log and its own retry state. One storefront ships a corrupt CSV and its instance fails on its own; the other eight load and commit. You retry the one that broke, not the whole batch.

Notice that files is the return value of list_order_files(), not a literal list. That’s the whole point — the fan-out width came from an upstream task’s output, so it was unknown until the run produced it. You can also expand over a literal list when the items are genuinely static (.expand(path=["a.csv", "b.csv"])), but the runtime-list case is why the feature exists.

`partial`: binding the arguments that don’t vary

expand maps every argument you give it — each one has to be a list, and each mapped instance pulls one element. But most real tasks have a mix: one argument varies per item, and the rest are constants that should be identical on every instance. A load task might take a per-file path plus a fixed target_table and a fixed batch_id.

If you tried to .expand(path=files, target_table="orders") Airflow would reject it, because expand wants a list for every argument and "orders" is a string, not a list. You don’t want to map over the table — you want the same table on all of them. That’s what .partial() is for. It binds the constant arguments up front; then .expand() maps only what’s left:

@task
def load_file(path: str, target_table: str, batch_id: str) -> int:
    return copy_csv_to_warehouse(path, target_table, batch_id)

load_file.partial(
    target_table="raw.orders",
    batch_id="{{ ds }}",
).expand(path=files)

target_table and batch_id are pinned to the same value on every instance; path varies. The rule to keep in your head: you cannot map a task without partial unless it has no constant arguments at all. The moment a mapped task takes even one fixed argument, partial is where it goes. partial/expand split the signature into “same on all” and “different on each.”

The mistake everyone makes: Cartesian vs. zipped

Now the sharp edge, and it’s the one thing in this chapter that will genuinely surprise you if nobody warns you.

You can pass more than one list to expand. When you do, Airflow does not walk the lists in lockstep — it takes the Cartesian product. Every combination.

process.expand(store=["oxford", "bath", "york"], fmt=["csv", "json"])
# 3 stores × 2 formats = SIX mapped instances, not three

That is sometimes exactly what you want — a grid of every store crossed with every export format. But it is very often not what people intend. The classic bug: you have three stores and three matching config files, you write .expand(store=stores, config=configs), you expect three instances pairing store-1 with config-1, and instead you get nine, pairing every store with every config. A ten-item fan-out that should have been ten becomes a hundred. Multiply two lists that each grow with the business and you can manufacture tens of thousands of task instances by accident.

When you want the lists paired — element i of one with element i of the other — you don’t use multiple expand arguments. You use expand_kwargs, which takes a single list of dictionaries, one dict of keyword arguments per instance:

load_file.partial(target_table="raw.orders").expand_kwargs([
    {"path": "/data/oxford.csv", "batch_id": "ox-01"},
    {"path": "/data/bath.csv",   "batch_id": "ba-01"},
    {"path": "/data/york.csv",   "batch_id": "yo-01"},
])
# exactly THREE instances — the dicts ARE the pairing

Three dicts, three instances, each with its own path and batch_id already zipped together. Here path and batch_id vary per instance while target_table stays constant via partial — the general shape when each item needs several fields.

The whole distinction lives in one line: multiple expand arguments = every combination; one expand_kwargs list = one instance per dict. If you find yourself with two lists you meant to pair, that’s the signal to reshape them into a list of dicts and reach for expand_kwargs.

And here’s the move that makes expand_kwargs shine: that list of dicts can itself be an upstream task’s return value. Instead of pairing lists at the last second and hoping you lined them up, you let one task build the exact work items — each already carrying every field its instance needs — and hand them straight to expand_kwargs:

@task
def plan_loads() -> list[dict]:
    # one dict per file, assembled where the pairing is obvious
    return [
        {"path": p, "batch_id": batch_id_for(p)}
        for p in find_csvs("/data/incoming/orders/")
    ]

@dag(schedule="@daily", catchup=False)
def load_orders():
    work = plan_loads()
    load_file.partial(target_table="raw.orders").expand_kwargs(work)

load_orders()

plan_loads decides both the fan-out width and the exact arguments for each instance, in one place, at run time. The zipping happens inside a single list comprehension where a human can see it, not implicitly across two expand arguments where a human can’t. When per-item work items get non-trivial, this “planner task feeds expand_kwargs” shape is the one to reach for — it keeps the pairing honest and the DAG code boring.

Collecting the results back: the reduce step

Fanning out is half the pattern. Usually you also want to fan back in — take the results of all those mapped instances and do one thing with them. A “reduce” step. In Airflow this is just an ordinary downstream task that consumes the mapped task’s output, and Airflow hands it the results as a list:

@task
def summarize(row_counts: list[int]) -> None:
    total = sum(row_counts)
    print(f"loaded {total:,} rows across {len(row_counts)} files")

@dag(schedule="@daily", catchup=False)
def load_orders():
    files = list_order_files()
    counts = load_file.partial(target_table="raw.orders").expand(path=files)
    summarize(counts)

load_orders()

counts is the collected output of every load_file instance. summarize receives it as a plain list — [812, 1_450, 903, ...] — and reduces it to one number. You didn’t do anything special to collect the results; passing the expanded task’s return into a normal task is the collect. This is the everyday shape of a mapping pipeline: a task that produces the work list, a mapped task that does the work in parallel, and a reduce task that consolidates. Discover, fan out, fan in.

When one of the mapped instances fails or skips

Real fan-outs have ragged edges — one store’s file is missing, another is empty. Two cases, and they behave differently, so it’s worth being precise.

A mapped instance that fails. By default the reduce task waits with the trigger rule all_success (the default you met in the branching chapter). One failed load_file instance means the mapped task, taken as a whole, isn’t fully successful, so summarize won’t run — which is often the right default, because a partial load you silently summarize is a partial load you silently ship wrong. When you genuinely want to proceed on what succeeded, give the reduce task a more permissive trigger rule such as all_done, and then defend it, because the collected list may now contain gaps from the instances that didn’t produce a value:

@task(trigger_rule="all_done")
def summarize(counts: list[int | None]) -> None:
    good = [c for c in counts if c is not None]
    print(f"loaded {sum(good):,} rows across {len(good)} files")
    if len(good) < len(counts):
        print(f"warning: {len(counts) - len(good)} file(s) did not load")

The all_done rule says “run once every mapped instance has reached some terminal state” — success, failure, or skip — so the reduce fires even with a failure in the batch. Filtering the Nones out before you sum is the defensive habit that keeps a warning from silently becoming a wrong total. The choice between the strict default and this permissive-plus-filter shape is a real decision about the pipeline’s contract: is a partial load acceptable and worth flagging, or must the whole batch be clean before anything downstream trusts it? Decide it deliberately rather than inheriting whichever trigger rule you happened to type.

A mapped instance that skips. If a particular item has legitimately no work — an empty file, a store that was closed — don’t fail it, skip it. Raise AirflowSkipException inside the task:

from airflow.sdk.exceptions import AirflowSkipException

@task
def load_file(path: str, target_table: str) -> int:
    rows = count_rows(path)
    if rows == 0:
        raise AirflowSkipException(f"{path} is empty, nothing to load")
    return copy_csv_to_warehouse(path, target_table)

A skip is not a failure — it’s “this instance had nothing to do.” The other instances are unaffected, and skipping is the honest signal for “no work here” rather than dressing it up as an error you’ll have to explain later. The difference between raising AirflowSkipException and letting a task error out is the difference between a clean empty-file run and a 3 a.m. page.

Readable map-index labels

One quality-of-life fix while you’re here. By default the mapped instances are labelled 0, 1, 2, … in the UI — fine until an instance fails and “instance 6 failed” tells you nothing about which store. You can name them with map_index_template, a Jinja string rendered against the task context, into which you write a value at run time:

from airflow.sdk import get_current_context, task

@task(map_index_template="{{ store_label }}")
def load_file(path: str) -> int:
    context = get_current_context()
    context["store_label"] = path.split("/")[-1]   # e.g. "oxford.csv"
    return copy_csv_to_warehouse(path)

Now the mapped node reads oxford.csv, bath.csv, york.csv instead of 0, 1, 2. The instance that failed names itself. On a nine-way fan-out this is the difference between reading logs and guessing.

Keeping the fan-out from stampeding

Dynamic does not mean unbounded, and a mapping that fans out too wide can hurt more than itself — a hundred instances all opening warehouse connections at once is a self-inflicted denial-of-service on the very database you’re loading. Three guardrails, from coarse to precise.

The hard ceiling: max_map_length. Airflow refuses to create more than [core] max_map_length mapped instances from a single mapping — the default is 1024. Return a list longer than that and the mapping fails rather than quietly spawning fifty thousand tasks. Treat this as a seatbelt, not a target: if you’re anywhere near 1024, the honest question is whether one Airflow task should be doing that much fan-out, or whether the work belongs in a batch engine you’re merely triggering.

Per-DAG throttle: max_active_tis_per_dag. Set this on the task to cap how many of its mapped instances run at the same time, across runs. You might have 40 files but only want 5 loading concurrently so the warehouse stays responsive:

@task(max_active_tis_per_dag=5)
def load_file(path: str) -> int:
    ...

All 40 instances still get created and still all run — five at a time, the rest queued. The fan-out width is unchanged; the burst is tamed.

The shared-resource valve: pools. max_active_tis_per_dag limits one task within one DAG. A pool limits concurrency across everything that draws on a shared resource. Give the warehouse a pool with, say, 10 slots, and tag every task that hits it — mapped or not, this DAG or another — with pool="warehouse". Now no matter how many DAGs fan out at once, only 10 warehouse-touching tasks run globally. Mapping against a pool is the standard way to stop a wide fan-out from becoming a warehouse stampede — it’s the difference between this task’s fan-out and the whole system’s pressure on one database.

Reach for the tightest tool that fits: max_map_length catches accidents, max_active_tis_per_dag shapes one task, a pool protects a shared resource from all comers.

TaskGroups: keeping the graph legible

Everything so far has been about how many tasks run. TaskGroups are about a completely different problem: how the graph reads. As pipelines grow, a flat DAG of thirty tasks becomes a wall of boxes nobody can parse at a glance. A TaskGroup bundles related tasks into one collapsible node in the UI and namespaces their task ids, so stage_orders.validate and publish.validate can coexist without colliding.

The ergonomic way is the @task_group decorator — it reads just like a function that contains tasks:

from airflow.sdk import dag, task, task_group

@task_group(group_id="stage_orders")
def stage_orders(path: str):
    cleaned = clean(path)
    validated = validate(cleaned)
    return load(validated)

Inside stage_orders, the three tasks wire together normally. From the outside, stage_orders behaves like a single unit you can depend on. Two things make groups genuinely useful beyond tidiness:

Group-level dependency wiring. You can set dependencies between whole groups, and Airflow expands that into the underlying edges:

extract_group >> stage_group >> publish_group

That one line says every task in extract_group must finish before any task in stage_group starts, and so on — the readable version of wiring dozens of individual edges.

Group default_args. Pass default_args to a group and every task inside inherits them — a natural home for “everything in this staging group retries twice and uses the warehouse pool,” set once instead of on each task.

Groups also nest — a group inside a group — which lets a large DAG read as a handful of top-level phases that you drill into only when something breaks:

@task_group(group_id="ingest")
def ingest():
    @task_group(group_id="orders")
    def orders():
        load(clean(extract_orders()))

    @task_group(group_id="inventory")
    def inventory():
        load(clean(extract_inventory()))

    orders()
    inventory()

Collapsed, ingest is one box. Open it and you see orders and inventory side by side; open one of those and you see its three tasks. The ids namespace all the way down — ingest.orders.load and ingest.inventory.load never collide even though both tasks are called load. On a fifty-task DAG this nesting is the difference between a diagram you can brief a teammate on and one you apologise for.

The one thing to internalise: a TaskGroup is pure presentation

Here’s the fact that prevents the most common misunderstanding. A TaskGroup has no runtime existence. It is not a sub-pipeline, it does not run as a unit, it has no state of its own, it cannot be retried or cleared as a whole. It is UI grouping and id namespacing — nothing more. At run time there is exactly one scheduler graph of individual tasks; the group is just how those tasks are drawn and named. Two tasks in the same group are no more isolated from the rest of the DAG than any other tasks.

That non-isolation is a feature, and it’s the crux of the SubDAG replacement story. Older Airflow had SubDAGs — an attempt to embed a whole DAG inside a task, with its own scheduling and its own executor semantics. They were a persistent source of grief: deadlocks, confusing concurrency that didn’t compose with the parent, and a second scheduling layer that behaved subtly differently from the first. SubDAGs are removed in Airflow 3. The replacement is the TaskGroup, and the replacement is better precisely because it does less: it gives you the visual grouping people actually wanted from SubDAGs, while keeping one scheduler graph with one consistent set of rules. You get the boxes without the second engine.

So if you’re arriving from an older tutorial (or an older codebase) reaching for a SubDAG, the translation is direct: it’s a TaskGroup now, and you should be relieved. You lose nothing real and shed a whole category of bugs.

Putting it together: the bookshop fan-out

Here’s the whole pattern in one DAG — discover today’s files, fan a per-file mini-pipeline out over them inside a group, then reduce:

from airflow.sdk import dag, task, task_group

@task
def list_order_files() -> list[str]:
    return find_csvs("/data/incoming/orders/")

@task_group(group_id="stage_orders")
def stage_orders(path: str) -> int:
    @task
    def validate(p: str) -> str:
        assert_schema(p)
        return p

    @task(map_index_template="{{ label }}")
    def load(p: str) -> int:
        from airflow.sdk import get_current_context
        get_current_context()["label"] = p.split("/")[-1]
        return copy_csv_to_warehouse(p, target_table="raw.orders")

    return load(validate(path))

@task
def summarize(counts: list[int]) -> None:
    print(f"loaded {sum(counts):,} rows across {len(counts)} files")

@dag(schedule="@daily", catchup=False)
def load_orders():
    files = list_order_files()
    counts = stage_orders.expand(path=files)   # map the GROUP over the files
    summarize(counts)

load_orders()

Two things are happening at once, and together they’re the payoff. stage_orders.expand(path=files) maps an entire TaskGroup over the file list — every file gets its own validate-then-load mini-pipeline, in parallel, each labelled by filename. And summarize(counts) reduces every load’s row count into one figure. Four files or nine, the DAG is the same three readable boxes; when a run goes wrong you expand exactly the one mapped instance that broke, named after the store that shipped the bad file. Discover, fan out, fan in — legible at any width.

Final thoughts

Dynamic task mapping is what lets a DAG’s shape follow the data instead of the data being forced to fit a DAG’s fixed shape. expand decides the width at run time, partial holds the constants still, and expand_kwargs is your defence against the Cartesian-product surprise that quietly multiplies a fan-out you meant to zip. Guard the width with max_map_length, max_active_tis_per_dag, and a pool, so parallelism is a lever you set rather than a stampede you discover. And TaskGroups keep all of it readable — the honest, better-because-simpler successor to SubDAGs, doing exactly one job: making a big graph legible without pretending to be a second engine underneath it.

Next: Templating, Macros, and Rendered Fields

expand: one task, many instances

partial: binding the arguments that don’t vary