Airflow Dynamic Task Mapping: Fan-Out That Works, and Fan-Out That Melts the Scheduler

I once wrote a DAG that mapped over every file in a landing bucket. It worked beautifully in testing, where there were nine files. In production the partner uploaded a backlog and there were 6,400.

Airflow did exactly what I asked. It created 6,400 mapped task instances, wrote 6,400 rows to the metadata database, and then the scheduler started evaluating all of them on every loop. The UI grid view timed out. Other DAGs stopped getting scheduled promptly because the scheduler was busy. The actual work — 6,400 small file loads — would have taken eleven minutes as a single task.

Dynamic task mapping is the right tool for a genuine fan-out. It is also the easiest way to turn a data problem into an orchestrator problem. The distinction is entirely about how many map indices you create and what each one is worth.

The mechanics, briefly

expand() creates one task instance per element of a list resolved at run time. partial() supplies the arguments that stay the same across all of them.

import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
)
def load_partner_files():

    @task
    def list_files(prefix: str) -> list[str]:
        # Resolved at run time. This return value becomes the fan-out width.
        import boto3
        s3 = boto3.client("s3")
        pages = s3.get_paginator("list_objects_v2").paginate(
            Bucket="lake-landing", Prefix=prefix
        )
        return [o["Key"] for page in pages for o in page.get("Contents", [])]

    @task(
        retries=2,
        # Throttle: never more than 8 of these running at once,
        # regardless of how wide the fan-out is.
        max_active_tis_per_dag=8,
    )
    def load_file(key: str, target_table: str) -> int:
        ...
        return 1

    keys = list_files(prefix="partners/acme/{{ ds }}/")
    load_file.partial(target_table="raw.acme_events").expand(key=keys)

load_partner_files()

Two things in that snippet do most of the safety work. max_active_tis_per_dag caps concurrency for that specific task, so a wide fan-out does not open two hundred simultaneous connections to a warehouse that allows forty. And the fan-out width comes from a task return value, which means it goes through XCom — a fact that becomes important shortly.

For heterogeneous arguments, expand_kwargs takes a list of dicts:

@task
def build_jobs() -> list[dict]:
    return [
        {"table": "orders",    "mode": "merge",  "warehouse": "loading_m"},
        {"table": "customers", "mode": "merge",  "warehouse": "loading_s"},
        {"table": "events",    "mode": "append", "warehouse": "loading_l"},
    ]

@task
def run_job(table: str, mode: str, warehouse: str) -> None:
    ...

run_job.expand_kwargs(build_jobs())

I prefer expand_kwargs with an explicit dict over several parallel expand() arguments. Multiple expand() arguments produce the cross product, which is a fine feature and a memorable outage when someone expects a zip. Two lists of thirty become nine hundred tasks.

Sizing the fan-out: the only question that matters

Before writing expand, I answer one question: how many map indices will this produce on the worst day, not the typical day?

My thresholds, from operating this on a mid-sized deployment:

• Under 100 indices: fine, no special handling.
• 100 to 500: fine if each task does real work — at least tens of seconds. Set max_active_tis_per_dag.
• 500 to 1,000: batch it. The UI is getting slow and the metadata cost is real.
• Over 1,000: you have designed the wrong thing. Batch, or push the parallelism into the system that is good at it.

Set a hard cap so the worst day cannot surprise you. max_map_length in config bounds it globally (default 1024), and the DAG should fail loudly rather than create a scheduler incident.

The second question: what does each index cost to schedule? A mapped task instance carries scheduler evaluation, at least two database writes for state transitions, an XCom row, a log file, and a row in the grid view. On a Kubernetes executor it also carries pod startup, which is seconds of overhead per index. If the work inside is a two-second API call, you are spending more on orchestration than on the task.

flowchart TD
  q1["Worst-case item count?"] -->|"< 100"| direct["expand directly"]
  q1 -->|"100-1000"| work["Is each item > 30s of work?"]
  q1 -->|"> 1000"| batch["Batch into chunks"]
  work -->|"yes"| direct
  work -->|"no"| batch
  batch --> chunk["expand over chunks, loop inside"]
  direct --> throttle["Set max_active_tis_per_dag"]
  chunk --> throttle

The batching branch is the one most teams skip, and it is where the scheduler incidents come from.

Batching, which is the pattern that scales

Batching converts N tiny tasks into N/k tasks that each loop over k items. You keep parallelism and retryability at a useful granularity and you stop paying orchestration overhead per row.

from itertools import islice
from airflow.sdk import dag, task

def chunked(items: list, size: int) -> list[list]:
    it = iter(items)
    return list(iter(lambda: list(islice(it, size)), []))

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def load_partner_files_batched():

    @task
    def list_files() -> list[str]:
        ...

    @task
    def make_batches(keys: list[str], batch_size: int = 25) -> list[list[str]]:
        batches = chunked(keys, batch_size)
        # Fail fast instead of melting the scheduler on a backlog day.
        if len(batches) > 200:
            raise ValueError(
                f"{len(keys)} files would create {len(batches)} tasks. "
                "Investigate the backlog before running."
            )
        return batches

    @task(retries=2, max_active_tis_per_dag=10)
    def load_batch(keys: list[str]) -> int:
        loaded = 0
        for key in keys:
            # One transaction per file keeps a mid-batch failure
            # re-runnable, since the load itself is idempotent on key.
            load_one(key)
            loaded += 1
        return loaded

    load_batch.expand(keys=make_batches(list_files()))

load_partner_files_batched()

The explicit guard in make_batches is the part I would not skip. It turns “the scheduler is unresponsive and nobody knows why” into a failed task with a message naming the cause. That trade is always worth it.

Batch size should reflect the retry unit you want. If a single file failure should not force twenty-four re-loads, either keep batches small or make the inner loop idempotent per file so a retry is cheap. Idempotency is doing the heavy lifting here, same as everywhere else in orchestration.

Mapped task groups for multi-step per-item work

When each item needs several steps, do not create three parallel mapped tasks and hope the indices line up. Map the task group.

from airflow.sdk import dag, task, task_group

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def per_region_pipeline():

    @task
    def regions() -> list[str]:
        return ["us-east", "eu-west", "ap-south"]

    @task_group
    def process(region: str):

        @task
        def extract(region: str) -> str:
            return f"s3://lake/staging/{region}/{{{{ ds }}}}/"

        @task
        def validate(path: str) -> str:
            # Raise here and only this region's branch fails.
            return path

        @task
        def publish(path: str) -> None:
            ...

        publish(validate(extract(region)))

    process.expand(region=regions())

per_region_pipeline()

Each region gets its own extract, validate, publish chain, and a failure in eu-west does not stop us-east. This is what people used to build with SubDAGs, which are gone in Airflow 3 for good reasons — see the Airflow 3 migration notes.

Note the group multiplies indices: three regions times three tasks is nine task instances. Apply the sizing thresholds to the product, not the list length.

XCom is the hidden constraint

The list driving expand is an XCom value. On the default database XCom backend it is serialized into the metadata database. A list of 6,400 S3 keys at eighty characters each is half a megabyte of JSON in a table the scheduler reads constantly, and each mapped instance also pulls its slice.

Symptoms of getting this wrong are indirect: the scheduler slows down, the metadata DB grows faster than expected, and the grid view becomes unusable. Nobody looks at the XCom table.

Two mitigations. Return references instead of payloads — a manifest path rather than the file contents, an ID list rather than rows. And configure a custom XCom backend that spills large values to object storage, keeping only a pointer in the DB.

# plugins/s3_xcom_backend.py
import json
import uuid
from typing import Any

import boto3
from airflow.models.xcom import BaseXCom

BUCKET = "airflow-xcom-spill"
THRESHOLD_BYTES = 64 * 1024


class S3XComBackend(BaseXCom):
    """Spill large XCom values to S3; keep a pointer in the metadata DB."""

    @staticmethod
    def serialize_value(value: Any, **kwargs):
        payload = json.dumps(value).encode()
        if len(payload) < THRESHOLD_BYTES:
            return BaseXCom.serialize_value(value, **kwargs)

        key = f"xcom/{uuid.uuid4()}.json"
        boto3.client("s3").put_object(Bucket=BUCKET, Key=key, Body=payload)
        return BaseXCom.serialize_value({"__s3_xcom__": key}, **kwargs)

    @staticmethod
    def deserialize_value(result):
        value = BaseXCom.deserialize_value(result)
        if isinstance(value, dict) and "__s3_xcom__" in value:
            obj = boto3.client("s3").get_object(
                Bucket=BUCKET, Key=value["__s3_xcom__"]
            )
            return json.loads(obj["Body"].read())
        return value

Set a lifecycle rule on that bucket. Spilled XComs are garbage the day after the run, and without expiry the bucket becomes an archive of every list you ever mapped over.

Throttling: three different knobs for three different limits

These get confused constantly and they protect different things.

max_active_tis_per_dag on the task limits how many instances of that task run at once across all DAG runs. This is the one that protects a downstream API or warehouse from your fan-out specifically.

Pools limit concurrency across tasks from any DAG. If four different DAGs all hit the same partner API, a shared pool of six slots is the only thing that enforces the real limit.

max_active_runs on the DAG limits concurrent runs. Without it, a catch-up or a backfill runs five days at once and each one fans out.

@task(
    pool="partner_api",              # shared across every DAG
    pool_slots=1,
    max_active_tis_per_dag=8,        # this task specifically
    retries=3,
    retry_exponential_backoff=True,
    retry_delay=timedelta(seconds=30),
    max_retry_delay=timedelta(minutes=10),
)
def call_partner_api(record_id: str) -> dict:
    ...

Exponential backoff matters more with mapped tasks than with regular ones. When a rate-limited API rejects your fan-out, a fixed retry delay means all failed indices retry simultaneously and get rejected again, in lockstep, until they exhaust retries. Backoff spreads them out.

Where teams get this wrong

Mapping over rows instead of batches. One task per row is the purest form of this mistake. If your fan-out list came from a SELECT, batching is almost certainly correct.

No upper bound on the list. The DAG works for a year, then a backlog or a bad upstream filter produces ten thousand indices. Add the guard.

Multiple expand() arguments expecting a zip. You get the cross product. Use expand_kwargs with pre-zipped dicts when you want pairwise.

Assuming a failed index fails the DAG usefully. By default a mapped task’s failure propagates downstream, but with hundreds of indices the useful information — which three failed and why — is buried. Aggregate failures in a downstream task that reads the results and reports them in one place.

Huge lists through the default XCom backend. Half a megabyte of JSON per run in the metadata database, forever, until someone finds it.

Mapping over something the scheduler must compute at parse time. The list must come from a task at run time. A list computed at parse time by querying a database means every DAG file parse hits that database, which is the Airflow version of the unguarded query problem described in dbt macros and Jinja patterns.

FAQ

How many mapped tasks is too many?

As a practical ceiling, keep a single expand under about 500 indices, and under 100 if each index is short work. max_map_length defaults to 1024 for a reason. If you need more parallelism than that, the parallelism belongs in Spark, a warehouse query, or a batch service — not in scheduler rows.

Can I map over the results of another mapped task?

Yes. The downstream task receives the collected list of upstream returns, and you can expand over it again. Be careful: this multiplies widths, and a 50-index task returning 50 items each becomes 2,500 indices.

How do I see which mapped index failed?

The grid view has a mapped task detail listing per-index states, and you can clear a single index for retry. At high widths this view is slow, which is another argument for batching — with 40 batches you can read the list.

Do mapped tasks work with deferrable operators?

Yes, and it is a good combination. A hundred deferred sensors cost triggerer watches rather than worker slots, which is far cheaper than a hundred poking sensors. The details are in deferrable operators.

Can a mapped task emit an asset event?

Yes. Put the outlet on the mapped task and the event fires as instances succeed, which usually means many events. If you want one event when the whole fan-out is done, put the outlet on a downstream aggregation task instead — see data-aware scheduling.

What happens if the list is empty?

Zero mapped instances are created and the task is marked skipped. Downstream tasks with the default trigger rule are then skipped too, which is usually right but occasionally surprising. If an empty list is an error condition, raise in the task that builds it.

What this means for your pipelines

Dynamic task mapping earns its place when the items are genuinely independent, genuinely worth retrying separately, and there are a sensible number of them. A dozen regions, forty partner feeds, a hundred tables to snapshot: that is what it is for, and it replaced a lot of ugly code that generated DAGs from config files.

What it is not is a parallel execution engine. Airflow’s scheduler tracks state for every index in a relational database, and that cost is per index regardless of whether the task takes two seconds or two hours. When the per-item work is small, batching is not a compromise — it is the correct design, and it usually runs faster end to end because you stop paying task startup overhead thousands of times.

So write the guard before you write the expand. Decide the worst-case width, cap it, batch anything that would exceed a few hundred, and set the concurrency limits that protect whatever your tasks are calling. Those four lines of defensive code cost you ten minutes now, and they are the difference between a fan-out that quietly handles a backlog day and a fan-out that takes the scheduler down with it.