Implementing Dynamic Task Mapping in Apache Airflow

Traditional Airflow DAGs require the number of tasks to be fixed when the file is parsed. If you need to process a list of files or database rows that is only known once the DAG starts running, generating tasks in a Python loop at parse time leads to scheduler bloat and DAGs that cannot adapt to the actual workload. Dynamic task mapping, introduced in Airflow 2.3, solves this by letting a single task definition fan out into many task instances at runtime.

The takeaway: use .expand() so the instance count is decided when the run reaches the task, keeping your DAG file lightweight while execution scales with the data.

Prerequisites

• Apache Airflow 2.3 or newer. Confirm with airflow version on the scheduler host or check the UI footer.
• A functioning scheduler and an executor that can run parallel task instances (LocalExecutor, CeleryExecutor, or KubernetesExecutor).
• An XCom backend and metadata database sized for the lists you plan to map over, because the producer's list travels through XCom. Very large lists in the default database-backed XCom can bloat the metadata DB and slow scheduling.

The Producer-Consumer Pattern

Structure the workflow as a producer task that returns a list and a consumer task that expands over it. The list length determines how many mapped instances run.

1. Define the producer task

The producer is typically an @task callable returning a Python list (often a list of dicts).

from airflow.decorators import task

@task
def get_files_to_process():
    # In a real scenario, query an API or list an S3 bucket
    return ["file_a.csv", "file_b.csv", "file_c.csv"]

2. Define the consumer with .expand()

Use .partial() for arguments shared by all instances and .expand() for the argument that varies per instance.

@task
def process_file(filename):
    print(f"Processing {filename}")

files_list = get_files_to_process()
process_file.expand(filename=files_list)

Keep per-instance work idempotent: retries and clearing re-execute individual mapped indices, so non-idempotent side effects can be applied twice.

Mapping Multiple Arguments

If the consumer needs several dynamic parameters, return a list of dictionaries and pass it via op_kwargs (for classic operators) or matching keyword arguments (for @task functions). Dictionary keys are matched to parameter names.

@task
def get_configs():
    return [
        {"id": 1, "priority": "high"},
        {"id": 2, "priority": "low"},
    ]

@task
def run_analysis(id, priority):
    print(f"Running task {id} with priority {priority}")

configs = get_configs()
run_analysis.expand(op_kwargs=configs)

Note that not every operator parameter is safely mappable; some operators validate or template arguments in ways that reject mapped values. Check the documentation for your installed minor version, since mapping capabilities (multiple mapped arguments, reduce patterns, dataset interactions) grew across the 2.x line.

Verification and UI Inspection

After triggering a test run, verify the fan-out:

• Grid view: the mapped task shows an instance count matching the list length (e.g., process_file [3]).
• Indexing: each instance carries an index in its task_id, such as process_file[0].
• Logs: open a specific index's log to confirm it received the correct element from the producer's list.
• CLI/DB check: list task instance states for the run (or inspect the task_instance table) to verify counts and outcomes.
• Downstream reduce: a downstream task should run only after all mapped instances finish.

Recovery and Error Handling

• Partial failure: in the Grid view, select only the failed index and clear it. Airflow retries just that instance without restarting the whole mapped set.
• Stale input list: if the producer logic or data changed, clear the producer task. This re-resolves the list and remaps the downstream instances.
• Throttling: mapped instances consume pool slots and are governed by max_active_tis_per_dag. Set it to throttle large fan-outs so one DAG does not starve others, and consider capping or batching list length to protect the scheduler and database.

Why Mapping Beats Parse-Time Loops

Prefer .expand() over generating tasks in a loop at parse time: mapped counts are resolved at runtime, so the DAG file stays small and parsing stays fast regardless of workload size. The trade-off is that you must manage XCom payload size and concurrency limits yourself.