Dynamic task generation with loops in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

💻 Command Output

intermediate

Output of dynamically generated Airflow tasks with loop

Given the following Airflow DAG snippet, what will be the output when the DAG is parsed and tasks are listed?

Apache Airflow

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from datetime import datetime

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

print([task.task_id for task in tasks])

A ['task_0', 'task_1', 'task_2']

B ['task_1', 'task_2', 'task_3']

C ['task_0', 'task_1']

D ['task_0', 'task_1', 'task_2', 'task_3']

💡 Hint

Remember that range(3) generates numbers 0, 1, 2.

The loop runs 3 times with i values 0, 1, and 2, creating tasks with those IDs.

🔀 Workflow

intermediate

Correct order of dynamic task dependencies in Airflow

Arrange the following code lines to correctly create a chain of tasks dynamically in Airflow using a loop.

A 4,2,1,3

B 4,1,3,2

C 4,1,2,3

D 1,4,2,3

💡 Hint

The loop must start first, then create task, set dependency, then update previous_task.

The loop starts, creates a task, sets dependency from previous_task to current task, then updates previous_task for next iteration.

❓ Troubleshoot

advanced

Error caused by dynamic task creation in Airflow

What error will this Airflow DAG code raise when parsing, and why? from airflow import DAG from airflow.operators.dummy import DummyOperator from datetime import datetime with DAG('error_dag', start_date=datetime(2024, 1, 1)) as dag: for i in range(3): task = DummyOperator(task_id='task')

A No error, DAG parses successfully

B SyntaxError: invalid syntax

C TypeError: DummyOperator() missing required positional argument

D AirflowDagValidationError: Duplicate task_id 'task'

💡 Hint

Task IDs must be unique in an Airflow DAG.

All tasks have the same task_id 'task', which is not allowed and causes a validation error.

🧠 Conceptual

advanced

Best practice for dynamic task creation in Airflow

Which statement best describes a best practice when generating tasks dynamically in Airflow using loops?

A Always assign unique task_ids to each task to avoid conflicts.

B Reuse the same task_id for all tasks to simplify the DAG.

C Create tasks outside the DAG context manager to improve performance.

D Avoid using loops; create each task manually for clarity.

💡 Hint

Think about how Airflow identifies tasks uniquely.

Unique task_ids are required for Airflow to distinguish tasks and avoid errors.

✅ Best Practice

expert

Handling dynamic task dependencies with variable task counts

You want to create a dynamic chain of tasks in Airflow where the number of tasks is not fixed and depends on external input. Which approach ensures the chain is correctly built regardless of the number of tasks?

A Create all tasks first, then outside the loop set dependencies manually between tasks by their index.

B Initialize a variable previous_task as None before the loop, then inside the loop, if previous_task is not None, set previous_task >> current_task, and update previous_task to current_task.

C Use a fixed number of tasks and ignore the external input to avoid complexity.

D Set all tasks to run in parallel without dependencies to simplify the DAG.

💡 Hint

Think about how to link tasks dynamically when you don't know how many there will be.

Initializing previous_task as None and updating it inside the loop allows chaining tasks dynamically regardless of count.

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