Dynamic task generation with loops in Apache Airflow - Mini Project: Build & Apply

Dynamic task generation with loops in Airflow

📖 Scenario: You are working as a data engineer using Apache Airflow to automate data processing tasks. You want to create multiple similar tasks dynamically instead of writing each task manually. This will save time and reduce errors.

🎯 Goal: Build an Airflow DAG that dynamically generates three Python tasks using a for loop. Each task will print a unique message identifying itself.

📋 What You'll Learn

Create a DAG with the id dynamic_task_dag

Use a for loop to generate three PythonOperator tasks

Each task must have a unique task_id in the format print_task_1, print_task_2, and print_task_3

Each task should run a Python function that prints Task number X is running where X is the task number

Set the DAG to run once with schedule_interval=None

💡 Why This Matters

🌍 Real World

In real projects, dynamically generating tasks helps automate workflows that have repetitive steps, like processing multiple files or datasets.

💼 Career

Understanding dynamic task generation is essential for Airflow users to build scalable and maintainable data pipelines efficiently.

1

Create the DAG and import modules

Write code to import DAG from airflow and PythonOperator from airflow.operators.python. Then create a DAG object called dynamic_task_dag with schedule_interval=None and start_date set to datetime(2024, 1, 1). Import datetime from datetime as well.

Apache Airflow

# Your code here

Need a hint?

Remember to import DAG, PythonOperator, and datetime. Then create the DAG with the exact id and parameters.

2

Define the Python function to print task number

Define a Python function called print_task_number that takes one argument task_num and prints the message Task number {task_num} is running using an f-string.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

dynamic_task_dag = DAG(
    dag_id='dynamic_task_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
)

# Define the function print_task_number here
# Your code here

Need a hint?

Use def print_task_number(task_num): and inside print the message with an f-string.

3

Create tasks dynamically using a for loop

Use a for loop with variable i in range(1, 4) to create three PythonOperator tasks. Each task should have task_id set to print_task_{i}, use print_task_number as the python_callable, and pass op_kwargs={'task_num': i}. Add each task to the dynamic_task_dag.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

dynamic_task_dag = DAG(
    dag_id='dynamic_task_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
)

def print_task_number(task_num):
    print(f"Task number {task_num} is running")

# Create tasks dynamically using a for loop
# Your code here

Need a hint?

Use a for loop from 1 to 3. Inside, create a PythonOperator with the correct task_id, python_callable, and op_kwargs.

4

Print confirmation message

Write a print statement that outputs the exact text Dynamic tasks created successfully.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

dynamic_task_dag = DAG(
    dag_id='dynamic_task_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
)

def print_task_number(task_num):
    print(f"Task number {task_num} is running")

for i in range(1, 4):
    task = PythonOperator(
        task_id=f'print_task_{i}',
        python_callable=print_task_number,
        op_kwargs={'task_num': i},
        dag=dynamic_task_dag
    )

# Print confirmation message
print("Dynamic tasks created successfully")

Need a hint?

Use print("Dynamic tasks created successfully") exactly.