Why not just build tasks in a for loop?

A loop runs at parse time, so the task count is fixed when the file is read and cannot depend on anything computed during the run. Dynamic task mapping decides the count at runtime from an upstream task's output, which is what you usually actually need.

Is there a limit on how many mapped tasks I can create?

max_map_length caps it at 1024 by default. Beyond a few hundred the scheduler and UI slow noticeably, since each mapped instance is a database row. For very wide fan-out, batch the work so each mapped task handles a chunk.

How do I pass more than one argument to a mapped task?

expand() maps over each keyword argument in parallel — same index together, not a cross product. For a cross product use expand_kwargs with a list of dicts, or build the pairs explicitly upstream.

What happens if one mapped task fails?

Only that instance fails. The others continue, and downstream tasks see the overall mapped task as failed. Use a trigger rule such as all_done on the downstream task if you want it to run on partial success.

Dynamic Task Mapping in Airflow

You often do not know how many things there are until the DAG is running — how many files landed, how many tenants are active, how many regions reported. Dynamic task mapping creates one task instance per item, at runtime.

The parse-time loop, and its limit

This is what people try first:

from datetime import datetime
from airflow.decorators import dag, task

REGIONS = ["uk", "us", "de"]        # known when the file is parsed

@dag(dag_id="static_loop", start_date=datetime(2026, 2, 1), schedule=None, catchup=False)
def pipeline():

    @task
    def process(region: str) -> int:
        print(f"processing {region}")
        return len(region)

    for r in REGIONS:
        process.override(task_id=f"process_{r}")(r)

pipeline()

airflow tasks list static_loop

process_de
process_uk
process_us

Three tasks, fine — as long as the list is a constant. It cannot come from a database query, an API call, or a directory listing, because the loop runs when the scheduler parses the file, not when the DAG runs.

Mapping over a runtime value

from datetime import datetime
from airflow.decorators import dag, task

@dag(dag_id="dynamic_map", start_date=datetime(2026, 2, 1), schedule=None, catchup=False)
def pipeline():

    @task
    def list_regions() -> list[str]:
        # In reality: a database query, an S3 listing, an API call.
        regions = ["uk", "us", "de", "fr", "jp"]
        print(f"found {len(regions)} regions")
        return regions

    @task
    def process(region: str) -> dict:
        rows = {"uk": 1284, "us": 3120, "de": 890, "fr": 1105, "jp": 2077}[region]
        print(f"processing {region}: {rows} rows")
        return {"region": region, "rows": rows}

    @task
    def summarise(results: list[dict]) -> None:
        total = sum(r["rows"] for r in results)
        print(f"{len(results)} regions, {total} rows total")
        for r in sorted(results, key=lambda x: -x["rows"]):
            print(f"  {r['region']:4} {r['rows']:6}")

    summarise(process.expand(region=list_regions()))

pipeline()

airflow tasks list dynamic_map

list_regions
process
summarise

Three tasks in the definition — but process becomes five instances at runtime:

airflow dags trigger dynamic_map && sleep 25
airflow tasks states-for-dag-run dynamic_map manual__2026-02-17T12:20:00+00:00

dag_id      | task_id      | map_index | state
============+==============+===========+=========
dynamic_map | list_regions | -1        | success
dynamic_map | process      | 0         | success
dynamic_map | process      | 1         | success
dynamic_map | process      | 2         | success
dynamic_map | process      | 3         | success
dynamic_map | process      | 4         | success
dynamic_map | summarise    | -1        | success

map_index identifies each instance. -1 means unmapped. Change the upstream list to seven regions and you get seven instances with no DAG edit.

The reduce step receives all the results as a list:

INFO - 5 regions, 8476 rows total
INFO -   us      3120
INFO -   jp      2077
INFO -   uk      1284
INFO -   fr      1105
INFO -   de       890

That fan-out then fan-in is the whole pattern.

Mapping over several arguments

expand zips its arguments by index:

    @task
    def load(region: str, table: str) -> str:
        result = f"{region}.{table}"
        print(f"loading {result}")
        return result

    load.expand(region=["uk", "us", "de"], table=["orders", "users", "events"])

INFO - loading uk.orders
INFO - loading us.users
INFO - loading de.events

Three instances, paired by position — not nine. For a cross product build the pairs yourself:

    @task
    def pairs() -> list[dict]:
        import itertools
        return [{"region": r, "table": t}
                for r, t in itertools.product(["uk", "us"], ["orders", "users"])]

    load.expand_kwargs(pairs())

INFO - loading uk.orders
INFO - loading uk.users
INFO - loading us.orders
INFO - loading us.users

expand_kwargs takes a list of dicts, one per instance, with each dict supplying that instance’s arguments.

Constants alongside mapped values

    @task
    def write(region: str, run_date: str, bucket: str) -> None:
        print(f"s3://{bucket}/{run_date}/{region}.parquet")

    write.partial(bucket="analytics-prod", run_date="2026-02-14").expand(
        region=["uk", "us", "de"]
    )

INFO - s3://analytics-prod/2026-02-14/uk.parquet
INFO - s3://analytics-prod/2026-02-14/us.parquet
INFO - s3://analytics-prod/2026-02-14/de.parquet

partial() fixes the arguments that do not vary; expand() supplies the ones that do.

Partial failure

    @task
    def flaky(region: str) -> str:
        if region == "de":
            raise ValueError(f"no data for {region}")
        return region

dynamic_map | flaky     | 0 | success
dynamic_map | flaky     | 1 | success
dynamic_map | flaky     | 2 | failed
dynamic_map | flaky     | 3 | success
dynamic_map | flaky     | 4 | success
dynamic_map | summarise | -1| upstream_failed

Four succeeded, one failed, and the reduce step did not run. To proceed on partial success:

    @task(trigger_rule="all_done")
    def summarise_partial(results: list) -> None:
        good = [r for r in results if r is not None]
        print(f"{len(good)} of {len(results)} succeeded")

INFO - 4 of 5 succeeded

Decide this deliberately. Continuing on partial data is right for a best-effort report and wrong for a financial reconciliation.

Mapping over files

The realistic version — you do not know what landed:

    @task
    def find_files() -> list[str]:
        from pathlib import Path
        files = sorted(str(p) for p in Path("/data/incoming").glob("*.csv"))
        print(f"found {len(files)} files")
        return files

    @task
    def load_file(path: str) -> int:
        import csv
        with open(path) as f:
            n = sum(1 for _ in csv.reader(f)) - 1
        print(f"{path}: {n} rows")
        return n

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

    report(load_file.expand(path=find_files()))

INFO - found 3 files
INFO - /data/incoming/2026-02-14-a.csv: 512 rows
INFO - /data/incoming/2026-02-14-b.csv: 431 rows
INFO - /data/incoming/2026-02-14-c.csv: 341 rows
INFO - loaded 1284 rows from 3 files

Three files today, twelve tomorrow, and the DAG is unchanged.

Limits

    @task
    def too_many() -> list[int]:
        return list(range(5000))

airflow.exceptions.AirflowException:
Object of type list has length 5000 which exceeds the max_map_length limit of 1024

Raise AIRFLOW__CORE__MAX_MAP_LENGTH if you must, but the better fix is batching:

    @task
    def batched() -> list[list[int]]:
        items = list(range(5000))
        size = 100
        return [items[i:i + size] for i in range(0, len(items), size)]

    @task
    def process_batch(batch: list[int]) -> int:
        return sum(batch)

INFO - 50 mapped instances, 100 items each

Fifty instances instead of five thousand. Each mapped instance is a scheduler database row and a UI element, so keeping the count in the tens or low hundreds keeps everything responsive.

Practice

1. Map over an empty list.

dynamic_map | process   | -1 | skipped
dynamic_map | summarise | -1 | success

Zero instances, and the mapped task is marked skipped rather than failing. The downstream task still runs, receiving an empty list — so handle that case, or a “no files today” run will divide by zero in your summary.

2. Chain two mapped tasks.

    second.expand(value=first.expand(x=[1, 2, 3]))

first  | 0,1,2 | success
second | 0,1,2 | success

Instance i of second consumes instance i of first — the mapping carries through without a reduce step in between, so you can build a mapped pipeline several stages deep.

3. Use expand with two lists of different lengths.

AirflowException: expand() got mismatched lengths: region has 3, table has 2

Airflow refuses rather than truncating or padding, which is the right call — silently dropping a region would be far worse than failing at parse time.

4. How many task instances does 3 regions x 4 tables produce with expand vs expand_kwargs?

expand with equal-length lists gives one instance per index. With 3 and 4 it errors. To get all 12 combinations, build the 12 dicts upstream and use expand_kwargs. Airflow deliberately does not do the cross product implicitly, because an accidental one is the difference between 12 tasks and 12,000.

Next: waiting for external conditions before a DAG proceeds.