How to loop over a list of values in Airflow (dynamic task mapping)

Dynamic task mapping turns one task into as many copies as there are items in a list. The list can be written in the DAG or returned by an upstream task, so a run can create three copies today and twelve tomorrow. Each copy runs, retries and logs on its own.

Copy and paste

One task per item of a fixed list:

from airflow.sdk import task

@task
def export_region(region: str):
    print(f"Exporting {region}")

export_region.expand(region=["eu", "us", "apac"])

A list computed at run time:

@task
def list_regions() -> list[str]:
    return ["eu", "us", "apac"]      # from a query, an API, a folder...

export_region.expand(region=list_regions())

A fixed argument next to the mapped one:

@task
def export_region(region: str, run_date: str) -> int:
    ...
    return rows_exported

counts = export_region.partial(run_date="{{ ds }}").expand(region=list_regions())

Collect the results:

@task
def total(counts) -> int:
    return sum(counts)

total(counts)

Classic operators:

from airflow.providers.standard.operators.bash import BashOperator

BashOperator.partial(task_id="load_file").expand(
    bash_command=["python load.py a.csv", "python load.py b.csv"],
)

Several arguments that change together:

@task
def copy(src: str, dst: str):
    ...

copy.expand_kwargs([
    {"src": "a.csv", "dst": "archive/a.csv"},
    {"src": "b.csv", "dst": "archive/b.csv"},
])

Readable names in the UI

Mapped copies appear as 0, 1, 2. With map_index_template they show a label you choose, here the region:

from airflow.sdk import task, get_current_context

@task(map_index_template="{{ region_label }}")
def export_region(region: str):
    get_current_context()["region_label"] = region
    ...

Limits and edge cases

Parallelism. max_active_tis_per_dag=2 in @task(...) or .partial(...) caps how many copies run at the same time, useful when they all hit the same database or API.
Long lists. A list longer than 1,024 items fails the task that returns it. Raise the limit with AIRFLOW__CORE__MAX_MAP_LENGTH, or map over batches instead of single items.
Empty lists. With an empty list the mapped task is skipped. Tasks after it follow their trigger rules, so with the default rule they are skipped as well.
Results. The downstream task receives the results as a lazy sequence in the order of the map index. It supports iteration, indexing, len() and sum(). Call list() on it for anything else.
Templates. Template strings in partial() and expand() are rendered for every copy: run_date="{{ ds }}" above becomes the date of the run.

Why not a for loop

A for loop in the DAG file creates tasks when the file is parsed, from a list that has to be known at that moment. Every change to the list changes the shape of the DAG, and the list cannot come from data the run produces. Mapping creates the copies at run time, after the upstream task has returned the list.

Tested with Airflow 3.3.2.

References

Airflow dynamic task mapping