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

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

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

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")

Hint

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

Practice

1. What is the main benefit of using loops for dynamic task generation in Airflow DAGs?

easy

A. It allows tasks to run in parallel without dependencies.

B. It creates multiple tasks with unique IDs automatically.

C. It reduces the number of DAG files needed.

D. It automatically schedules tasks at different times.

Solution

1. Step 1: Understand dynamic task generation

   Dynamic task generation uses loops to create many tasks without writing each one manually.

2. Step 2: Identify the key benefit

   Each task must have a unique task_id, and loops help generate these automatically, keeping the DAG clean.

3. Final Answer:

   It creates multiple tasks with unique IDs automatically. -> Option B

4. Quick Check:

   Loops generate unique tasks = C [OK]

Hint: Loops create unique tasks automatically [OK]

Common Mistakes:

• Thinking loops schedule tasks differently
• Confusing task creation with scheduling
• Assuming loops reduce DAG files only

2. Which of the following is the correct way to define a dynamic task inside a loop in Airflow using the TaskFlow API?

easy

A. for i in range(3): task = task_function(task_id=f'task_{i}')

B. for i in range(3): task = task_function(task_id='task')

C. for i in range(3): task = task_function.override(task_id=f'task_{i}')()

D. for i in range(3): task = task_function.override(task_id='task')()

Solution

1. Step 1: Recall TaskFlow API dynamic task syntax

   To create dynamic tasks, use the override method with unique task_id inside the loop and call the function.

2. Step 2: Check each option

   for i in range(3): task = task_function.override(task_id=f'task_{i}')() correctly uses override with unique task_id and calls the function. Others miss calling or unique IDs.

3. Final Answer:

   for i in range(3): task = task_function.override(task_id=f'task_{i}')() -> Option C

4. Quick Check:

   Use override + unique task_id + call = A [OK]

Hint: Use override() with unique task_id inside loop [OK]

Common Mistakes:

• Not calling the function after override
• Using same task_id for all tasks
• Missing override method

3. Given the following code snippet inside an Airflow DAG, what tasks will be created?

tasks = []
for day in ['mon', 'tue', 'wed']:
    task = BashOperator(
        task_id=f'print_{day}',
        bash_command=f'echo {day}'
    )
    tasks.append(task)

medium

A. Tasks with IDs: print_mon, print_tue, print_wed

B. One task with ID: print_day

C. Tasks with IDs: print_0, print_1, print_2

D. No tasks created due to syntax error

Solution

1. Step 1: Analyze the loop and task_id

   The loop iterates over ['mon', 'tue', 'wed'], creating a task with task_id 'print_' plus the day string.

2. Step 2: Identify the task IDs created

   Tasks created are 'print_mon', 'print_tue', and 'print_wed'.

3. Final Answer:

   Tasks with IDs: print_mon, print_tue, print_wed -> Option A

4. Quick Check:

   Loop days + task_id f-string = D [OK]

Hint: Check loop values used in task_id f-string [OK]

Common Mistakes:

• Assuming numeric indexes used in task_id
• Confusing variable name with string literal
• Thinking syntax error occurs here

4. Identify the error in this dynamic task generation code snippet:

for i in range(3):
    task = BashOperator(
        task_id='task_i',
        bash_command='echo Hello'
    )

medium

A. task_id is not unique for each task

B. bash_command syntax is incorrect

C. Missing parentheses after BashOperator

D. Range function is used incorrectly

Solution

1. Step 1: Check task_id uniqueness

   The task_id is set as the literal string 'task_i' for all tasks, not using the loop variable i.

2. Step 2: Understand impact of non-unique task_id

   Airflow requires unique task_ids; repeating 'task_i' causes conflicts or errors.

3. Final Answer:

   task_id is not unique for each task -> Option A

4. Quick Check:

   Unique task_id needed = B [OK]

Hint: Use f-string with loop variable for unique task_id [OK]

Common Mistakes:

• Thinking bash_command is wrong
• Missing calling BashOperator (it's a class, needs parentheses)
• Assuming range usage is incorrect

5. You want to dynamically create tasks for multiple file names and set dependencies so each task runs after the previous one. Which code snippet correctly achieves this?

hard

A. prev_task = None
for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']:
task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}')
task >> prev_task
prev_task = task

B. for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']:
task = BashOperator(task_id='process_file', bash_command=f'cat {file}')
task.set_upstream(prev_task)

C. tasks = []
for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']:
task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}')
tasks.append(task)
tasks[0] >> tasks[1] >> tasks[2]

D. prev_task = None
for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']:
task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}')
if prev_task:
prev_task >> task
prev_task = task

Solution

1. Step 1: Understand dependency chaining

   To run tasks sequentially, each task must depend on the previous one using '>>' operator.

2. Step 2: Analyze each option

   prev_task = None for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']: task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}') if prev_task: prev_task >> task prev_task = task correctly sets prev_task >> task and updates prev_task. for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']: task = BashOperator(task_id='process_file', bash_command=f'cat {file}') task.set_upstream(prev_task) uses undefined prev_task. tasks = [] for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']: task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}') tasks.append(task) tasks[0] >> tasks[1] >> tasks[2] chains after loop, but tasks list is not linked sequentially. prev_task = None for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']: task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}') task >> prev_task prev_task = task reverses dependency causing wrong order.

3. Final Answer:

   prev_task = None for file in ['a.txt', 'b.txt', 'c.txt', 'd.txt']: task = BashOperator(task_id=f'process_{file}', bash_command=f'cat {file}') if prev_task: prev_task >> task prev_task = task -> Option D

4. Quick Check:

   Use prev_task >> task in loop for sequence = A [OK]

Hint: Chain tasks with prev_task >> current_task inside loop [OK]

Common Mistakes:

• Using same task_id for all tasks
• Setting dependencies in wrong order
• Not updating prev_task inside loop