Dynamic task generation with loops in Apache Airflow - Step-by-Step Execution

Process Flow - Dynamic task generation with loops

Start DAG definition

↓

Define task list

↓

For each item in list

↓

Create task dynamically

↓

Set task dependencies

↓

DAG ready to run

This flow shows how a loop creates multiple tasks dynamically in an Airflow DAG, then sets their order.

Execution Sample

Apache Airflow

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

dag = DAG('loop_dag', start_date=datetime(2024,1,1))
tasks = []
for i in range(3):
    task = BashOperator(task_id=f'task_{i}', bash_command=f'echo {i}', dag=dag)
    tasks.append(task)

This code creates 3 BashOperator tasks dynamically in a loop inside an Airflow DAG.

Process Table

Step	Loop Index i	Action	Task Created	Task List Length
1	0	Create task with task_id 'task_0'	task_0	1
2	1	Create task with task_id 'task_1'	task_1	2
3	2	Create task with task_id 'task_2'	task_2	3
4	-	Loop ends after i=2	-	3

💡 Loop ends after i=2 because range(3) produces 0,1,2

Status Tracker

Variable	Start	After 1	After 2	After 3	Final
i	-	0	1	2	-
tasks	[]	[task_0]	[task_0, task_1]	[task_0, task_1, task_2]	[task_0, task_1, task_2]

Key Moments - 2 Insights

Why does the loop stop after creating 3 tasks?

Because the range(3) generates numbers 0, 1, and 2 only, so the loop runs exactly 3 times as shown in execution_table rows 1-3.

How do we know each task has a unique task_id?

The task_id is created using f'task_{i}', so each loop iteration uses a different i value, producing unique IDs as seen in the 'Task Created' column.

Visual Quiz - 3 Questions

Look at the execution table, what is the task_id created at step 2?

A task_2

B task_0

C task_1

D task_3

💡 Hint

Check the 'Task Created' column at step 2 in the execution_table.

At which step does the loop end according to the execution table?

A Step 4

B Step 2

C Step 3

D Step 1

💡 Hint

Look for the row where the action says 'Loop ends' in the execution_table.

If range(5) was used instead of range(3), how many tasks would be in the tasks list after the loop?

A 4

B 5

C 3

D 6

💡 Hint

Refer to variable_tracker for how tasks list length changes with loop iterations.

Concept Snapshot

Dynamic task generation in Airflow:
- Use a loop to create tasks dynamically
- Each task needs a unique task_id
- Append tasks to a list for dependency management
- Loop runs as many times as range() specifies
- Tasks are added to DAG automatically

Full Transcript

This visual execution shows how to generate multiple tasks dynamically in an Airflow DAG using a loop. The loop runs three times, creating tasks with unique IDs 'task_0', 'task_1', and 'task_2'. Each task is appended to a list for later use. The loop stops after the third iteration because range(3) produces numbers 0, 1, and 2. This method helps create many similar tasks without writing repetitive code.

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