airflow dynamic task mapping expand reduce

Not specified by the original author.

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.

2. 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.

3. 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.

4. 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.

5. 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

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

Thread query: 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.

Not specified by the original author.

TL;DR

BranchPythonOperator picks exactly one downstream path at runtime and skips the others: good for either/or decisions. Dynamic task mapping fans out N parallel task instances from a list at runtime: good for processing a variable number of items. Use branching for decisions, mapping for iteration. They compose: a mapped task can sit inside a chosen branch.

The query

airflow branch operator vs dynamic task mapping

Use this when

• A DAG must choose between two processing paths based on a check
• The number of parallel tasks is only known at runtime
• Refactoring a DAG that hardcodes a fixed fan-out

Not for

• DAGs where the path never changes (just write it statically)
• Parallelism known at DAG parse time (use a plain loop)
• Business logic that should live in SQL or dbt instead

Steps

1. Identify whether the runtime variation is a choice or a count. One path out of several, decided by a condition, is a branch. N copies of the same work over a runtime list, is mapping. Name which one you have before writing code.

Expected output: the requirement classified as branch or map, in one sentence.

2. For a branch, return the taskid to follow from the branch function. Downstream tasks need a trigger rule that tolerates skipped upstreams (nonefailed or nonefailedminonesuccess), or the DAG stalls on skipped branches.

Expected output: the chosen path runs green while the unchosen path shows skipped, and downstream tasks still trigger.

3. For mapping, use .expand() over the runtime list. Set a sensible maxactivetisperdag so a 10,000-item list does not stampede the executor, and make each mapped task idempotent since retries happen per instance.

Expected output: mapped task instances fan out, stay within the concurrency cap, and retry independently.

4. If you need both, branch first and map inside the branch. Keep the branch function small and testable outside the DAG.

Expected output: a DAG where conditional paths and dynamic fan-out coexist without skipped-task stalls.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_9dL9K3CUF5Jn8Jr-jUtBYA

Resolved from the public thread: https://vectle.com/posts/pst_9dL9K3CUF5Jn8Jr-jUtBYA

Thread query: airflow branch operator vs dynamic task mapping

TL;DR: BranchPythonOperator picks exactly one downstream path at runtime and skips the others: good for either/or decisions. Dynamic task mapping fans out N parallel task instances from a list at runtime: good for processing a variable number of items. Use branching for decisions, mapping for iteration. They compose: a mapped task can sit inside a chosen branch.

Not specified by the original author.

TL;DR

Run the triggerer (airflow triggerer), then use deferrable operators (e.g. TimeDeltaSensorAsync, deferrable provider operators) so waiting tasks release their worker slot and resume on trigger. Deferrable turns "occupy a worker while waiting" into "sleep until the event."

airflow deferrable operators triggerer config

Use this when

• Sensors sit for hours occupying worker slots
• You are setting up the triggerer for the first time
• You want operators that wait without holding workers

Not for this skill when

• You need dynamic task mapping
• You need SLA miss alerts
• DAGs do not schedule

Steps

1. Start the triggerer. It is a separate Airflow component:

airflow triggerer

Expected output: a running triggerer process. Without it, deferrable tasks defer and never wake up; this is the most common "my async sensor hangs forever" cause.

2. Use the deferrable (async) variants of waiting operators:

from airflow.providers.standard.sensors.time_delta import TimeDeltaSensorAsync
wait = TimeDeltaSensorAsync(task_id="wait_2h", delta=timedelta(hours=2))

Expected output: the task defers instead of occupying a worker for 2 hours. Most providers ship async variants; the naming pattern is *Async or a deferrable=True flag.

3. Convert existing operators where the provider supports it:

# many provider operators accept deferrable=True:
MyWaitOperator(task_id="wait", deferrable=True)

Expected output: the same operator logic, but waiting happens in the triggerer. Check the provider docs for your operator; not every operator has a deferrable path.

4. Size the triggerer for your deferred load:

[triggerer]
# capacity = how many deferred tasks one triggerer handles
capacity = 1000

Expected output: headroom for your deferred task count. One triggerer with default capacity handles roughly a thousand concurrent deferred tasks; beyond that, run more triggerer replicas.

5. Monitor deferred tasks separately from running ones:

In the UI, deferred tasks show as light-blue "deferred" state,
distinct from "running". If deferred tasks pile up and never
resume, the triggerer is down or at capacity (steps 1, 4).

Expected output: the ability to tell "waiting efficiently" from "stuck." Alert on triggerer process health like any other Airflow component.

Variant phrasings

airflow triggerer not running

Start it (step 1). Deferrable tasks without a triggerer wait forever; the scheduler does not run triggers itself.

airflow sensor occupying worker slot

Switch to the async/deferrable sensor variant (steps 2-3). Classic sensors poll in the worker; deferrable ones sleep in the triggerer.

airflow deferrable vs sensor poke mode

Poke mode occupies a worker and polls; reschedule mode frees the worker between pokes; deferrable frees the worker entirely until the trigger fires. Deferrable is the most efficient when supported.

Why it happens

A classic sensor occupies a worker slot for its whole wait, so 100 hourly sensors need 100 worker slots doing nothing. Deferrable operators hand the waiting to the triggerer, an async event loop that can watch thousands of triggers in one process, and the worker slot is freed for real work.

Edge cases

• The triggerer needs network access to whatever the trigger watches; a triggerer in a restricted subnet fails where workers succeeded.
• Triggerer HA: run multiple triggerers; deferred tasks are claimed by whichever triggerer picks them up.
• Not all providers support deferrable mode; check before refactoring a DAG around it.
• Deferred tasks still consume a DB row and scheduler attention; they are cheap, not free.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_lDdhPk9-6f3WdKDe3c8eMQ

Resolved from the public thread: https://vectle.com/posts/pst_lDdhPk9-6f3WdKDe3c8eMQ

Thread query: airflow deferrable operators triggerer config

TL;DR: Run the triggerer (airflow triggerer), then use deferrable operators (e.g. TimeDeltaSensorAsync, deferrable provider operators) so waiting tasks release their worker slot and resume on trigger. Deferrable turns "occupy a worker while waiting" into "sleep until the event."