Airflow Dynamic Task Mapping: Fan-Out at Runtime + Expand / Partial Patterns

airflow dynamic task mapping is the runtime fan-out primitive that replaced every "loop-generated DAG" pattern the Airflow community shipped between 2015 and 2022 — and the single feature senior data engineers reach for the moment they see a for loop building tasks in a top-level DAG file. Before Airflow 2.3, if you wanted to process N files, N regions, or N tenants in parallel, you looped over a hard-coded list at DAG-parse time to build N task nodes. That pattern is brittle (the list has to be known at parse time), unbounded (nothing stops the loop from generating 10 000 tasks that crush the scheduler), and impossible to unit-test (the DAG structure changes with the environment). The .expand() / .partial() API collapses all of that into one line — process_file.expand(key=list_files.output) — and the mapped instances materialise at runtime from whatever the upstream returned.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "what's the difference between airflow expand and airflow partial?" or "what happens if the upstream returns 100 000 items?" or "how do you unit-test a mapped task without spinning up Airflow?" It walks through why loop-generated DAGs are the anti-pattern, the four axes every senior interview probes (airflow map_index, expand vs partial, upstream fan-in, cardinality limits), the taskflow expand decorator-first workflow, the dynamic task groups pattern that maps whole subgraphs across airflow parallel tasks, and the production knobs — max_map_length, max_active_tasks_per_dag, backfill semantics, and the pytest recipe — that senior engineers ship into every mapped-task DAG. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse on the SQL practice library →, and sharpen the tuning axis with the optimization practice library →.

---

On this page

• Why dynamic task mapping replaced loop-generated DAGs
• Basic expand + partial
• Upstream fan-in — mapping from a task's output
• Nested expand + TaskGroup mapping
• Production patterns — limits, backpressure, testing
• Cheat sheet — dynamic mapping recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why dynamic task mapping replaced loop-generated DAGs

Fan-out at parse time is the anti-pattern — airflow dynamic task mapping moves the decision to run time

The one-sentence invariant: before Airflow 2.3, fan-out required a Python for loop at DAG-parse time that hard-coded the number and shape of the mapped tasks; airflow dynamic task mapping (2.3+) uses .expand(...) to defer the decision until run time, when the upstream task's actual output is known. The distinction sounds pedantic and is anything but — it is the difference between a DAG whose structure is fixed at deploy time and a DAG that adapts to whatever the upstream returned on this particular run. Every senior Airflow question you will face about airflow fan out, airflow parameterised tasks, or airflow map reduce is a consequence of that one design choice.

The four axes interviewers actually probe.

• .expand() vs .partial(). Two decorators on the same task, doing complementary jobs: partial pins the fixed keyword arguments once, expand maps over the varying keyword arguments to spawn one instance per element. The senior signal is naming both without prompting and knowing that .partial().expand() is a single call chain — you cannot use one without understanding the other.
• airflow map_index. Each mapped instance carries a zero-based map_index that shows up in the UI, in log filenames, in XCom keys, and in context["ti"].map_index inside the task body. Interviewers use map_index questions to check that the candidate has actually shipped mapped tasks (as opposed to just read the docs) — the accounting only matters once you have debugged one failure among 200 instances.
• Upstream fan-in. The killer pattern: an upstream task returns a list; the downstream declares .expand(key=upstream.output) and Airflow generates one mapped instance per element of the returned list. The list can have zero elements (downstream is skipped), one element (downstream runs once with map_index=0), or 10 000 elements (downstream fans out to 10 000 mapped instances). Getting the last case wrong is how mapped-task DAGs OOM the scheduler.
• Cardinality limits. The max_map_length per-task cap (default 1024 in 2.x, configurable higher) and the max_active_tasks_per_dag DAG-level backpressure. Both exist because the scheduler cannot cheaply materialise millions of task instances; the senior interview answer names both without being prompted.

What changed in Airflow 2.3, 2.5, 2.7, and 2.9.

• 2.3 (April 2022). .expand() and .partial() land. Single-level mapping only. expand accepts positional-ish keyword arguments where every value is a list-like or an XComArg.
• 2.5 (December 2022). expand_kwargs() variant lands, letting you fan out over a list of kwarg dictionaries rather than one list per kwarg. The right escape hatch when your fan-out unit is a heterogeneous dict, not a scalar.
• 2.7 (August 2023). Nested dynamic task mapping. A mapped task's output can itself be .expand()-ed by a downstream task — the multiplicative fan-out story. This is the version where the "explosion trap" starts mattering.
• 2.9 (April 2024). Mapping over TaskGroups. Whole subgraphs (a group of two-to-ten tasks) can be mapped as a single unit, so the "one region → three-step subgraph" pattern becomes native rather than hand-written.
• 2026 reality. Every modern Airflow deployment relies on dynamic mapping for at least one workflow. Interviewers assume you know it; the differentiator is whether you know the limits — cardinality caps, backfill semantics, XComArg chaining — not the syntax.

Why the loop-generated pattern is dead.

• Parse-time cardinality. A for key in list_files(): task = PythonOperator(...) block runs every time the scheduler parses the DAG file — every 30 seconds by default. If list_files() hits S3, you have made an S3 call every 30 seconds forever, in every scheduler and worker process. Dynamic mapping runs list_files() once per DAG run at the actual task time.
• Unbounded fan-out. The loop pattern has no cap; the fan-out is whatever the list happens to contain. Dynamic mapping ships with max_map_length per task and max_active_tasks_per_dag per DAG — the scheduler refuses to materialise more instances than the cap allows.
• Impossible to unit-test. The loop pattern couples the DAG structure to the environment (dev vs prod S3 buckets). Dynamic mapping decouples: the DAG structure is one node marked "mapped"; the runtime cardinality is decided by whatever the upstream returned in that environment.
• UI unusable at scale. The loop pattern renders N graph nodes; at 200 nodes the graph view becomes unreadable. Dynamic mapping renders one node with a small "200" badge and the mapped instances live on a paginated grid view — designed for high cardinality from the start.
• Backfill blows up. The loop pattern regenerates the tasks at every parse; a change to the underlying list during a backfill silently changes the DAG structure mid-run. Dynamic mapping locks the mapped cardinality per DAG run — a backfill run replays the exact set of instances that ran the first time.

What interviewers listen for.

• Do you say "parse-time vs run-time cardinality" in the first sentence when asked why dynamic mapping was added? — senior signal.
• Do you name .expand(), .partial(), expand_kwargs(), and map_index in the first 30 seconds? — required baseline.
• Do you mention the max_map_length cap and max_active_tasks_per_dag backpressure without being prompted? — senior signal.
• Do you push back on "just use a loop over the DAG" with the parse-time / testability / backfill argument? — required answer.
• Do you describe mapped tasks as "one task node, N instances" rather than as "N tasks"? — required framing.

Worked example — the loop-generated anti-pattern

Detailed explanation. The textbook anti-pattern that senior interviewers open with: a team writes a DAG that lists S3 keys at DAG-parse time using a Python for loop, spawning one PythonOperator per key. The DAG "works" in dev, then explodes in prod when the bucket contains 5 000 files, and the scheduler process starts using 4 GB of RAM parsing the DAG every 30 seconds. Walk an interviewer through what actually happens and why .expand() fixes it.

• The symptom. Scheduler process CPU pinned at 100%; DAG-parse latency > 60 seconds; airflow tasks list hangs.
• The naive diagnosis. "The bucket is big; we need a bigger scheduler." Wrong — the bucket is a symptom, not the cause.
• The real bug. Every parse re-lists the bucket. 5 000 files × 2 parses/minute × 60 minutes/hour = 600 000 S3 LIST calls per hour just to keep the DAG loaded.
• The fix. Move the LIST call into an actual task that runs once per DAG run and return the list; .expand(key=list_task.output) downstream.

Question. A team ships this DAG. Point out every problem, then rewrite it using dynamic task mapping. Quantify the S3-cost and scheduler-CPU improvement.

Input.

Metric	Loop-generated DAG	Target after fix
S3 LIST calls per hour	600 000	24 (once per DAG run, hourly schedule)
Scheduler CPU during parse	100%	~5%
DAG-parse latency	60 s	< 1 s
Files listed	5 000	5 000 (unchanged; just once)
Mapped instances at run	5 000 explicit tasks	1 mapped node × 5 000 instances

Code.

# ANTI-PATTERN — loop-generated DAG (Airflow 2.2 style)
from datetime import datetime
import boto3
from airflow import DAG
from airflow.operators.python import PythonOperator

def process_file(key: str) -> None:
    # ... process one S3 object ...
    pass

# THIS RUNS EVERY DAG PARSE — every 30 seconds forever
s3 = boto3.client("s3")
resp = s3.list_objects_v2(Bucket="raw-events")
keys = [obj["Key"] for obj in resp.get("Contents", [])]

with DAG(
    dag_id="anti_pattern_loop",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
) as dag:
    for k in keys:                                  # 5000 iterations at parse time
        PythonOperator(
            task_id=f"process_{k.replace('/', '_')}",
            python_callable=process_file,
            op_kwargs={"key": k},
        )

# CORRECT — dynamic task mapping (Airflow 2.3+)
from datetime import datetime
import boto3
from airflow.decorators import dag, task

@dag(
    dag_id="mapped_process_files",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
)
def mapped_process_files() -> None:

    @task
    def list_files(bucket: str) -> list[str]:
        s3 = boto3.client("s3")
        resp = s3.list_objects_v2(Bucket=bucket)
        return [obj["Key"] for obj in resp.get("Contents", [])]

    @task
    def process_file(key: str) -> None:
        # ... process one S3 object ...
        pass

    keys = list_files(bucket="raw-events")
    process_file.expand(key=keys)                   # 5000 instances at runtime

mapped_process_files()

Step-by-step explanation.

1. The anti-pattern makes boto3.client("s3").list_objects_v2(...) a top-level module import side effect. Every Airflow scheduler DAG-file parse (default every 30 seconds) triggers a fresh S3 LIST — 5 000 keys returned, 5 000 task nodes constructed, all before the scheduler decides whether any of this DAG's tasks even need to run.
2. The corrected pattern moves the LIST into a @task decorated function. It runs once per DAG run, at the actual scheduled time, on a worker. The scheduler only sees two nodes in the DAG graph — list_files and process_file (marked as mapped).
3. The list_files.output return value is an XComArg — a promise object that tells the scheduler "the value of this XCom is a list; use it for .expand() at runtime." No S3 call happens at parse time; the promise is just wire.
4. At run time, when list_files finishes, the scheduler pulls its XCom value (the 5 000-key list) and materialises 5 000 mapped instances of process_file, each with a distinct map_index (0 through 4 999) and op_kwargs={"key": keys[i]}.
5. The scheduler CPU during parse drops from 100% to ~5% — parsing a two-node DAG is trivial. The S3 LIST cost drops from 600 000 per hour to 24 per day at hourly schedule. The functional behaviour — 5 000 files processed per hour — is unchanged.

Output.

Metric	Loop-generated	Dynamic mapping	Delta
S3 LIST calls / hour	600 000	1	600 000× reduction
Scheduler CPU (parse)	100%	5%	20× reduction
DAG-parse latency	60 s	< 1 s	60× faster
Nodes in DAG graph	5 000	2	2 500× fewer
Backfill semantics	breaks (list changes)	stable (list frozen per run)	correctness fix

Rule of thumb. Any for loop that constructs operators in a DAG file body is an anti-pattern in 2026. If the loop's iterable comes from an I/O call, the fix is .expand(...) off an upstream task. If the iterable is a genuinely static list (say, the 4 quarters of the year), the loop is technically fine — but even then, prefer .expand() for uniformity.

Worked example — the four axes an interviewer probes

Detailed explanation. A senior interviewer opens with "walk me through Airflow dynamic task mapping." The candidate who lists .expand() and .partial() and stops has answered the junior-level question. The senior-level answer names four axes — expand vs partial, map_index, upstream fan-in, cardinality limits — and shows a minimal DAG that exercises all four. This is the answer that ends the topic block.

• The axes. expand/partial, map_index, upstream fan-in, cardinality caps.
• The minimal DAG. One upstream returns a list; one downstream .expand()-es over it with a partial-fixed kwarg; a downstream reducer runs after all mapped instances complete.
• The signal. Naming all four axes without prompting; showing map_index in the log; naming the max_map_length cap.

Question. Produce a minimal DAG that touches all four axes and walk through what the interviewer sees in the UI.

Input.

Axis	Feature	Where it shows
expand	mapped kwarg	task decorator
partial	fixed kwarg	task decorator
map_index	per-instance index	UI grid, log filename, ti.map_index
upstream fan-in	XComArg chaining	.expand(key=upstream.output)
cardinality	max_map_length	task attribute or airflow.cfg

Code.

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

@dag(
    dag_id="four_axes_demo",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
)
def four_axes_demo() -> None:

    @task
    def list_keys() -> list[str]:
        # Axis 3 — upstream fan-in source
        return ["k1", "k2", "k3", "k4", "k5"]

    @task(max_active_tis_per_dag=3)                 # Axis 4 — cardinality control
    def process_key(bucket: str, key: str, **context) -> str:
        # Axis 2 — read map_index from the task instance
        idx = context["ti"].map_index
        print(f"map_index={idx} bucket={bucket} key={key}")
        return f"{bucket}/{key}"

    @task
    def reduce(processed: list[str]) -> int:
        # Reducer runs once after all mapped instances complete
        return len(processed)

    keys = list_keys()
    # Axis 1 — partial pins bucket; expand maps over key
    results = process_key.partial(bucket="raw").expand(key=keys)
    reduce(results)

four_axes_demo()

Step-by-step explanation.

1. list_keys is a standard @task; it returns a Python list of five string keys. Its return value becomes an XComArg — a lazy reference to "the list this task returned."
2. process_key.partial(bucket="raw") freezes the bucket argument as the constant "raw" for every mapped instance. .expand(key=keys) fans out over the keys list; each mapped instance receives key=keys[i] and the fixed bucket="raw". The result: 5 mapped instances of process_key, each with distinct map_index values 0 through 4.
3. Inside process_key, context["ti"].map_index returns the per-instance index. This is how the task body knows which mapped instance it is — useful for map_index-aware logging, per-instance file naming, or targeted retry logic.
4. max_active_tis_per_dag=3 is the per-task concurrency cap: at most three mapped instances of process_key run concurrently. The remaining two queue and run as the first three finish. This is the per-task backpressure valve.
5. reduce(results) takes the list of return values from all mapped process_key instances and runs once as a standard (non-mapped) task. This is the map-reduce pattern — reduce receives the collected list, which Airflow assembles automatically from each mapped instance's XCom return value.

Output.

UI element	What appears
DAG graph view	list_keys → process_key → reduce (3 nodes, process_key badged "5")
Grid view	5 rows under process_key: map_index 0, 1, 2, 3, 4
Log filename	dag_id=four_axes_demo/task_id=process_key/run_id=.../map_index=0/attempt=1.log
Task instance detail	map_index=2 shown in the sidebar; key="k3" in rendered template
Reducer input	Python list of 5 strings, in the order the mapped instances completed

Rule of thumb. The four-axes answer — expand vs partial, map_index, upstream fan-in, cardinality — is the phrase that separates the junior "I read the docs" answer from the senior "I ship this in prod" answer. Rehearse the minimal DAG above; if you can produce it from memory, you have covered the senior-level baseline.

Worked example — parse-time vs run-time cardinality accounting

Detailed explanation. A common interview trap: the candidate writes a DAG that mixes parse-time cardinality (a Python for loop) with dynamic mapping. The DAG runs in dev, then breaks in prod because the parse-time list and the run-time list disagree. Walk through why this pattern is unstable and the two-decorator fix.

• The trap. The developer partially trusts dynamic mapping and partially trusts the loop, ending up with both.
• The instability. Parse-time cardinality is fixed at deploy; run-time cardinality varies per run; conflict between them surfaces as "task X was expected but never ran" errors.
• The fix. Choose one — dynamic mapping — and eliminate every parse-time cardinality assumption.

Question. A DAG loops over a hard-coded list of 3 regions and, inside each region, uses .expand() over an S3 key list. In prod, one region has 0 keys. Trace what happens under both patterns and rewrite as a proper nested-expand DAG.

Input.

Layer	Parse-time (loop)	Run-time (expand)
Regions	["us", "eu", "asia"]	list_regions() → XComArg
Files per region	list at parse	list_files.expand(region=...)
Behaviour when region is empty	3 tasks exist but process_file skipped	mapped instance count = 0; downstream skipped or empty

Code.

# Mixed pattern — parse-time regions + run-time files (fragile)
from datetime import datetime
from airflow.decorators import dag, task

@dag(dag_id="mixed_bad", start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def mixed_bad() -> None:
    for region in ["us", "eu", "asia"]:            # PARSE TIME — hard-coded

        @task(task_id=f"list_files_{region}")
        def list_files() -> list[str]:
            return fetch_keys(region)              # RUN TIME — fetched at run

        @task(task_id=f"process_{region}")
        def process_file(key: str) -> None:
            handle(region, key)

        process_file.expand(key=list_files())

# Fully dynamic — the right pattern
from datetime import datetime
from airflow.decorators import dag, task

@dag(dag_id="fully_dynamic", start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def fully_dynamic() -> None:

    @task
    def list_regions() -> list[str]:
        return fetch_regions()                     # returns whatever exists today

    @task
    def list_files(region: str) -> list[str]:
        return fetch_keys(region)

    @task
    def process_file(region: str, key: str) -> None:
        handle(region, key)

    regions = list_regions()
    keys_per_region = list_files.expand(region=regions)
    # Flatten (region, key) pairs — one instance per (region, key)
    process_file.partial().expand_kwargs(
        [{"region": r, "key": k}
         for r, ks in zip(regions, keys_per_region) for k in ks]
    )

fully_dynamic()

Step-by-step explanation.

1. In the mixed pattern, three task ids are hard-coded at parse time (process_us, process_eu, process_asia). If the deployment later adds latam, the DAG must be edited and redeployed — the run-time list of regions cannot influence the DAG structure.
2. In the fully dynamic pattern, list_regions() returns whatever regions exist today. The next run may see 3 or 4 or 12 regions; the DAG adapts without redeployment.
3. list_files.expand(region=regions) fans out list_files — one instance per region. Each returns a list of keys. The result is a list of lists — nested cardinality.
4. process_file.partial().expand_kwargs(...) uses the kwargs variant to fan out over a flat list of (region, key) dictionaries. This is the correct way to consume nested cardinality when each atomic unit of work needs multiple parameters.
5. The key correctness win: when a region has 0 keys, its list_files returns [], contributing 0 entries to the expand_kwargs list. process_file is skipped for that region without any bookkeeping — Airflow's zero-mapping semantics handle it natively.

Output.

Scenario	Mixed pattern	Fully dynamic
3 regions, all with files	works	works
3 regions, 1 empty	works but 1 mapped task shows "no work"	0 instances for empty region; clean
4th region added in prod	needs code change	picked up automatically
Backfill semantics	region list drifts with edits	frozen per run

Rule of thumb. Never mix parse-time and run-time cardinality in the same DAG. Pick one — always run-time — and shape the DAG so every "how many" question is answered by an upstream task, not by the DAG file body.

Senior interview question on why dynamic task mapping replaced loop-generated DAGs

A senior interviewer often opens with: "Walk me through what changed in Airflow 2.3 that made loop-generated fan-out obsolete, and what problems .expand() actually solves in production."

Solution Using the parse-time vs run-time cardinality framing

# The framing answer as a single DAG that documents the trade-off
from datetime import datetime
from airflow.decorators import dag, task

@dag(
    dag_id="parse_vs_runtime",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
)
def parse_vs_runtime() -> None:

    @task
    def list_files_at_runtime() -> list[str]:
        # Runs ONCE per DAG run, on a worker, at scheduled time
        return list_s3_keys("raw-events")

    @task(retries=2)
    def process_file(bucket: str, key: str) -> None:
        # One instance per key at runtime
        handle(bucket, key)

    keys = list_files_at_runtime()
    process_file.partial(bucket="raw-events").expand(key=keys)

parse_vs_runtime()

Step-by-step trace.

Step	Before (loop-generated)	After (dynamic mapping)
DAG-parse cost	S3 LIST every 30 s	zero I/O at parse
Scheduler nodes	N (one per key)	2 (list + mapped)
Cardinality decided at	parse time	run time
Cardinality cap	none	max_map_length
Backfill stability	list drifts	frozen per DAG run
UI usability at N=5000	5000 graph nodes (unusable)	2 nodes + paginated grid

After the rollout, the DAG's parse latency drops from 60 seconds to under 1 second; the scheduler CPU drops from 100% to 5%; the S3 LIST bill drops from 600 000 calls per hour to 24 per day; and the DAG's structure no longer changes with the environment. The mapped-task cardinality is decided by whatever list_files_at_runtime returns on this particular run, and the max_map_length cap protects the scheduler from unbounded fan-out.

Output:

Metric	Before	After
DAG-parse latency	60 s	< 1 s
Scheduler CPU (parse)	100%	5%
S3 LIST calls / hour	600 000	1
Nodes in graph	5 000	2
Instances at run	5 000	5 000 (unchanged)
Cardinality cap	none	max_map_length=1024

Why this works — concept by concept:

• Parse-time vs run-time separation — the fundamental Airflow rule: DAG files parse cheaply and often; tasks execute expensively and once. Dynamic mapping is the primitive that lets you keep the parse cheap while still deciding the mapped cardinality per run.
• XComArg chaining — list_files_at_runtime() returns an XComArg (a lazy XCom reference), which .expand(key=...) unwraps at run time. The scheduler sees "this task's fan-out equals the length of the XCom" without materialising the value at parse.
• One node, N instances — the mapped task is one node in the DAG graph with a variable number of instances at run time. The graph view stays readable; the grid view shows the instances. This is the UX design that makes 5 000-way fan-out livable.
• Cardinality caps as a safety net — max_map_length (per task) and max_active_tasks_per_dag (per DAG) are the belt-and-braces protection against a runaway upstream. The scheduler refuses to materialise more instances than the cap allows and surfaces a clear AirflowException instead of OOMing.
• Cost — .expand() costs one XCom pull at run time plus O(N) scheduler bookkeeping to materialise the instances. The loop pattern costs O(N) at every parse, forever. The savings compound: parse frequency × instance count × deployment lifetime.

ETL
Topic — etl
ETL fan-out and parameterised-task problems

Practice →

Optimization Topic — optimization Optimization problems on scheduler cost and parse latency

Practice →

---

2. Basic expand + partial

.expand() fans over mapped args, .partial() pins fixed args — one call chain, one mapped node

The mental model in one line: .partial(fixed_kwarg=value) pins the arguments that are identical across all instances, and .expand(mapped_kwarg=iterable) spawns one mapped instance per element of the iterable, with map_index numbering them 0, 1, 2, ..., N-1. Every subsequent piece of airflow expand and airflow partial interview trivia flows from that one sentence — which arguments are fixed, which are mapped, how they combine, and how you know which instance you are inside the task body.

The four axes of expand + partial.

• Which kwargs are mapped vs fixed. Anything you pass to .partial() is the same for every instance; anything you pass to .expand() becomes a list-of-length-N and each instance sees the ith element. Positional arguments are not supported in .expand() — everything is keyword-only, which forces clarity at the call site.
• The order of the call chain. .partial(...).expand(...) is the canonical order; .expand(...).partial(...) is not supported. Mnemonic: "pin the fixed bits first, then fan out."
• map_index accounting. Every mapped instance carries a 0-based map_index. It is visible in the UI grid, in the log filename (map_index=N), in XCom keys (each instance has its own return XCom keyed by map_index), and inside the task body via context["ti"].map_index.
• What can be expanded. Any list-like at parse time (a Python literal) or any XComArg (an upstream task's output). You cannot .expand() over a generator (Airflow needs to know the length before materialising instances) or over a dict where the mapping semantics are ambiguous.

The three expand variants.

• .expand(kwarg=[a, b, c]). Fans out over a single kwarg with a literal list. Three instances; kwarg=a on map_index=0, kwarg=b on 1, kwarg=c on 2.
• .expand(kwarg=upstream.output). Fans out over the XComArg of an upstream task's return value. The cardinality equals the length of the returned list, decided at run time.
• .expand_kwargs([{...}, {...}]). Fans out over a list of kwarg dictionaries. Each dict is the full set of kwargs for one instance. Use when instances have heterogeneous parameters that do not decompose cleanly into "one list per kwarg."

The three partial variants.

• .partial(fixed_kwarg=value). Pins one kwarg to a constant for every instance. Repeat for every fixed kwarg (or pass them all at once).
• .partial(retries=3, execution_timeout=timedelta(minutes=10)). Pins task-level attributes (retries, timeout, pool) that apply to every mapped instance — the same attributes you would set on a non-mapped task.
• Nothing at all. If every argument is mapped, .partial() is optional — but conventionally people write process.partial().expand(...) even with no pinned args, to make the mapping intent explicit.

Common interview probes on expand + partial.

• "What's the difference between .expand() and .partial()?" — expand fans, partial pins; the answer is that specific.
• "Can I pass a generator to .expand()?" — no, Airflow needs the length at materialisation time.
• "How do I set retries on a mapped task?" — .partial(retries=3) — retries is a task attribute, pinned via partial.
• "What is map_index?" — the 0-based per-instance index visible in UI, logs, XCom keys, and ti.map_index.
• "What if .expand() gets an empty list?" — zero mapped instances; downstream tasks receive an empty list from XCom and typically become skipped.

Worked example — expand a list of S3 keys with a partial bucket

Detailed explanation. The canonical "mapped task processing one file per S3 key" pattern. The bucket is the same for every instance (.partial(bucket="raw-events")); the key is different per instance (.expand(key=[...])). One task node in the graph; N instances at run time. This is the "hello world" of dynamic task mapping and the pattern interviewers ask you to reproduce from memory.

• Fixed args. bucket="raw-events" — same for all instances.
• Mapped args. key — one instance per element of the list.
• map_index accounting. Instance 0 processes keys[0]; instance N-1 processes keys[N-1].

Question. Write a DAG that lists a static set of 4 S3 keys, fans out process_file, and reduces the results with a downstream count.

Input.

Parameter	Value
Bucket	raw-events (constant)
Keys	["k1", "k2", "k3", "k4"]
Fan-out	4 mapped instances
Reducer	receives list of 4 return values

Code.

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

@dag(
    dag_id="basic_expand_partial",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
)
def basic_expand_partial() -> None:

    @task
    def process_file(bucket: str, key: str) -> int:
        # Return the size of the processed object (mocked)
        print(f"processing {bucket}/{key}")
        return len(key)

    @task
    def reduce(sizes: list[int]) -> int:
        return sum(sizes)

    results = process_file.partial(bucket="raw-events").expand(
        key=["k1", "k2", "k3", "k4"]
    )
    reduce(results)

basic_expand_partial()

Step-by-step explanation.

1. process_file is a plain @task. Its parameters (bucket, key) look identical to a non-mapped task; the mapping is applied at the call site, not baked into the signature.
2. .partial(bucket="raw-events") returns a partial mapping object — Airflow's internal representation of "this call will supply bucket='raw-events' to every instance." No task is spawned yet; this is a builder pattern.
3. .expand(key=["k1", "k2", "k3", "k4"]) finalises the fan-out. The scheduler sees a mapped process_file node with map_length=4; at run time it materialises 4 instances with kwargs {bucket: "raw-events", key: "k1"} through {bucket: "raw-events", key: "k4"} and map_index 0 through 3.
4. The return value results is an XComArg that resolves at run time to the list [len("k1"), len("k2"), len("k3"), len("k4")] — Airflow assembles the list by pulling each mapped instance's return XCom in order of map_index.
5. reduce(results) runs once (non-mapped) with sizes=[2, 2, 2, 2] and returns 8. This is the map-reduce pattern — mapped fan-out followed by an assemble-and-summarise reducer.

Output.

Instance	map_index	bucket	key	return value
process_file[0]	0	raw-events	k1	2
process_file[1]	1	raw-events	k2	2
process_file[2]	2	raw-events	k3	2
process_file[3]	3	raw-events	k4	2
reduce	(not mapped)	—	—	8

Rule of thumb. Use .partial().expand() for the "one bucket, many keys" pattern — that is, whenever the fan-out unit is a single mapped kwarg and the rest of the arguments are constants. This is the pattern 80% of mapped-task use cases resolve to.

Worked example — inspecting map_index inside the task body

Detailed explanation. A mapped task often needs to know its own map_index — for per-instance file names, per-instance sub-logging, or targeted retry logic. The context["ti"].map_index attribute (or get_current_context()["ti"].map_index when the task is not decorated with **context) exposes the index. Walk through a version that uses map_index to derive per-instance temporary file paths.

• The task needs. Its own map_index at runtime.
• The access. context["ti"].map_index inside a **context-taking task, or get_current_context()["ti"].map_index inside a decorated task that does not accept context.
• The use case. Per-instance file naming, per-instance sub-log directory, per-instance temp scratch space.

Question. Modify the DAG so each mapped instance writes to a per-instance temp file /tmp/process_map_{map_index}.log.

Input.

Parameter	Value
Keys	["a", "b", "c"]
Temp file pattern	/tmp/process_map_{map_index}.log
Access pattern	context["ti"].map_index

Code.

from datetime import datetime
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context

@dag(
    dag_id="map_index_demo",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
)
def map_index_demo() -> None:

    @task
    def process_key(bucket: str, key: str) -> str:
        ctx = get_current_context()
        idx = ctx["ti"].map_index                   # 0-based per-instance index
        log_path = f"/tmp/process_map_{idx}.log"
        with open(log_path, "w") as f:
            f.write(f"instance {idx} processed {bucket}/{key}\n")
        return log_path

    logs = process_key.partial(bucket="raw").expand(key=["a", "b", "c"])
    print("mapped logs:", logs)

map_index_demo()

Step-by-step explanation.

1. get_current_context() returns the standard Airflow task context dictionary even when the task decorator did not accept **context in its signature. This is the idiomatic way to read context values from a decorated task.
2. ctx["ti"].map_index is the 0-based per-instance index. It is -1 for a non-mapped task (a sentinel meaning "not applicable") and 0..N-1 for mapped instances.
3. Each instance derives a distinct file path (/tmp/process_map_0.log, /tmp/process_map_1.log, /tmp/process_map_2.log). Because the paths are per-instance, there is no file collision even when instances run concurrently on the same worker.
4. The return value (the log path) is XComed back. logs is a list of three paths; the downstream can read them without knowing the mapped cardinality ahead of time.
5. The map_index=-1 sentinel is worth internalising: any downstream logic that needs to distinguish mapped from non-mapped tasks can check ti.map_index >= 0. This is the standard pattern in shared utility functions that must handle both cases.

Output.

map_index	key	log_path
0	a	/tmp/process_map_0.log
1	b	/tmp/process_map_1.log
2	c	/tmp/process_map_2.log

Rule of thumb. map_index is your per-instance identifier for anything that must not collide across concurrent runs — file paths, sub-log directories, per-instance retry markers, cache keys. Reach for get_current_context()["ti"].map_index any time the mapped task needs to know "which one am I."

Worked example — expand_kwargs for heterogeneous instance parameters

Detailed explanation. Sometimes each mapped instance needs its own dictionary of parameters — a full (region, bucket, prefix) triple per instance rather than one list per kwarg. expand_kwargs fans out over a list of kwarg dicts. Use it when the "fan-out unit" is a heterogeneous struct.

• The trigger. Each instance has 3+ parameters that vary together (region-specific bucket, region-specific prefix, etc.).
• The old workaround. Multiple parallel .expand(bucket=[...], prefix=[...]) lists — brittle because Airflow zips them together and any mismatch is a silent bug.
• The right answer. expand_kwargs([{region:"us", bucket:"us-events", prefix:"in/"}, ...]) — one dict per instance, all kwargs pinned atomically.

Question. Rewrite a DAG that processes region-specific config bundles using expand_kwargs.

Input.

Region	Bucket	Prefix
us	us-events	in/
eu	eu-events	data/
asia	asia-events	raw/

Code.

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

@dag(
    dag_id="expand_kwargs_demo",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
)
def expand_kwargs_demo() -> None:

    @task
    def process_region(region: str, bucket: str, prefix: str) -> str:
        return f"processed region={region} bucket={bucket} prefix={prefix}"

    configs = [
        {"region": "us",   "bucket": "us-events",   "prefix": "in/"},
        {"region": "eu",   "bucket": "eu-events",   "prefix": "data/"},
        {"region": "asia", "bucket": "asia-events", "prefix": "raw/"},
    ]
    process_region.expand_kwargs(configs)

expand_kwargs_demo()

Step-by-step explanation.

1. process_region takes three kwargs — region, bucket, prefix. Each mapped instance receives all three from a single kwarg dict.
2. configs is a Python list of three dicts. Each dict is a complete kwarg set — no zipping, no positional correspondence, no room for silent misalignment bugs.
3. .expand_kwargs(configs) fans out over the list of dicts. Instance 0 receives configs[0]; instance 1 receives configs[1]; and so on. map_length = len(configs) = 3.
4. Compare against the "parallel .expand()" alternative: process_region.expand(region=["us","eu","asia"], bucket=[...], prefix=[...]). Airflow would zip the three lists together, but if you accidentally reorder one list the zip silently produces the wrong pairing. expand_kwargs is atomically correct by construction.
5. expand_kwargs also accepts XComArg — pass an upstream task's output that returns a list of dicts, and the fan-out cardinality is decided at run time from whatever the upstream produced.

Output.

map_index	region	bucket	prefix	return
0	us	us-events	in/	processed region=us ...
1	eu	eu-events	data/	processed region=eu ...
2	asia	asia-events	raw/	processed region=asia ...

Rule of thumb. If instances have more than one varying kwarg, reach for expand_kwargs — even in the "hello world" case where you could write parallel .expand() lists. The atomic-dict pattern eliminates a whole class of silent zip-alignment bugs and reads more clearly at the call site.

Senior interview question on basic expand + partial

A senior interviewer might ask: "Write a mapped task that processes a static list of 100 S3 keys with a fixed bucket, retry count of 3, and per-instance temp file naming. Then show what changes if the list becomes an upstream task's output."

Solution Using partial for the fixed pieces and expand for the runtime list

from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context

@dag(
    dag_id="senior_expand_partial",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
)
def senior_expand_partial() -> None:

    @task
    def list_keys(bucket: str) -> list[str]:
        # In real life: boto3 list; here mocked
        return [f"k{i:03d}" for i in range(100)]

    @task(
        retries=3,
        retry_delay=timedelta(minutes=1),
        execution_timeout=timedelta(minutes=15),
    )
    def process_key(bucket: str, key: str) -> str:
        ctx = get_current_context()
        idx = ctx["ti"].map_index
        tmp = f"/tmp/proc_{idx:04d}.log"
        with open(tmp, "w") as f:
            f.write(f"instance {idx} bucket={bucket} key={key}\n")
        return tmp

    # Variant 1 — static list of 100 keys
    static_logs = process_key.partial(bucket="raw-events").expand(
        key=[f"k{i:03d}" for i in range(100)]
    )

    # Variant 2 — dynamic list from an upstream task
    dyn_keys = list_keys(bucket="raw-events")
    dyn_logs = process_key.partial(bucket="raw-events").expand(key=dyn_keys)

    print("static logs:", static_logs)
    print("dynamic logs:", dyn_logs)

senior_expand_partial()

Step-by-step trace.

Piece	Where it lives	Behaviour
Fixed bucket	.partial(bucket="raw-events")	Same for all 100 instances
Fixed retries=3	task decorator	Applied per-instance
Fixed execution_timeout	task decorator	Applied per-instance
Mapped key (static)	.expand(key=[...]) literal	100 instances at parse time
Mapped key (dynamic)	.expand(key=dyn_keys) XComArg	100 instances at runtime, count from upstream
map_index	ctx["ti"].map_index	0..99 per instance

After the rollout, both variants produce the same 100-instance fan-out; the difference is when the cardinality is fixed. Variant 1's cardinality is 100 at parse time (change the code to change it). Variant 2's cardinality equals whatever list_keys returned on this DAG run — could be 0, 50, or 100 000 depending on the environment. Both keep the DAG graph at two nodes; both surface 100 rows in the grid view; both write per-instance /tmp/proc_NNNN.log files without collision.

Output:

Concern	Variant 1 (static)	Variant 2 (dynamic)
Cardinality decided at	parse time	run time
Change cardinality	edit DAG file	change upstream data
Retries per instance	3	3
Execution timeout	15 min	15 min
Backfill stability	list frozen in code	list frozen per DAG run
Preferred?	rarely	almost always

Why this works — concept by concept:

• Partial for fixed, expand for mapped — the split reads at the call site. .partial() collects the constants once; .expand() fans over the varying kwarg. This is the pattern interviewers want to see written from memory.
• Task-level attributes via partial — retries, retry_delay, execution_timeout, pool, and every other task-level setting are set on the decorator. They apply per-instance automatically. Never try to override them in .expand() — that path is not supported.
• map_index as a per-instance identifier — reach for get_current_context()["ti"].map_index any time the task needs to know its instance number. The 4-digit-zero-padded format f"{idx:04d}" sorts correctly in file listings.
• Static vs dynamic — the syntax difference is one word ([...] literal vs upstream_task.output), but the semantic difference is enormous. Prefer dynamic unless you have a genuine reason for parse-time cardinality (say, the four quarters of the year).
• Cost — .partial().expand() costs O(1) at parse time and O(N) at run time to materialise instances. Task-level attributes (retries etc.) are copied to each instance; the per-instance retry state is stored in the metadata DB with map_index as a discriminator.

ETL
Topic — etl
ETL mapped-task and parameterised-processing problems

Practice →

Python Topic — list-comprehension Python list-comprehension problems for kwarg generation

Practice →

---

3. Upstream fan-in — mapping from a task's output

.expand(kwarg=upstream.output) is the whole point — a list-returning task becomes downstream fan-out

The mental model in one line: any Airflow task whose return value is a list becomes a natural source of downstream fan-out — the downstream declares .expand(kwarg=upstream.output) and Airflow generates one mapped instance per element of the list at run time. This is the "map" in map-reduce, and it is the pattern that unlocks airflow map reduce workflows over dynamically-sized inputs — the exact scenario every real analytics DAG faces.

The four axes of upstream fan-in.

• XComArg chaining. upstream_task() returns an XComArg — a promise object that says "the value of this task's return XCom." Passing that object to .expand(kwarg=...) tells Airflow to pull the XCom at run time and fan out over its length. No manual xcom_pull required.
• Cardinality decided at run time. The mapped instance count equals the length of the returned list. If the upstream returns [a, b, c] you get 3 instances; if it returns [] you get 0; if it returns a 10 000-element list you get 10 000 instances (capped by max_map_length).
• The reducer pattern. A downstream non-mapped task that receives mapped_task.output as a list — Airflow assembles the return XComs of every mapped instance into a Python list in map_index order. This is the "reduce" in map-reduce.
• The empty-list case. When the upstream returns [], the mapped task materialises with map_length = 0. Downstream tasks receive an empty list and typically transition to skipped — a clean semantic that means "there was nothing to do."

The parallelism cap knobs.

• max_active_tasks_per_dag. DAG-level backpressure. If the mapped fan-out is 500 but max_active_tasks_per_dag = 32, only 32 mapped instances run concurrently; the rest queue and drain as capacity frees up.
• max_map_length. Per-task cardinality cap. Default 1024 in Airflow 2.x, configurable up (AIRFLOW__CORE__MAX_MAP_LENGTH). If the upstream returns more than the cap, the DAG fails with a clear AirflowException — a fail-fast safety net.
• max_active_tis_per_dag. Per-mapped-task concurrency limit. Set on the mapped task itself to throttle a specific task's concurrency without touching the DAG-wide setting.
• Pools. The classic Airflow pool mechanism still applies: task.partial(pool="external_api").expand(...) restricts every mapped instance to consume from the named pool, providing a shared concurrency budget across DAGs.

When to reach for map-reduce.

• File-per-key ETL. Upstream lists a bucket; each mapped instance processes one file; downstream sums.
• Region-per-partition analytics. Upstream fetches active regions; each mapped instance runs a per-region query; downstream unions results.
• Fan-out-fan-in RPC. Upstream produces a work list; each mapped instance calls an external service; downstream aggregates responses.
• Not for streaming. If the work list is genuinely unbounded, Airflow is the wrong tool — reach for a stream processor (Kafka + Flink / Spark Streaming) instead.

Common interview probes on upstream fan-in.

• "Show me the DAG for list_files → process_file → aggregate." — every senior interview asks this shape.
• "What happens if list_files returns an empty list?" — 0 mapped instances; downstream tasks skip.
• "How do you cap the fan-out?" — max_map_length per task; max_active_tasks_per_dag per DAG.
• "What does the reducer's input look like?" — a Python list of the mapped instances' return values, in map_index order.
• "Does .expand() wait for the upstream to finish?" — yes; the mapped instances are materialised after the upstream succeeds.

Worked example — list_files → process_file.expand() → aggregate

Detailed explanation. The canonical map-reduce pattern in Airflow. An upstream list_files task returns a list of S3 keys; a mapped process_file runs one instance per key; a downstream aggregate collects the return values and produces a summary. This is the pattern every senior interview asks you to draw on the whiteboard.

• List step. list_files returns list[str].
• Map step. process_file.expand(key=list_files()) fans out.
• Reduce step. aggregate(process_file.output) receives the list of return values.

Question. Implement the full map-reduce DAG. Show what the grid view looks like when 5 files are listed and one instance fails.

Input.

Task	Type	Cardinality
list_files	@task	1 (returns list of 5)
process_file	mapped	5 instances
aggregate	@task	1 (reducer)

Code.

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

@dag(
    dag_id="map_reduce_files",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
)
def map_reduce_files() -> None:

    @task
    def list_files(bucket: str) -> list[str]:
        return [f"k{i}" for i in range(5)]

    @task
    def process_file(bucket: str, key: str) -> int:
        # Return per-key byte count (mocked)
        if key == "k2":
            raise RuntimeError("simulated failure for k2")
        return 1000 + len(key)

    @task
    def aggregate(sizes: list[int]) -> int:
        return sum(sizes)

    keys = list_files(bucket="raw-events")
    per_key = process_file.partial(bucket="raw-events").expand(key=keys)
    total = aggregate(per_key)

map_reduce_files()

Step-by-step explanation.

1. list_files returns ["k0", "k1", "k2", "k3", "k4"]. Its return value is an XComArg — a lazy reference. The scheduler sees "the fan-out of the next task is the length of this XCom."
2. .expand(key=keys) materialises 5 mapped process_file instances at run time, once list_files succeeds. The instances have map_index 0 through 4 and receive key="k0" through key="k4" respectively.
3. Instance map_index=2 (processing key="k2") raises RuntimeError. Airflow marks that instance as failed but other mapped instances continue — the mapped-task failure model is per-instance, not per-node.
4. When the mapped node's instances have all completed (successfully or otherwise), Airflow evaluates the downstream aggregate's trigger rule. The default all_success means aggregate is upstream_failed because one mapped instance failed.
5. If you want the reducer to run when some mapped instances succeed, set aggregate = aggregate.override(trigger_rule="none_failed_min_one_success") — Airflow's mapped-task-aware trigger rules that reduce on the survivors.

Output.

UI view	What appears
Graph view	list_files → process_file → aggregate (3 nodes; process_file badged "5")
Grid view	5 rows for process_file: [OK, OK, FAILED, OK, OK]
aggregate state	upstream_failed (default trigger_rule=all_success)
Per-instance logs	map_index=2 log shows RuntimeError traceback
Rerun a single instance	UI supports clicking one instance in the grid to rerun just that

Rule of thumb. Map-reduce is the pattern; per-instance failure is the reality. Choose the reducer's trigger_rule deliberately — all_success for "abort if any mapped instance failed"; none_failed_min_one_success for "reduce over the survivors"; all_done for "always run the reducer, even if all failed."

Worked example — capping fan-out with max_map_length

Detailed explanation. The upstream task might, in production, return a list of 100 000 elements — either because a bucket exploded or because the upstream had a bug. Without a cap, the scheduler tries to materialise 100 000 mapped instances, blows through worker memory, and takes the whole cluster down. max_map_length is the safety net that turns this into a clean, fast failure.

• The cap. Default 1024 in Airflow 2.x; configurable via AIRFLOW__CORE__MAX_MAP_LENGTH.
• The failure mode with cap. DAG fails with AirflowException: max_map_length exceeded — clear message, no partial fan-out.
• The failure mode without cap. Scheduler OOMs; workers churn; recovery takes hours.

Question. Configure max_map_length = 1024 in the DAG and show what happens when the upstream returns 5 000 keys.

Input.

Setting	Value
max_map_length	1024 (Airflow default)
Upstream returned list length	5000
Expected behaviour	fail fast with clear error

Code.

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

@dag(
    dag_id="capped_fanout",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
    max_active_tasks=32,                            # DAG-wide backpressure
)
def capped_fanout() -> None:

    @task
    def list_keys() -> list[str]:
        return [f"k{i}" for i in range(5000)]       # oversized on purpose

    @task
    def guard(keys: list[str], cap: int = 1024) -> list[str]:
        if len(keys) > cap:
            raise AirflowException(
                f"list_keys returned {len(keys)} keys; refusing to fan out beyond cap={cap}"
            )
        return keys

    @task(max_active_tis_per_dag=8)                 # per-task concurrency cap
    def process_key(key: str) -> None:
        pass

    raw = list_keys()
    safe = guard(raw, cap=1024)
    process_key.expand(key=safe)

capped_fanout()

# airflow.cfg — the global cap
[core]
max_map_length = 1024

Step-by-step explanation.

1. list_keys returns 5 000 keys — the oversize scenario. Without any guard, the downstream .expand() would attempt to materialise 5 000 instances.
2. max_map_length = 1024 in airflow.cfg is the hard, global cap. Airflow would refuse the fan-out with AirflowException: max_map_length exceeded. That is the belt.
3. The guard task is the braces — a defensive check in the DAG itself that raises a readable exception naming the actual list length. The exception message includes the number of keys returned, which pinpoints the offending upstream instantly.
4. max_active_tasks=32 at the DAG level and max_active_tis_per_dag=8 on the mapped task apply backpressure — they do not reject the fan-out; they just serialise the work. Useful for controlling external-service concurrency without failing the run.
5. The senior signal is naming both mechanisms — the cap (fail-fast) and the backpressure (throttle) — as two orthogonal safety nets. Junior answers name only one; senior answers name both and explain when each applies.

Output.

Mechanism	Value	Behaviour on 5000-key list
max_map_length (cap)	1024	AirflowException; DAG fails
guard task (defensive)	cap=1024	AirflowException with readable message
max_active_tasks (backpressure)	32	(would not activate; run fails first)
max_active_tis_per_dag (per-task)	8	(would not activate; run fails first)

Rule of thumb. Set max_map_length in airflow.cfg and additionally add an in-DAG guard task with a readable exception. The .cfg cap protects against forgetting; the guard task turns "opaque AirflowException" into "here is the exact number the upstream returned, go debug it."

Worked example — reducer with the right trigger rule

Detailed explanation. After a mapped fan-out, the reducer often needs to run even if some mapped instances failed — say, an aggregation over successful shards, ignoring the shards that timed out. The default trigger_rule = all_success blocks the reducer if any mapped instance failed. Override with the mapped-task-aware trigger rules to reduce over the survivors.

• all_success (default). Reducer runs only if every mapped instance succeeded.
• none_failed_min_one_success. Reducer runs if at least one succeeded and none failed (skipped is OK).
• all_done. Reducer runs when every mapped instance is in a terminal state (success, failed, or skipped).

Question. Implement a reducer that runs over the surviving mapped instances even if a minority failed. Show what changes with each trigger rule.

Input.

Trigger rule	When reducer runs
all_success	0 mapped instance failures
none_failed_min_one_success	≥1 success; 0 failures (skipped OK)
all_done	all instances reached terminal state (any outcome)

Code.

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

@dag(
    dag_id="reducer_trigger_rules",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
)
def reducer_trigger_rules() -> None:

    @task
    def list_shards() -> list[int]:
        return [0, 1, 2, 3, 4]

    @task
    def process_shard(shard: int) -> int:
        if shard == 2:
            raise RuntimeError("simulated failure for shard 2")
        return shard * 10

    @task(trigger_rule="all_done")
    def aggregate_over_survivors(results: list[int]) -> dict:
        # Successful returns are ints; failed instances contribute nothing to the list
        # NOTE: Airflow 2.9+ mapped-task collection excludes failed instances by default
        return {"survivors": len(results), "sum": sum(results)}

    shards = list_shards()
    per_shard = process_shard.expand(shard=shards)
    aggregate_over_survivors(per_shard)

reducer_trigger_rules()

Step-by-step explanation.

1. list_shards returns [0, 1, 2, 3, 4] — 5 shards. process_shard.expand(shard=shards) fans out 5 mapped instances.
2. Instance map_index=2 fails (shard == 2); the other four succeed and return 0, 10, 30, 40 respectively.
3. Default trigger_rule="all_success" would leave aggregate_over_survivors in upstream_failed. Overriding to trigger_rule="all_done" means the reducer runs once every mapped instance has finished, regardless of outcome.
4. When Airflow collects the mapped return values for the reducer, instances that failed contribute nothing to the collected list — the reducer receives [0, 10, 30, 40], four entries, in map_index order minus the failure.
5. aggregate_over_survivors returns {"survivors": 4, "sum": 80} — a summary over the surviving shards. This is the pattern for degradation-tolerant aggregations: the reducer runs; the metadata reflects "we saved what we could."

Output.

Scenario	Trigger rule	Reducer runs?	Reducer input
All 5 succeed	all_success (default)	yes	[0, 10, 20, 30, 40]
1 fails, 4 succeed	all_success (default)	no (upstream_failed)	—
1 fails, 4 succeed	all_done	yes	[0, 10, 30, 40]
1 fails, 4 succeed	none_failed_min_one_success	no (there is a failure)	—

Rule of thumb. Pick the trigger rule based on the business meaning of the reducer's output. "All shards must succeed" → all_success. "Reduce over whichever succeeded" → all_done. "Reduce as long as at least one succeeded and no failures" → none_failed_min_one_success. Never default; the trigger rule is a semantic decision, not a footnote.

Senior interview question on upstream fan-in

A senior interviewer might ask: "Design a DAG that lists an unknown number of files, processes each in parallel with a mapped task, aggregates the results, and handles the case where 10% of instances fail transiently. What caps do you set, what trigger rule do you pick, and what does the runbook look like when the reducer fires with only 90% of the data?"

Solution Using guarded fan-in with degradation-tolerant reducer

from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.exceptions import AirflowException

@dag(
    dag_id="senior_upstream_fan_in",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
    max_active_tasks=64,                            # DAG-level backpressure
)
def senior_upstream_fan_in() -> None:

    @task
    def list_files(bucket: str) -> list[str]:
        return list_s3_keys(bucket)                 # imagined helper

    @task
    def guard_cardinality(keys: list[str], cap: int = 1024) -> list[str]:
        if len(keys) == 0:
            raise AirflowException("list_files returned no keys — refusing to skip silently")
        if len(keys) > cap:
            raise AirflowException(
                f"list_files returned {len(keys)} keys > cap {cap} — investigate upstream"
            )
        return keys

    @task(
        retries=2,
        retry_delay=timedelta(seconds=30),
        max_active_tis_per_dag=16,
    )
    def process_key(bucket: str, key: str) -> dict:
        # per-key handling; can raise on transient errors
        return {"key": key, "bytes": handle(bucket, key)}

    @task(trigger_rule="all_done")
    def aggregate(results: list[dict]) -> dict:
        return {
            "survivors": len(results),
            "total_bytes": sum(r["bytes"] for r in results),
        }

    @task
    def audit(summary: dict, expected: int) -> None:
        pct = 100.0 * summary["survivors"] / expected
        if pct < 90:
            raise AirflowException(f"only {pct:.1f}% of mapped instances succeeded — page on-call")

    keys = list_files(bucket="raw-events")
    safe_keys = guard_cardinality(keys, cap=1024)
    per_key = process_key.partial(bucket="raw-events").expand(key=safe_keys)
    summary = aggregate(per_key)
    audit(summary, expected=len(safe_keys))

senior_upstream_fan_in()

Step-by-step trace.

Task	Role	Behaviour
list_files	source	returns list of keys at runtime
guard_cardinality	defensive guard	fails-fast on empty or oversize list
process_key	mapped fan-out	2 retries, 16-concurrency cap
aggregate	reducer	trigger_rule=all_done; runs over survivors
audit	quality gate	fails if survivor% < 90

After the rollout, the DAG survives transient failures (retries), fails-fast on structural issues (empty/oversize list), reduces over survivors when the failure rate is tolerable (< 10%), and pages the on-call when data quality degrades beyond threshold (audit). The runbook when the audit fires: (a) check which mapped instances failed by clicking through the grid view; (b) look at the per-instance logs for the failure root cause; (c) if transient, clear the failed instances and rerun; (d) if systemic, investigate upstream.

Output:

Scenario	Behaviour
All 100 succeed	aggregate over 100; audit passes; DAG succeeds
5 transient failures, retries succeed	aggregate over 100; audit passes; DAG succeeds
5 hard failures	aggregate over 95; audit passes (95% > 90%); DAG succeeds with warning
20 hard failures	aggregate over 80; audit fails; on-call paged
0 keys returned	guard_cardinality fails; DAG fails immediately
2000 keys returned	guard_cardinality fails; DAG fails immediately

Why this works — concept by concept:

• Guarded upstream fan-in — the guard task turns opaque failures ("something went wrong at fan-out") into readable exceptions ("upstream returned 0 keys" or "upstream returned 2000 keys > cap"). The debug time savings compound across incidents.
• Per-instance retries — retries=2 on the mapped task applies to each mapped instance independently. A transient failure on map_index=17 retries only that instance, not the whole node.
• Concurrency backpressure — max_active_tis_per_dag=16 on the mapped task caps concurrent instances at 16, protecting downstream services from a thundering herd. The remaining instances queue and drain naturally.
• Reducer over survivors — trigger_rule="all_done" lets the reducer run even when some instances failed. The mapped-task collection excludes failed instances from the reducer's input list — the reducer sees only the survivors.
• Data-quality audit — the audit task turns "reduce over survivors" into "reduce over survivors only if the survivor% is high enough." This is the pattern that keeps degradation-tolerant DAGs honest — quality gates catch what per-instance retries cannot.
• Cost — one guard task (~1 second), N mapped instances at 16 concurrency, one reducer, one audit. Total scheduler load is O(N) at run time; parse cost stays O(1). The cost of NOT having the guard + audit is one 3 AM incident per year.

ETL
Topic — etl
ETL fan-in and map-reduce problems

Practice →

Optimization Topic — optimization Optimization problems on cardinality caps and backpressure

Practice →

---

4. Nested expand + TaskGroup mapping

Airflow 2.7+ nests .expand() calls, 2.9+ maps whole TaskGroups — multiplicative fan-out is a foot-gun

The mental model in one line: nested .expand() (Airflow 2.7+) lets one mapped task's output feed another .expand(), multiplying cardinality; TaskGroup mapping (Airflow 2.9+) treats a whole subgraph as the "unit" being mapped, spawning a full sub-DAG per outer element. Both features unlock cleanly parameterised multi-dimensional workflows — and both make it terrifyingly easy to detonate the scheduler if you forget the cardinality-multiplication math.

The three axes of nested and TaskGroup mapping.

• Nested .expand(). A task that runs downstream of a mapped task, itself declared .expand(kwarg=mapped_task.output), fans out its own instances per element of every mapped output. Total cardinality = outer × inner.
• TaskGroup mapping. Introduced in 2.9. A @task_group decorated function accepts arguments and can itself be mapped via .expand(arg=...). Each mapped invocation of the group creates a full sub-DAG of its constituent tasks.
• Cardinality arithmetic. Cardinality multiplies across every mapping layer. Outer 10 × inner 20 = 200 instances. Two layers of 100 each = 10 000. The senior habit is doing this math before shipping.

When nested expand is the right tool.

• Regions × files. One region-list; per-region file-lists; process every (region, file) pair. The 2-D fan-out matches the data shape.
• Tenants × queries. Per-tenant list of analytics queries; one instance per (tenant, query). Same 2-D shape.
• Buckets × prefixes. Multi-bucket cross-partition ETL. Same shape again.

When TaskGroup mapping wins over nested expand.

• Whole subgraph per outer element. If each outer element needs multiple tasks (list → process → validate → publish), a TaskGroup maps the whole subgraph. Nested expand can only map a single task.
• Shared state within the group. All tasks in one mapped group instance share map_index — the group is the atomic unit of mapping.
• UI clarity. A mapped TaskGroup collapses into a single visible node in the graph view with a nested grid; nested expand of individual tasks renders as multiple mapped nodes side by side.

The explosion trap — cardinality arithmetic.

• Outer × inner = total. 100 regions × 100 files per region = 10 000 instances. The scheduler handles this only if you have raised max_map_length above 1024 and max_active_tasks_per_dag above 32.
• Three-layer nesting is a footgun. 100 × 100 × 100 = 1 000 000. Airflow technically supports it; your scheduler does not.
• The default cap. max_map_length = 1024 per task. A nested expand that would produce > 1024 mapped instances at any level fails fast with a clear exception.
• The senior habit. Compute total cardinality before writing the code. If it exceeds ~10 000, redesign — either flatten into a single-level expand_kwargs over pre-computed pairs or move to a batch-processing tool (Spark) that handles high cardinality more gracefully.

Common interview probes on nested + TaskGroup mapping.

• "Can you .expand() nested?" — yes since 2.7; explain the multiplicative cardinality.
• "Can you map a TaskGroup?" — yes since 2.9; a whole subgraph fans out.
• "What is the risk with nested expand?" — cardinality explosion; do the arithmetic first.
• "How do you cap nested expand?" — max_map_length still applies per mapped task; the total is capped implicitly at outer × inner.

Worked example — nested expand across regions × files

Detailed explanation. The canonical 2-D fan-out pattern. One upstream lists regions; a mapped list_files_per_region runs one instance per region, each returning a list of file keys; a nested process_file .expand()-es over the flattened set of (region, key) pairs to spawn one instance per file. This is the 2-D map-reduce case.

• Outer. Regions list (3 entries).
• Middle. Per-region file list (4 entries per region).
• Inner. process_file instances (3 × 4 = 12 total).

Question. Write the nested-expand DAG. Show the total instance count and how you would enforce a total cap.

Input.

Layer	Cardinality
Regions	3
Files per region	4
Total process_file instances	12
Total scheduler load	3 + 3 + 12 = 18 mapped instances

Code.

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

@dag(
    dag_id="nested_expand_2d",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    max_active_tasks=32,
)
def nested_expand_2d() -> None:

    @task
    def list_regions() -> list[str]:
        return ["us", "eu", "asia"]

    @task
    def list_files_for_region(region: str) -> list[dict]:
        # Each element is a dict for expand_kwargs downstream
        return [{"region": region, "key": f"{region}/f{i}"} for i in range(4)]

    @task
    def flatten(nested: list[list[dict]], cap: int = 1024) -> list[dict]:
        # Airflow gives us the outer list; flatten to (region, key) pairs
        flat = [item for sub in nested for item in sub]
        if len(flat) > cap:
            raise AirflowException(f"flat cardinality {len(flat)} > cap {cap}")
        return flat

    @task
    def process_file(region: str, key: str) -> str:
        return f"{region}/{key}:done"

    regions = list_regions()
    nested = list_files_for_region.expand(region=regions)
    flat = flatten(nested, cap=1024)
    process_file.expand_kwargs(flat)

nested_expand_2d()

Step-by-step explanation.

1. list_regions returns 3 region strings. list_files_for_region.expand(region=regions) fans out 3 mapped instances of the middle task, each returning a list of 4 dicts for its region — nested cardinality (list of lists).
2. flatten collapses the list of lists into a flat list of {region, key} dicts. This is the crucial rewrite step: nested expand of a single task is fine, but flattening to expand_kwargs reads more clearly than chaining two .expand() calls.
3. flatten also enforces the total cap. The nested arithmetic (3 × 4 = 12) is fine; if either dimension inflated (30 regions × 40 files = 1200), the guard fails-fast with a readable exception.
4. process_file.expand_kwargs(flat) fans out 12 instances of process_file — one per (region, key) pair. Each instance receives both the region and the key as separate kwargs, no re-parsing required.
5. The DAG graph shows 4 nodes: list_regions → list_files_for_region (mapped, 3) → flatten → process_file (mapped, 12). The grid view splits by node; each mapped node's instances are individually visible and rerunnable.

Output.

Layer	Instances	map_index range
list_regions	1	(not mapped)
list_files_for_region	3	0, 1, 2
flatten	1	(not mapped)
process_file	12	0..11
Total	17	—

Rule of thumb. For 2-D fan-out, prefer flatten-then-expand_kwargs over chained .expand() calls. The flatten step is one extra task and one obvious cap-check; it reads more clearly, backfills more predictably, and gives you one obvious place to enforce the total cardinality cap.

Worked example — mapping a TaskGroup across tenants

Detailed explanation. Airflow 2.9+ can map whole TaskGroups. Each mapped invocation of the group creates a full sub-DAG of its constituent tasks — say, fetch → process → validate → publish. This is the right pattern when each outer element needs a multi-task subgraph, not just one task.

• Outer. Tenants list (5 tenants).
• Inner. A 4-task subgraph per tenant: fetch → process → validate → publish.
• Total task instances. 5 tenants × 4 tasks = 20 task instances, organised as 5 mapped groups.

Question. Implement a per-tenant TaskGroup mapped across the tenant list. Show the grid view structure.

Input.

Tenant	Subgraph tasks
t1	fetch → process → validate → publish
t2	fetch → process → validate → publish
t3	fetch → process → validate → publish
t4	fetch → process → validate → publish
t5	fetch → process → validate → publish

Code.

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

@dag(
    dag_id="mapped_taskgroup",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    max_active_tasks=16,
)
def mapped_taskgroup() -> None:

    @task
    def list_tenants() -> list[str]:
        return [f"t{i}" for i in range(1, 6)]

    @task_group(group_id="per_tenant")
    def per_tenant(tenant: str) -> None:

        @task
        def fetch(t: str) -> dict:
            return {"tenant": t, "raw": f"raw-for-{t}"}

        @task
        def process(row: dict) -> dict:
            row["processed"] = True
            return row

        @task
        def validate(row: dict) -> dict:
            if not row.get("processed"):
                raise ValueError("not processed")
            return row

        @task
        def publish(row: dict) -> None:
            print(f"published {row['tenant']}")

        publish(validate(process(fetch(tenant))))

    tenants = list_tenants()
    per_tenant.expand(tenant=tenants)

mapped_taskgroup()

Step-by-step explanation.

1. The @task_group decorator wraps the per-tenant subgraph — four tasks that chain fetch → process → validate → publish. The group accepts a tenant parameter and passes it into the first task.
2. list_tenants returns 5 tenant names. per_tenant.expand(tenant=tenants) maps the whole group — 5 mapped invocations, each running the 4-task subgraph with a different tenant argument.
3. Total task instances: 5 tenants × 4 tasks = 20. In the DAG graph, this appears as a single collapsed per_tenant node badged "5"; expanding it shows the 4-task subgraph; each subgraph position has 5 mapped instances (one per tenant).
4. Failure isolation is per-tenant-per-task. If validate fails for tenant t3, the publish for t3 is upstream_failed but the other four tenants continue independently. This is the correctness win over hand-written parallel DAGs.
5. max_active_tasks=16 at the DAG level caps concurrent running task instances across all mapped groups. With 20 total tasks and cap 16, up to 16 run concurrently; the rest queue and drain.

Output.

Grid view row	Instances	Behaviour
per_tenant.fetch	5 (t1..t5)	independent per-tenant
per_tenant.process	5	after per-tenant fetch
per_tenant.validate	5	after per-tenant process
per_tenant.publish	5	after per-tenant validate
Total	20	5 tenants × 4 tasks

Rule of thumb. Reach for @task_group + .expand() when the fan-out unit is a subgraph, not just one task. If your per-outer-element work is one task, use plain .expand(). If it is three or more tasks that chain together, mapped TaskGroups are dramatically cleaner than chained .expand() calls.

Worked example — the cardinality explosion trap

Detailed explanation. A team ships a nested-expand DAG that runs fine in dev with small data. In prod, the outer list grows to 200 and the inner list grows to 100, producing 20 000 process instances. The DAG hits max_map_length and fails fast — good, but the team needs to redesign. Walk through the arithmetic and the two escape hatches.

• The trap. Outer × inner grows non-linearly with production data.
• The cap. max_map_length = 1024 per task fails at 20 000.
• Escape 1. Flatten + cap check with actionable error.
• Escape 2. Batch inside the mapped task (each instance handles K units of work).

Question. A DAG needs to process 200 regions × 100 files per region = 20 000 file operations. Design two variants: one that fans out to 20 000 instances (with proper caps), and one that batches to 200 instances processing 100 files each.

Input.

Design	Instance count	Per-instance work
Full fan-out	20 000	1 file
Batched by region	200	100 files
Trade-off	max parallelism vs scheduler cost

Code.

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

@dag(
    dag_id="explosion_escape",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
    max_active_tasks=128,
)
def explosion_escape() -> None:

    @task
    def list_regions() -> list[str]:
        return [f"r{i:03d}" for i in range(200)]

    @task
    def list_files_for_region(region: str) -> list[str]:
        return [f"{region}/f{i:03d}" for i in range(100)]

    # Variant A — full fan-out to 20000 (requires max_map_length >> 20000)
    @task
    def flatten_all(nested: list[list[str]]) -> list[dict]:
        flat = [{"key": k} for sub in nested for k in sub]
        return flat

    @task(max_active_tis_per_dag=32)
    def process_one_file(key: str) -> None:
        pass

    # Variant B — batched: one instance per region, processes all files internally
    @task(max_active_tis_per_dag=32)
    def process_region_batch(region: str, keys: list[str]) -> int:
        for k in keys:
            handle(k)
        return len(keys)

    regions = list_regions()
    nested = list_files_for_region.expand(region=regions)

    # Variant A
    # flat = flatten_all(nested)
    # process_one_file.expand_kwargs(flat)

    # Variant B — recommended for high cardinality
    process_region_batch.expand_kwargs(
        [{"region": r, "keys": ks} for r, ks in zip(regions, nested)]
    )

explosion_escape()

Step-by-step explanation.

1. Variant A fans out to 20 000 instances. Requires max_map_length = 25000 (or higher) in airflow.cfg. The scheduler must materialise 20 000 task instances into the metadata DB — measurable overhead, and the grid view becomes hard to navigate.
2. Variant B collapses each region's 100 files into a single mapped instance that loops internally over the keys. 200 instances total; each does 100 units of work. The scheduler load drops 100×.
3. The trade-off: Variant A gives the maximum parallelism (each file is a separate task instance, retryable independently). Variant B trades granular retryability for scheduler efficiency; if one file within a batch fails, the whole batch fails and retries all 100.
4. The senior heuristic: if per-file processing is fast (< 10 s) and cardinality is high (> 5000), batch. If per-file processing is slow (minutes) or per-file retry matters (each file is a customer's data), fan out.
5. The max_active_tis_per_dag=32 cap on either variant limits concurrent mapped instances — a per-task safety net that keeps external services (S3, downstream APIs) from being hammered by 20 000 concurrent calls.

Output.

Metric	Variant A (full fan-out)	Variant B (batched)
Task instances	20 000	200
Scheduler load	20 000 rows in TaskInstance	200 rows
Grid view rows	20 000	200
Per-file retry	independent	batch-level only
Concurrent processing	up to max_active_tis_per_dag	up to max_active_tis_per_dag × 100 files
Recommended for	slow per-file work, per-file retry matters	fast per-file work, high cardinality

Rule of thumb. Multiply the cardinality before you write the DAG. If total instance count > 10 000, batch inside the mapped task rather than fanning out further. If per-instance work is fast enough that scheduler bookkeeping dominates the run time, batching is a straight efficiency win.

Senior interview question on nested + TaskGroup mapping

A senior interviewer might ask: "You have a workflow that processes 50 tenants, and per tenant does a 3-step subgraph of fetch → transform → publish, and the fetch step itself returns a list of shards that need per-shard processing. Design the DAG. What's the total instance count, what mapping primitives do you use, and how do you keep the cardinality manageable?"

Solution Using TaskGroup mapping for the outer dimension and expand_kwargs for the inner

from datetime import datetime
from airflow.decorators import dag, task, task_group
from airflow.exceptions import AirflowException

@dag(
    dag_id="senior_nested_taskgroup",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    max_active_tasks=64,
)
def senior_nested_taskgroup() -> None:

    @task
    def list_tenants() -> list[str]:
        return [f"t{i:02d}" for i in range(50)]     # 50 tenants

    @task_group(group_id="per_tenant")
    def per_tenant(tenant: str) -> None:

        @task
        def fetch(t: str) -> list[dict]:
            # Returns per-shard configs — cardinality ~10 per tenant
            return [{"tenant": t, "shard": i} for i in range(10)]

        @task(max_active_tis_per_dag=8, retries=2)
        def transform_shard(tenant: str, shard: int) -> dict:
            return {"tenant": tenant, "shard": shard, "rows": 1000}

        @task
        def publish(processed: list[dict], t: str) -> None:
            print(f"tenant {t} published {len(processed)} shards")

        shards = fetch(tenant)
        processed = transform_shard.expand_kwargs(shards)
        publish(processed, t=tenant)

    tenants = list_tenants()
    per_tenant.expand(tenant=tenants)

senior_nested_taskgroup()

Step-by-step trace.

Layer	Cardinality	Mapping primitive
list_tenants	1	—
per_tenant TaskGroup	50	.expand(tenant=tenants)
fetch (per tenant)	1 per group	inside group
transform_shard (per tenant × per shard)	10 per group	.expand_kwargs inside group
publish (per tenant)	1 per group	inside group
Total task instances	50 tenants × (1 + 10 + 1) = 600	—

After the rollout, the DAG runs 50 tenant sub-DAGs in parallel (capped by max_active_tasks=64), each running the fetch → transform_shard × 10 → publish chain independently. Total task instances is 600 — well below the max_map_length cap and well within scheduler comfort. Failure isolation is per-tenant: if tenant t17's transform_shard fails, only tenant t17's publish is blocked; the other 49 continue.

Output:

Metric	Value
Tenants	50
Shards per tenant	10
Total task instances	600
Grid rows	50 rows per subgroup node
Failure blast radius	one tenant's shards
max_map_length OK?	yes (max 50 or 10 per level)
Backfill semantics	per-tenant-per-shard replay works

Why this works — concept by concept:

• TaskGroup mapping for the outer axis — 50 tenants × a multi-task subgraph is the exact shape @task_group + .expand() was designed for. Each mapped group runs its own fetch → process → publish independently; failure isolation is automatic.
• expand_kwargs inside the group for the inner axis — the shard-per-tenant fan-out uses expand_kwargs because each shard has multiple parameters (tenant + shard number). Cardinality per group is 10; total inner cardinality is 500.
• Cardinality arithmetic first — the senior habit: compute total instances (50 × 12 = 600) before writing code. 600 is comfortable; 60 000 would need batching.
• Per-instance retries at the shard level — retries=2 on transform_shard retries only the failed shard, not the whole tenant subgraph. Transient failures self-heal without operator involvement.
• Concurrency caps at multiple layers — max_active_tasks=64 (DAG) + max_active_tis_per_dag=8 on the mapped inner task. The DAG-level cap protects the scheduler; the per-task cap protects downstream services.
• Cost — 600 task instances in the metadata DB; DAG parse cost still O(1) because the graph structure is fixed (list_tenants → per_tenant). Scheduler evaluates 600 dependencies at run time; well within scheduler comfort at even 1-minute run frequency.

ETL
Topic — etl
ETL multi-dimensional fan-out and TaskGroup problems

Practice →

Optimization Topic — optimization Optimization problems on cardinality math and batching

Practice →

---

5. Production patterns — limits, backpressure, testing

Five patterns every senior ships — max_map_length cap, active-tasks backpressure, backfill sanity, unit test, monitoring

The mental model in one line: a mapped-task DAG in production is only healthy when it has (1) a max_map_length cap set on the specific mapped tasks, (2) max_active_tasks_per_dag and per-task concurrency backpressure, (3) predictable backfill semantics with the frozen-list contract, (4) a pytest recipe that unit-tests the underlying callable without spinning up Airflow, and (5) monitoring on mapped-instance failure rate. Every one of these is table stakes; missing any one is a future incident.

Pattern 1 — cardinality caps.

• max_map_length (per task). Set in airflow.cfg under [core]; default 1024. Override per-DAG or per-task where needed. If a mapped task's upstream returns more than the cap, the DAG fails fast with a clear AirflowException.
• In-DAG guard task. Add a guard task that checks the upstream list length before the mapped task materialises. Turns the opaque "max_map_length exceeded" into a readable message with the actual number.
• Total cardinality math. For nested expand or mapped TaskGroups, multiply the per-layer caps. If total > ~10 000, redesign — batch inside the mapped task or move to a batch-processing tool.

Pattern 2 — backpressure and concurrency control.

• max_active_tasks_per_dag (DAG level). Caps concurrent task instances across all tasks in the DAG. The mapped-task-aware value — the scheduler counts each mapped instance as one task.
• max_active_tis_per_dag (per mapped task). Caps concurrent instances of a specific mapped task. Use to throttle a single mapped task's fan-out without touching the whole DAG.
• Pools. The classic Airflow pool mechanism. task.partial(pool="external_api").expand(...) restricts every mapped instance to consume from the named pool — the right pattern for shared external-service concurrency budgets across DAGs.

Pattern 3 — backfill semantics.

• Frozen list contract. Once a DAG run starts, the mapped cardinality is frozen. A backfill of a past run replays the exact same list length the run originally used. The list is stored in the metadata DB alongside the DAG run.
• Backfill of a mapped DAG. Airflow supports airflow dags backfill for mapped-task DAGs. Each backfilled run uses its own frozen list.
• The catch. If the upstream's return value depends on external state (S3 bucket contents), the backfill re-runs the upstream, which pulls whatever is in S3 now — not what was there in the past. Design the upstream to be idempotent given the DAG run's logical_date (use partitioned data, execution_date-parameterised queries).

Pattern 4 — testing without Airflow.

• Unit-test the callable. The @task decorator produces a Python function; its underlying callable is task_fn.function or task_fn.__wrapped__. Call it directly with test kwargs in pytest — no Airflow scheduler needed.
• Test the fan-out expansion. For DAG-structure tests, use airflow.utils.dag_cycle_test or a lightweight dag.test() invocation. Assert the mapped task's map_length given a mocked upstream.
• Test the reducer separately. Reducer takes list[X]; test it with [a, b, c] directly. No Airflow needed.

Pattern 5 — monitoring.

• Mapped-instance failure rate. A dashboard metric: (# failed mapped instances) / (total mapped instances) per DAG run. Alert on failure rate > 10% for 2 runs.
• Total mapped instances per run. Trend line; unexpected spikes flag an upstream data change.
• Skipped mapped tasks. Mapped tasks that skipped (upstream returned empty list) are often a data-freshness issue upstream, not a healthy zero-work case.
• Per-map_index retry count. Track which map_index values retry most often — surfaces per-partition data quality issues that lurk under aggregate metrics.

Common interview probes on production patterns.

• "How do you cap fan-out?" — max_map_length + guard task; name both.
• "How do you throttle a mapped task?" — max_active_tis_per_dag (per-task) or pools.
• "How do you unit-test a mapped task?" — call the underlying callable directly with test kwargs.
• "What are the backfill semantics?" — frozen list per run; upstream must be execution_date-idempotent.

Worked example — DAG-level and per-task backpressure

Detailed explanation. A mapped DAG fans out 200 instances that each call an external HTTP service with a 100 QPS rate limit. Without backpressure, all 200 instances launch simultaneously, saturate the service, and half of them fail with 429 errors. The fix is two-layer backpressure — max_active_tasks_per_dag at the DAG level and pool-based rate limiting on the mapped task.

• The problem. 200 concurrent HTTP calls hit an external 100-QPS limit.
• Fix layer 1. max_active_tasks_per_dag = 32 — no more than 32 mapped instances run concurrently.
• Fix layer 2. Pool external_api with slots = 8 — no more than 8 tasks from any DAG hit the external service at once.

Question. Configure the DAG and pool such that the external service never sees more than 8 concurrent requests, and show the pool config command.

Input.

Layer	Setting	Purpose
DAG-level	max_active_tasks_per_dag = 32	Total concurrent tasks across the DAG
Per-task	max_active_tis_per_dag = 16	Per-mapped-task concurrency
Pool	external_api with 8 slots	Shared across DAGs

Code.

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

@dag(
    dag_id="throttled_fan_out",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
    max_active_tasks=32,                            # DAG-level cap
)
def throttled_fan_out() -> None:

    @task
    def list_targets() -> list[str]:
        return [f"target-{i:03d}" for i in range(200)]

    @task(
        pool="external_api",                        # shared 8-slot pool
        max_active_tis_per_dag=16,                  # per-task cap
        retries=3,
    )
    def call_external(target: str) -> dict:
        # HTTP call to the rate-limited service
        return {"target": target, "status": "ok"}

    targets = list_targets()
    call_external.expand(target=targets)

throttled_fan_out()

# Create the pool once via CLI (or use the UI at Admin → Pools)
airflow pools set external_api 8 "Rate-limited external HTTP service (8 concurrent)"

Step-by-step explanation.

1. list_targets returns 200 target strings. The mapped call_external fans out to 200 instances.
2. max_active_tasks=32 at the DAG level caps total concurrent task instances across the DAG at 32. Even though 200 instances exist, only 32 run at once.
3. pool="external_api" on the mapped task means each instance requires a pool slot to run. The pool has 8 slots (from the CLI command). So even though 32 tasks could run per the DAG-level cap, only 8 can actually acquire a pool slot at a time.
4. max_active_tis_per_dag=16 is a per-task cap — no more than 16 instances of this mapped task run concurrently. Redundant given the pool cap of 8; useful for cases where you want a per-task cap independent of the shared pool.
5. The effective concurrency is min(max_active_tasks, max_active_tis_per_dag, pool.slots) = 8. All 200 instances still eventually run; they queue on the pool and drain at 8-at-a-time. The external service sees a steady 8-concurrent request load.

Output.

Layer	Value	Effective?
DAG max_active_tasks	32	(not binding; pool caps first)
Per-task max_active_tis_per_dag	16	(not binding; pool caps first)
Pool slots	8	binding — external service sees ≤ 8 concurrent
External service load	8 concurrent	within 100 QPS limit
Total instances processed	200	(over ~25 seconds at 8-concurrent)

Rule of thumb. Use pools for cross-DAG rate limits (any DAG that talks to the same external service shares the pool). Use max_active_tis_per_dag for per-task throttling that does not need to coordinate with other DAGs. Use max_active_tasks at the DAG level to protect the scheduler from a single runaway DAG dominating the workers.

Worked example — unit-testing a mapped task without Airflow

Detailed explanation. A senior team writes unit tests for every mapped task's underlying callable. The @task decorator's underlying function is directly callable; it accepts the same kwargs the mapped instance receives. This means the test suite runs in milliseconds, has no Airflow dependency, and catches business-logic bugs before they ever reach a scheduler.

• The pattern. Each mapped task's business logic lives in a plain Python function; the @task decorator wraps it for Airflow.
• The test. Call the plain function directly with hand-crafted kwargs; assert the return value.
• The benefit. No Airflow scheduler, no metadata DB, no XCom — pure Python speed.

Question. Refactor a mapped task's logic into a plain callable and write pytest tests for it.

Input.

File	Contents
dags/process_file.py	plain function process_file_impl(bucket, key)
dags/process_file_dag.py	@task-wrapped version that calls the impl
tests/test_process_file.py	pytest suite for process_file_impl

Code.

# dags/process_file.py — plain callable, no Airflow imports
def process_file_impl(bucket: str, key: str) -> dict:
    """Business logic — no Airflow context needed."""
    if not key:
        raise ValueError("empty key")
    return {"bucket": bucket, "key": key, "size": len(key)}

# dags/process_file_dag.py — Airflow wrapper
from datetime import datetime
from airflow.decorators import dag, task
from dags.process_file import process_file_impl

@dag(
    dag_id="process_files",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
)
def process_files() -> None:

    @task
    def process_file(bucket: str, key: str) -> dict:
        return process_file_impl(bucket, key)       # delegate to plain callable

    @task
    def list_keys() -> list[str]:
        return ["k1", "k2", "k3"]

    process_file.partial(bucket="raw").expand(key=list_keys())

process_files()

# tests/test_process_file.py — pytest, no Airflow needed
import pytest
from dags.process_file import process_file_impl

def test_process_file_impl_happy_path():
    result = process_file_impl(bucket="raw", key="k1")
    assert result == {"bucket": "raw", "key": "k1", "size": 2}

def test_process_file_impl_empty_key_raises():
    with pytest.raises(ValueError, match="empty key"):
        process_file_impl(bucket="raw", key="")

@pytest.mark.parametrize("key,expected_size", [("a", 1), ("ab", 2), ("abcd", 4)])
def test_process_file_impl_size_matches(key, expected_size):
    result = process_file_impl(bucket="raw", key=key)
    assert result["size"] == expected_size

Step-by-step explanation.

1. process_file_impl is a plain Python function. It has no Airflow imports, no task decorator, no XCom dependency — pure business logic, testable with pytest.
2. The @task-decorated process_file inside the DAG is a one-line wrapper that delegates to process_file_impl(bucket, key). This separation is the senior habit: DAG file has orchestration wiring; impl file has business logic.
3. The pytest suite imports process_file_impl and calls it directly with test kwargs. No DagBag, no dag.test(), no TaskInstance — millisecond-fast unit tests.
4. pytest.mark.parametrize runs the same test with multiple inputs — the pattern for exhaustively testing the mapped task's edge cases (empty key, huge key, non-ASCII characters, etc.).
5. Integration testing (does the DAG actually run correctly under Airflow?) is a separate, coarser layer — a dag.test() call with a fresh in-memory metadata DB. Unit tests catch 90% of bugs at 1000× the speed.

Output.

Test	Speed	Verifies
test_process_file_impl_happy_path	< 1 ms	Business logic returns correct dict
test_process_file_impl_empty_key_raises	< 1 ms	Validation raises on empty key
test_process_file_impl_size_matches (parametrised, 3 cases)	< 3 ms total	Return value invariants

Rule of thumb. Extract every mapped task's business logic into a plain callable. The DAG file becomes a thin orchestration layer; the impl file becomes a testable Python module. Ship 100% pytest coverage on the impl, and rely on a small integration-test suite for the DAG wiring. This is the pattern every senior Airflow team ships.

Worked example — monitoring mapped-instance failure rate

Detailed explanation. Mapped tasks fan out to hundreds of instances; per-instance failures are the noisy failure surface. Aggregate-level monitoring — "how many mapped instances failed out of the total" — is the metric that tells you whether the DAG is degrading. Wire it into Prometheus + Grafana + PagerDuty for the full production loop.

• The metric. airflow_task_instances_failed_total{mapped=true} (from the Airflow StatsD/Prometheus exporter).
• The dashboard. Failure rate per DAG per run; trend line; alert threshold.
• The alert. Rate > 10% for 2 consecutive runs pages on-call.

Question. Write a Python snippet that queries the Airflow metadata DB for the last 20 runs of a mapped DAG and computes the mapped-instance failure rate. Show the alert threshold.

Input.

Metric	Threshold	Action
Mapped failure rate	> 10% for 2 runs	page on-call
Mapped skip rate	> 20% for 2 runs	investigate upstream
Total mapped instance count	> 2× last week average	audit upstream data

Code.

# Query the Airflow metadata DB for mapped-instance failure rates
from datetime import datetime, timedelta
import psycopg2
from psycopg2.extras import RealDictCursor

def mapped_failure_rate(dag_id: str, hours: int = 24) -> dict:
    """Return failure and skip rate for mapped instances of a DAG over the last N hours."""
    conn = psycopg2.connect("postgres://airflow@airflow-db:5432/airflow")
    conn.autocommit = True
    since = datetime.utcnow() - timedelta(hours=hours)
    with conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute("""
            SELECT
                run_id,
                COUNT(*) FILTER (WHERE map_index >= 0)                              AS total_mapped,
                COUNT(*) FILTER (WHERE map_index >= 0 AND state = 'failed')        AS failed_mapped,
                COUNT(*) FILTER (WHERE map_index >= 0 AND state = 'skipped')       AS skipped_mapped,
                COUNT(*) FILTER (WHERE map_index >= 0 AND state = 'success')       AS success_mapped
            FROM task_instance
            WHERE dag_id = %s
              AND start_date > %s
              AND map_index >= 0
            GROUP BY run_id
            ORDER BY run_id DESC
            LIMIT 20;
        """, (dag_id, since))
        rows = cur.fetchall()

    latest = rows[0] if rows else None
    if latest and latest["total_mapped"] > 0:
        failure_rate = latest["failed_mapped"] / latest["total_mapped"]
        skip_rate    = latest["skipped_mapped"] / latest["total_mapped"]
        return {
            "run_id":         latest["run_id"],
            "total_mapped":   latest["total_mapped"],
            "failure_rate":   round(failure_rate, 4),
            "skip_rate":      round(skip_rate, 4),
            "alert_page":     failure_rate > 0.10,
            "alert_investigate": skip_rate > 0.20,
        }
    return {}

if __name__ == "__main__":
    print(mapped_failure_rate("process_files"))

Step-by-step explanation.

1. The query filters task_instance rows to those with map_index >= 0 — the standard way to identify mapped-task instances in the Airflow metadata DB. Non-mapped tasks have map_index = -1.
2. The aggregation counts total, failed, skipped, and successful mapped instances per DAG run. The result is a per-run summary that can be trended over the last N runs.
3. The failure rate = failed / total. Alert when > 10% for 2 consecutive runs (the "2 consecutive" guard is applied in the alerting rule, not the query — Prometheus's for: 20m clause).
4. The skip rate = skipped / total. Skips are usually a data-freshness issue upstream — a partition that should have data but does not. Alert with a higher threshold (20%) because occasional skips are normal.
5. The runbook when the failure-rate alert fires: (a) open the Airflow grid view for the failing DAG; (b) inspect a sample of failing map_index values; (c) look at the per-instance logs; (d) diagnose whether the failures are transient (retry) or systemic (fix upstream data).

Output.

DAG run	Total mapped	Failed	Skipped	Failure rate	Alert?
2026-06-22 12:00	200	8	0	4.0%	no
2026-06-22 13:00	200	28	12	14.0%	yes — page
2026-06-22 14:00	195	22	5	11.3%	yes — 2 consecutive

Rule of thumb. Aggregate failure-rate monitoring catches degradation that per-instance retries mask. Ship the failure-rate query as a Grafana panel + PagerDuty alert the same week the mapped DAG lands in production. The metric is O(1) per DAG run to compute; the operational visibility is 10× the cost.

Senior interview question on production mapped-task patterns

A senior interviewer might ask: "You inherit a mapped-task DAG in production. The upstream sometimes returns 5000 keys and the DAG has been failing intermittently. Walk through your first week: what caps do you set, what tests do you write, what monitoring do you add, and how do you communicate the changes to the team."

Solution Using a five-day production hardening plan

First-week hardening plan — mapped-task DAG in production
==========================================================

Day 1 — Cardinality caps
  - Set max_map_length in airflow.cfg (default 1024; raise deliberately if needed)
  - Add in-DAG guard task with readable exception on oversize / empty upstream
  - Verify: force upstream to return 2000 keys; DAG fails fast with clear message

Day 2 — Backpressure
  - Set max_active_tasks on the DAG (protect scheduler)
  - Set max_active_tis_per_dag on the mapped task (throttle per-task fan-out)
  - Create Airflow pool for shared external services; assign via .partial(pool=...)

Day 3 — Unit tests
  - Extract mapped-task business logic into plain callable
  - pytest suite covering happy path + edge cases + parameterised inputs
  - Add coverage gate in CI (>= 90% for impl module)

Day 4 — Backfill semantics
  - Verify upstream is execution_date-idempotent (partitioned data, dated queries)
  - Test a backfill of a past run — mapped cardinality must match the original
  - Document the frozen-list contract in the DAG docstring

Day 5 — Monitoring + on-call runbook
  - Wire mapped-instance failure rate metric to Prometheus
  - Grafana panel: failure_rate over last 20 runs
  - PagerDuty alert: failure_rate > 10% for 2 consecutive runs
  - Runbook: grid view → per-instance logs → root cause → retry or fix

Step-by-step trace.

Day	Activity	Output
1	Cardinality caps	max_map_length + guard task; fail-fast on oversize
2	Backpressure	DAG + per-task + pool concurrency caps
3	Unit tests	impl module + pytest suite + coverage gate
4	Backfill sanity	idempotent upstream + tested backfill
5	Monitoring	failure_rate metric + Grafana + PagerDuty + runbook

After day 5, the DAG has the five production patterns wired in. The team can sleep at night; the on-call has a paved runbook; the next transient upstream spike becomes a "10-minute inspect and retry" rather than a "3-hour post-mortem."

Output:

Pattern	Before	After
Cardinality caps	none	max_map_length + guard task
Backpressure	none	DAG + per-task + pool caps
Unit tests	none	impl module + pytest suite
Backfill semantics	undocumented	tested + documented
Monitoring	none	failure rate + on-call alert

Why this works — concept by concept:

• Cardinality caps as safety net — max_map_length + guard task turn opaque OOMs into fast, actionable failures. The DAG fails within seconds with a clear message naming the exact upstream count.
• Backpressure at every layer — DAG-level, per-task, and pool caps compose. The most restrictive wins; the effective concurrency is min(all caps). This is the pattern that protects downstream services from a single misbehaving DAG.
• Extracted impl modules — the plain callable is the unit-test target; the @task-wrapped version is orchestration. Ship pytest coverage on the impl; rely on integration tests for the DAG. Millisecond test suite; second-scale CI.
• Idempotent upstream for backfill — the frozen-list contract only works if the upstream returns the same list for the same DAG run. Design the upstream to be execution_date-parameterised (partitioned tables, dated S3 prefixes) so backfills reproduce exactly.
• Failure-rate metric — per-instance retries mask individual failures; aggregate failure rate surfaces systemic degradation. The metric is cheap to compute and priceless during an incident.
• Cost — 5 days of senior-DE time for the hardening rollout; each pattern is 1-2 hours of work. The avoided cost of one 3 AM incident pays for the entire week. O(1) per DAG at parse time; O(N) per run at task time.

ETL
Topic — etl
ETL production hardening and monitoring problems

Practice →

Optimization
Topic — optimization
Optimization problems on concurrency caps and backpressure

Practice →

---

Cheat sheet — dynamic mapping recipes

• Basic .expand() + .partial() template. task.partial(bucket="raw", retries=3).expand(key=[...]) — pin fixed kwargs first, then map the varying kwarg. One task node, N instances at run time. map_index numbers instances 0..N-1.
• Upstream fan-in from list output. process_file.partial(bucket="raw").expand(key=list_files.output) — the .output attribute of an upstream @task returns an XComArg; the mapped cardinality equals the length of the returned list at run time. This is the map in map-reduce.
• expand_kwargs for heterogeneous kwargs. process.expand_kwargs([{k1:v1, k2:v2}, ...]) — fan out over a list of kwarg dicts. Prefer this over parallel .expand() lists when instances need multiple varying kwargs together — atomic dicts eliminate zip-alignment bugs.
• Nested .expand() pattern. outer_mapped.expand(...) then inner_mapped.expand(x=outer_mapped.output) — cardinality multiplies. Total instances = outer × inner. Do the arithmetic before shipping; batch inside the mapped task if the product > ~10 000.
• TaskGroup mapping (Airflow 2.9+). @task_group decorated subgraph accepts kwargs; .expand(kwarg=...) maps the whole group. Each mapped invocation creates a full sub-DAG of the constituent tasks. Use when the per-outer-element work is a multi-task chain.
• max_map_length cap. Set in airflow.cfg under [core] max_map_length = 1024. Global per-task limit; the scheduler refuses to materialise more instances. Add an in-DAG guard task with a readable exception on oversize upstream to turn opaque failures into actionable ones.
• max_active_tasks_per_dag backpressure. DAG-level cap on concurrent task instances. dag(max_active_tasks=32). Effective concurrency = min(this, per_task_cap, pool_slots). Protects the scheduler from a runaway DAG.
• Per-task concurrency + pools. task.partial(pool="external_api", max_active_tis_per_dag=16).expand(...). Pools are shared across DAGs; max_active_tis_per_dag is per-DAG per-task. Combine to enforce a service-level rate limit and a per-DAG cap.
• Reducer trigger rules. all_success (default) blocks reducer on any mapped failure; all_done runs reducer over survivors; none_failed_min_one_success runs reducer if no failures and at least one success. Pick deliberately — this is a semantic decision, not a footnote.
• map_index from inside the task body. context["ti"].map_index (if the task accepts **context) or get_current_context()["ti"].map_index (if not). Returns 0..N-1 for mapped instances; -1 for non-mapped. Use for per-instance file paths, log names, cache keys.
• Zero-instance semantics. If .expand() gets an empty list, 0 mapped instances materialise; downstream tasks receive an empty list and typically transition to skipped. Treat empty upstream as a possible healthy case or as an alert case, but never as a silent no-op.
• Backfill contract. The mapped cardinality is frozen per DAG run; a backfill of a past run replays the exact same list. Design the upstream to be execution_date-idempotent (partitioned data, dated queries) so backfills reproduce.
• Unit test the plain callable. Extract mapped-task logic into a plain function process_impl(bucket, key). The @task decorator is a one-line wrapper. Pytest tests the plain function; no Airflow scheduler needed. Millisecond-fast tests, 90%+ coverage achievable.
• Monitoring metric. Track mapped-instance failure rate per DAG per run. Alert on rate > 10% for 2 consecutive runs. Aggregate metric catches degradation that per-instance retries mask.
• When to batch instead of fan out. If total cardinality > 10 000 or per-instance work is fast (< 10 s), batch inside a single mapped instance (process_batch(keys: list[str])) rather than fanning out further. Scheduler bookkeeping dominates fast per-instance work; batching wins.

Frequently asked questions

What is airflow dynamic task mapping and when did it land?

airflow dynamic task mapping is the runtime fan-out primitive introduced in Airflow 2.3 (April 2022) via the .expand() and .partial() decorator methods. It replaces the pre-2.3 "loop-generated DAG" pattern — a Python for loop that constructed operators at DAG-parse time — with a single task node in the DAG graph that materialises N instances at run time based on the upstream's actual output. Airflow 2.5 added expand_kwargs() for heterogeneous per-instance parameter dicts; 2.7 added nested mapping (a mapped task's output can feed another .expand()); 2.9 added TaskGroup mapping (a whole subgraph can be mapped as one unit). The senior-DE framing: parse-time cardinality is the anti-pattern; run-time cardinality via .expand() is the primitive that unlocks parameterised parallelism at scale. Every modern Airflow deployment relies on it for at least one workflow; interviewers assume you know the syntax and probe your knowledge of the limits — cardinality caps, backfill semantics, XComArg chaining, and the airflow map_index accounting.

airflow expand vs airflow partial — what's the difference?

.expand() fans out — it takes an iterable (either a Python literal or an upstream task's XComArg) and spawns one mapped instance per element. Each instance receives the ith element of the iterable in the named kwarg. .partial() pins — it fixes kwargs that are the same for every mapped instance. The canonical call chain is task.partial(fixed_kwarg=value).expand(mapped_kwarg=iterable); the mnemonic is "pin the fixed bits first, then fan out." Task-level attributes (retries, execution_timeout, pool, max_active_tis_per_dag) are set via .partial() because they apply per-instance identically. You cannot use .expand() alone if you have fixed kwargs to set — the .partial() step is where they go. Positional arguments are not supported in .expand(); everything is keyword-only, which forces call-site clarity. The interview one-liner: expand fans; partial pins; the combination gives you the full parameterisation of one task node into N instances at runtime.

What is airflow map_index and how do I use it inside a mapped task?

airflow map_index is the 0-based per-instance index that identifies which mapped instance you are inside the task body. It appears in the Airflow UI grid view (one row per index), in log filenames (.../map_index=0/attempt=1.log), in XCom keys (each instance's return value is stored under its own map_index), and in context["ti"].map_index inside the task callable. For decorated tasks that do not accept **context explicitly, use get_current_context()["ti"].map_index. Non-mapped tasks have map_index = -1 (a sentinel meaning "not applicable"). Common uses: per-instance file naming (/tmp/proc_{map_index}.log), per-instance sub-log directories, per-instance cache keys, targeted retry logic (clear only failed indices from the UI). The senior signal is naming map_index=-1 as the non-mapped sentinel and knowing where the index appears in every UI/log/XCom surface — the accounting only matters once you have debugged one failure among 200 instances.

Can I nest .expand() calls? What are the risks?

Yes — Airflow 2.7+ supports nested dynamic task mapping. A mapped task's output (list-returning) can itself be .expand()-ed by a downstream task. Cardinality multiplies across layers: outer 100 × inner 100 = 10 000 instances. The two risks are the cardinality explosion trap — three-layer nesting (100 × 100 × 100 = 1 000 000) is technically supported by Airflow but destroys the scheduler — and the composition complexity of nested list-of-lists XCom values, which are often clearer as a flatten step followed by .expand_kwargs() over the flattened pairs. The senior habit is doing the cardinality arithmetic before writing the code: if total instances > 10 000, redesign — either flatten into a single-level .expand_kwargs() over pre-computed pairs, or batch inside each mapped instance so one instance handles K units of work. Airflow 2.9's TaskGroup mapping is often the cleaner alternative to two-level nested expand when the per-outer-element work is a multi-task subgraph rather than a single task.

How do I cap the fan-out so an oversized upstream does not detonate the scheduler?

Three layers, name all three. max_map_length in airflow.cfg under [core] is the global per-task hard cap; default 1024. If a mapped task's upstream returns more than the cap, Airflow refuses the fan-out with a clear AirflowException. In-DAG guard task — a defensive task between the upstream and the mapped task that checks the list length and raises a readable exception naming the actual count. Turns "opaque max_map_length exceeded" into "upstream returned 5432 keys > cap 1024 — investigate list_files." max_active_tasks_per_dag (DAG level) and max_active_tis_per_dag (per mapped task) are backpressure — they do not reject the fan-out, they serialise it. If max_active_tasks = 32 and fan-out is 500, at most 32 instances run concurrently; the rest queue. Pools add a fourth layer: task.partial(pool="external_api").expand(...) restricts every mapped instance to consume from a named pool, giving you cross-DAG rate limiting. Senior answer: cap protects against forgetting; backpressure protects against overwhelming downstream; both are non-negotiable in production.

Does dynamic task mapping work with TaskFlow, and how do I unit-test a mapped task?

Yes — the TaskFlow API (@dag + @task decorators) is the preferred way to write mapped tasks. @task-decorated Python functions expose .expand(), .partial(), .expand_kwargs(), and .output (the XComArg). The whole mapped-DAG pattern in this guide uses TaskFlow; classic operators support the same API but the decorator form reads more clearly. For unit testing, extract the task's business logic into a plain Python function (process_file_impl(bucket, key)) with no Airflow imports; make the @task-decorated wrapper a one-line delegation. Pytest then tests the plain function directly — no Airflow scheduler, no metadata DB, no XCom mocking, no TaskInstance fixtures. Millisecond test suite; 90%+ coverage achievable. For DAG-structure integration tests, use dag.test() (Airflow 2.5+) which runs the whole DAG in-process against an ephemeral metadata DB — coarser and slower, but exercises the XCom wiring end-to-end. The senior pattern is 90% unit tests on the impl module + 10% integration tests on the DAG wiring.

Practice on PipeCode

• Drill the ETL practice library → for the fan-out, parameterised-task, and orchestration problems senior interviewers love.
• Rehearse on the SQL practice library → for the aggregation and reducer-over-mapped-results problems that pair with Airflow map-reduce.
• Sharpen the tuning axis with the optimization practice library → for the cardinality-cap, scheduler-cost, and backpressure problems.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the .expand() / .partial() intuition against real graded inputs.

Lock in dynamic-mapping muscle memory

Airflow docs explain syntax. PipeCode drills explain the decision — when `.expand()` replaces the loop, when nested mapping is a footgun, when `TaskGroup` mapping wins, when batching beats fan-out. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice optimization problems →