Airflow: Dynamic Task Mapping and Dynamic DAGs

Defining the first task

We start by creating a file called mapped_dag.py.
The imports are:

from airflow.decorators import DAG, task

We also print the number of files so we can see activity in the logs:

@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

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

Building the DAG

@dag(
    dag_id="mapped_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
)
def mapped_dag():

Inside the DAG function, we call get_files() and store its result in the files variable:

    files = get_files()
    process_file.expand(file_name=files)

Finally, we register the DAG by calling the function:

mapped_dag()

Running the DAG

After saving the file, go to the Airflow UI and find the mapped_dag. Trigger a single run. Click on process_file and then on "Task Instances". You will see that each task is numbered with an index in brackets.

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