Airflow: Dynamic Task Mapping and Dynamic DAGs

Dynamic Task Mapping

Dynamic task mapping lets Airflow generate tasks at runtime based on the output of another task. Instead of writing a fixed number of tasks, you write a single task and tell Airflow to run it once per item in a list.

Two things make this different from a Python for loop:

• Each mapped instance is a separate Airflow task with its own logs, status, and retry behavior.
• The list can come from an upstream task's return value. It does not need to be known when the DAG file is written.

Creating the DAG File

In VS Code, navigate to the Airflow directory, then to the DAGs folder. Create a file called mapped_dag.py.

Only two imports are required:

from airflow.decorators import DAG, task
from datetime import datetime

• DAG and task come from the TaskFlow API for decorator-based DAGs.
• datetime sets the start_date.

Defining the First Task

The first task returns a list of file names:

@task
def get_files():
    files = ["sales.csv", "users.csv", "events.csv", "logs.csv"]
    print(f"Number of files: {len(files)}")
    return files

Defining the Mapped Task

The second task is written as if it only handles a single item:

@task
def process_file(file_name):
    print(f"Processing {file_name}")
    return f"{file_name} done"

Building the DAG

@DAG(
    dag_id="mapped_dag",
    start_date=datetime(2023, 1, 1),
    schedule=None,
    catchup=False
)
def my_dag():
    files = get_files()
    process_file.expand(file_name=files)

mapped_dag = my_dag()

Key fields:

• schedule=None – the DAG runs only when manually triggered.
• catchup=False – no backfilling.
• process_file.expand(file_name=files) – tells Airflow to map process_file over the files list. One task instance is created per element.

The line mapped_dag = my_dag() registers the DAG with Airflow.

Running the DAG

Open the Airflow UI, search for mapped_dag, and trigger a single run.

Expected behavior:

• get_files runs once and returns a list of four file names.
• process_file generates four task instances, one for each file.

Click on process_file in the graph view, then open Task Instances. Each instance is indexed:

• Index 0 – processes sales.csv
• Index 1 – processes users.csv
• Index 2 – processes events.csv
• Index 3 – processes logs.csv