Dynamic task generation with loops in Apache Airflow - Commands & Configuration

Introduction

Sometimes you need to create many similar tasks in a workflow without writing each one by hand. Dynamic task generation with loops helps you create multiple tasks automatically using a loop, saving time and reducing errors.

When you want to process multiple files with the same steps but different file names.

When you need to run the same task for different dates or time periods.

When you want to create tasks for multiple users or customers dynamically.

When your workflow needs to scale by adding tasks based on input data size.

When you want to avoid repeating similar code for each task in your pipeline.

Config File - my_dag.py

my_dag.py

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

def process_item(item):
    print(f"Processing {item}")

def create_dag():
    with DAG(dag_id='dynamic_loop_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily', catchup=False) as dag:
        items = ['file1.csv', 'file2.csv', 'file3.csv']
        tasks = []
        for item in items:
            task = PythonOperator(
                task_id=f'process_{item.replace(".", "_")}',
                python_callable=process_item,
                op_args=[item]
            )
            tasks.append(task)
        return dag

dag = create_dag()

This DAG file defines a workflow that runs daily starting from January 1, 2024.

The process_item function prints the item being processed.

Inside create_dag, a list of items is defined.

A loop creates a PythonOperator task for each item, giving each a unique task_id.

All tasks are added to the DAG and returned.

Commands

List all available DAGs to verify that the new DAG is recognized by Airflow.

Terminal

airflow dags list

Expected Output

dynamic_loop_dag

example_bash_operator

example_python_operator

Trigger the dynamic_loop_dag manually to start the workflow and run all dynamically created tasks.

Terminal

airflow dags trigger dynamic_loop_dag

Expected Output

Created <DagRun dynamic_loop_dag @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

List all tasks in the dynamic_loop_dag to see the dynamically generated task IDs.

Terminal

airflow tasks list dynamic_loop_dag

Expected Output

process_file1_csv

process_file2_csv

process_file3_csv

Test run the task process_file2_csv for the given date to verify it runs correctly and prints the expected output.

Terminal

airflow tasks test dynamic_loop_dag process_file2_csv 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {python.py:114} INFO - Processing file2.csv

[2024-06-01 12:00:00,100] {taskinstance.py:1234} INFO - Task exited with return code 0

Key Concept

If you remember nothing else from this pattern, remember: use a loop inside your DAG definition to create multiple similar tasks dynamically with unique IDs.

Common Mistakes

Using the same task_id for all tasks inside the loop.

Airflow requires unique task IDs; duplicate IDs cause errors and prevent the DAG from loading.

Generate unique task IDs by including the loop variable, like the item name, in the task_id.

Defining tasks outside the DAG context manager.

Tasks must be defined inside the DAG context to be properly registered with the DAG.

Always create tasks inside the 'with DAG(...) as dag:' block.

Not passing the loop variable as an argument to the task's callable.

All tasks would run the same code without knowing which item to process, causing incorrect behavior.

Pass the loop variable as an argument using op_args or op_kwargs to each task.

Summary

Define a list of items to process inside your DAG.

Use a loop to create a task for each item with a unique task_id.

Trigger the DAG and verify tasks run as expected with airflow CLI commands.

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