Airflow at scale: dynamic task mapping, data-aware scheduling with datasets/assets and pool limits

Two features change how large Airflow deployments are built. Dynamic task mapping creates one task instance per input at run time. Data-aware scheduling (datasets in Airflow 2, assets in Airflow 3) starts a DAG when the data it depends on is updated, instead of guessing with a cron and sensors.

#Dynamic task mapping

from datetime import datetime
from airflow.decorators import dag, task     # Airflow 3: from airflow.sdk import dag, task

@dag(schedule="@daily", start_date=datetime(2026, 10, 1), catchup=False, max_active_tasks=32)
def ingest_partners():

    @task
    def list_files(**ctx) -> list[str]:
        # e.g. list objects under s3://landing/partners/{ds}/
        return [f"s3://landing/partners/{ctx['ds']}/p{i}.csv" for i in range(40)]

    @task(retries=3, max_active_tis_per_dag=10)
    def load_file(path: str) -> int:
        ...
        return 1000

    @task
    def summarise(counts: list[int]):
        print(f"{len(counts)} files, {sum(counts)} rows")

    summarise(load_file.expand(path=list_files()))

ingest_partners()

• .expand() creates one mapped instance per element. Each retries and logs independently.
• max_active_tis_per_dag caps how many mapped instances run at once, protecting the target system.
• The downstream task receives the list of results (a lazy sequence; iterate or convert to list).

#Constant and mapped arguments

load_file.partial(target_table="raw.partners").expand(path=list_files())

#Mapping over pairs

@task
def make_jobs() -> list[dict]:
    return [{"src": "a.csv", "dst": "t_a"}, {"src": "b.csv", "dst": "t_b"}]

copy.expand_kwargs(make_jobs())

Warning

Each mapped instance is a row in the metadata database and a unit of scheduling work. Mapping over 50,000 items per run will slow the scheduler for everyone. Batch inputs (e.g. 500 files per mapped task) or move very large fan-outs to Spark or a batch service. Airflow also enforces max_map_length (default 1024) per expand.

#Partial failures

By default a failed mapped instance fails the downstream task (trigger rule all_success). If you want to continue with what succeeded:

@task(trigger_rule="all_done")
def summarise(counts): ...

Then report which instances failed and rerun only those with "Clear" on the failed map indexes.

#Data-aware scheduling

Producer DAG declares outlets:

from airflow.datasets import Dataset          # Airflow 3: from airflow.sdk import Asset
orders_ds = Dataset("s3://lake/silver/orders")

@dag(schedule="0 * * * *", start_date=datetime(2026, 10, 1), catchup=False)
def build_orders():
    @task(outlets=[orders_ds])
    def write_orders(): ...
    write_orders()
build_orders()

Consumer DAG schedules on it:

@dag(schedule=[orders_ds], start_date=datetime(2026, 10, 1), catchup=False)
def refresh_reporting():
    ...

The consumer runs after each successful update of the dataset. With several datasets in the list, it runs when all have been updated since its last run. Conditional logic (orders_ds | customers_ds, &) is available in Airflow 2.9+.

Combine time and data (Airflow 2.9+):

from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

schedule = DatasetOrTimeSchedule(timetable=CronTriggerTimetable("0 7 * * *", timezone="UTC"), datasets=[orders_ds])

#What datasets are not

• Airflow does not watch the storage. The event is emitted when the producer task succeeds. If another system writes the data, emit the event through the REST API instead.
• They do not carry data, only the fact of an update (plus optional extra metadata).

curl -X POST "$AIRFLOW/api/v1/datasets/events" -H "Content-Type: application/json" \
  -u "$USER:$PASS" -d '{"dataset_uri": "s3://lake/silver/orders"}'

The endpoint path differs in Airflow 3 (/api/v2/assets/events); check your version.

#Pools for shared resources

Mapping makes it easy to hit a database with 200 connections. Pools cap concurrent tasks across all DAGs:

airflow pools set warehouse_pool 16 "Shared warehouse connections"

@task(pool="warehouse_pool", pool_slots=1)
def load_file(path: str): ...

#Debugging

airflow tasks states-for-dag-run ingest_partners <run_id>
airflow tasks list ingest_partners

In the UI, the Grid view shows a mapped task with a count; open it to see each map index and its log. Dataset/asset views show producer and consumer relationships and recent events.

#Prevent it next time

• Cap every mapped task with max_active_tis_per_dag or a pool.
• Batch inputs so map lengths stay in the hundreds, not tens of thousands.
• Replace "sensor waiting for upstream DAG" chains with datasets/assets; keep sensors for external systems.
• Document each dataset URI as a contract between teams.