airflow dynamic task mapping expand reduce

TL;DR

Decorate the task with @task and call .expand(arg=[...]) to fan out one task per list element at runtime; collect results with .expand() on a downstream task or reduce manually with XCom. Mapping replaces code-generated static tasks with data-driven parallelism.

airflow dynamic task mapping expand reduce

Use this when

• The number of parallel tasks depends on runtime data
• You currently generate tasks in a Python loop
• You need to aggregate mapped task outputs

Not for this skill when

• You are configuring deferrable operators
• You need SLA miss alerts
• Sensors time out

Steps

1. The basic expand pattern:

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

@dag(start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def mapped_pipeline():
    @task
    def list_tables():
        return ["orders", "users", "events"]
    @task
    def process(table: str):
        print(f"processing {table}")
    process.expand(table=list_tables())

mapped_pipeline()

Expected output: three process task instances at runtime, one per table. The list comes from an upstream task, so the fan-out is data-driven, not hardcoded.

1. Expand over multiple arguments with expand_kwargs:

@task
def load(table: str, date: str):
    ...
load.expand_kwargs([{"table": t, "date": d} for t in tables for d in dates])

Expected output: one task instance per dict. Use expand() for a single list argument, expand_kwargs() for combinations.

1. Reduce: aggregate mapped outputs in a downstream task:

@task
def process(table: str) -> int:
    return row_count(table)
@task
def summarize(counts: list[int]):
    print(f"total rows: {sum(counts)}")
counts = process.expand(table=list_tables())
summarize(counts)

Expected output: summarize receives the list of all mapped return values. Passing a mapped XCom into a regular task automatically reduces it.

1. Control parallelism so a 10,000-element expand does not crush the system:

@task(max_active_tis_per_dag=16)
def process(table: str):
    ...

Expected output: at most 16 mapped instances running at once per DAG run. Without a cap, a large expand can starve every other DAG of workers.

1. Know the limits of mapping vs static tasks:

Mapped tasks: great for data-driven fan-out; the graph shows one
node that expands at runtime. Debugging one bad element means
finding it among N instances.
Static loop tasks: visible individually in the graph; better when
the set is fixed and small.

Expected output: the right choice per use case. Do not map over a fixed list of 3 known tables just because you can; a static loop is clearer.

Variant phrasings

airflow dynamic tasks variable number

.expand() over an upstream task's output list (step 1). The count is decided at runtime.

airflow mapped task xcom aggregate

Pass the mapped result into a downstream task; Airflow collects the list (step 3).

airflow expand vs static tasks

Expand for runtime-determined parallelism, static loops for fixed small sets (step 5).

Why it happens

Classic Airflow builds the DAG at parse time, so "N tasks" meant N task objects in code. Dynamic task mapping defers the fan-out to runtime: one mapped task definition expands into N task instances based on actual data, which is what data pipelines usually need.

Edge cases

• The expand argument must be a list (or XCom resolving to one); a generator or lazy object fails at runtime.
• Mapped tasks need Airflow 2.3+; max_active_tis_per_dag needs 2.6+.
• Very large expands (100k+) strain the scheduler and the metadata DB; chunk the input list.
• Mapped task instance logs are per map index; the UI groups them, use the index filter to find the failing element.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_c91K0PkO3Bg4LOdvMakgbA