Fan Out Over Files You Haven't Seen Yet: Dynamic Task Mapping in Airflow

The nightly pipeline with an unknown workload

Every night, new files land in an object-storage bucket. Some nights there are three; some nights there are three hundred. Your Airflow DAG has to process each one — download it, parse it, load it — and you'd like per-file retries, per-file logs, and a grid view where you can see exactly which file failed.

The two naive options both hurt. You can pre-create a fixed number of tasks and let extras sit idle, or you can hide a for loop inside one big task and lose per-item visibility entirely. The useful takeaway: Airflow's dynamic task mapping, stable since Airflow 2.3, lets the DAG itself decide at runtime how many task instances to create — one per element of a list produced by an upstream task.

What .expand() actually does

Normally, the number of tasks in a DAG is fixed when the scheduler parses the DAG file. Dynamic task mapping breaks that rule in a controlled way. When you call .expand() on an operator or a @task function, you tell the scheduler: wait until the upstream task finishes, take its return value, and create one mapped task instance per element.

Each mapped instance is a real task instance. It gets its own log stream, its own retry counter, its own row in the grid view, and its own state. If file 47 of 200 fails to parse, you retry file 47 — not the whole batch. That is the entire value proposition, and it is exactly what a loop inside a single task cannot give you.

This is also why mapping beats generating tasks in a Python loop at parse time. Parse-time generation needs the input list when the scheduler reads the DAG file, which happens repeatedly and must stay fast. Data that changes between runs — like tonight's set of new files — simply isn't available then. Mapping defers the decision to the moment the data actually exists.

A worked example: list, process, summarize

Here is the shape of a file-processing DAG using the TaskFlow API. It assumes Airflow 2.3 or newer (check with airflow version on any node where the CLI is installed):

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

@dag(start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def nightly_files():

    @task
    def list_pending_keys() -> list[str]:
        # Return object keys, not file contents.
        return list_new_objects(bucket="incoming", prefix="nightly/")

    @task
    def process_file(key: str) -> dict:
        local = download(bucket="incoming", key=key)
        rows = parse(local)
        load(rows)
        return {"key": key, "rows": len(rows)}

    @task
    def summarize(results: list[dict]):
        total = sum(r["rows"] for r in results)
        print(f"Processed {len(results)} files, {total} rows")

    keys = list_pending_keys()
    outcomes = process_file.expand(key=keys)
    summarize(outcomes)

Three things are worth noticing. First, list_pending_keys returns references — object keys — not file contents. Mapped inputs travel through XCom, Airflow's mechanism for passing values between tasks, and by default XCom values are stored in the metadata database. Pushing megabytes of file data through it will bloat that database fast.

Second, summarize depends on the mapped task, so it waits for all mapped instances and receives their collected outputs as a list. Third, if tonight's bucket is empty, the mapped task simply expands to zero instances and the DAG still succeeds cleanly.

Two knobs cover the common variations. Use partial() to pin constant arguments while expand() supplies the mapped one — for example process_file.partial(bucket="incoming").expand(key=keys). And when each item needs several arguments, expand_kwargs() accepts a list of dicts, one per instance.

The trade-offs you'll actually feel

Capacity planning gets fuzzier. Because the instance count is only known after the upstream task runs, you can't size a worker pool from the DAG file. A quiet night creates three instances; a busy one creates three hundred that all want a worker slot at once. Airflow exposes max_active_tis_per_dag to throttle how many mapped instances run concurrently — but its exact semantics have shifted between releases, so confirm the behavior in the docs for your version before relying on it.

Scheduler and metadata-database load scale with the list. Every mapped instance is a row in the task-instance table and work for the scheduler. Hundreds of items is comfortable territory for a typical deployment. Tens of thousands is a design conversation, not a default. If your lists routinely run that large, batch the keys into chunks and map over the chunks instead.

Everything passes through XCom. Both the expanded input list and each instance's return value are serialized into XCom. Keep payloads small — keys, paths, counts — and write anything bulky to storage, passing only the pointer.

Try it this week

Take one DAG where a loop is currently hidden inside a single task. Split it into a list-producing task and a mapped task with .expand(), keep the payloads to references, and set a per-item retry policy. Then run it once with a five-item input and open the grid view: you should see five mapped instances, each with its own logs and retry button. That five-minute check tells you more than any benchmark — and if the per-file visibility alone doesn't sell the refactor, the first time you retry one failed file instead of the whole batch will.