Dynamic task generation with loops in Apache Airflow - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to create a dynamic task inside the DAG using a loop.

Apache Airflow

for i in range(3):
    task = PythonOperator(task_id=f'task_[1]', python_callable=my_function, dag=dag)

Drag options to blanks, or click blank then click option'

Ai

Bindex

Ctask

Dcount

Attempts:

3 left

💡 Hint

Common Mistakes

Using a variable name not defined in the loop like 'index' or 'count'.

Forgetting to use the loop variable to create unique task IDs.

The variable i is used in the loop to generate unique task IDs dynamically.

2fill in blank

medium

Complete the code to add tasks to the DAG dynamically using a loop and list comprehension.

Apache Airflow

tasks = [PythonOperator(task_id=f'task_[1]', python_callable=my_function, dag=dag) for [1] in range(5)]

Drag options to blanks, or click blank then click option'

Acount

Btask

Cindex

Di

Attempts:

3 left

💡 Hint

Common Mistakes

Using a variable name not defined in the comprehension like 'task' or 'count'.

Not matching the loop variable in both places.

The loop variable i is used inside the list comprehension to create multiple tasks with unique IDs.

3fill in blank

hard

Fix the error in the dynamic task creation loop by completing the missing part.

Apache Airflow

for i in range(3):
    task = PythonOperator(task_id=f'task_[1]', python_callable=my_function, dag=dag)
    task.set_downstream(next_task)

Drag options to blanks, or click blank then click option'

Atask

Bi

Cnext_task

Dindex

Attempts:

3 left

💡 Hint

Common Mistakes

Using a fixed string for task_id causing duplicate task IDs.

Using an undefined variable in the f-string.

The loop variable i must be used to create unique task IDs to avoid task ID conflicts.

4fill in blank

hard

Fill both blanks to create tasks dynamically and set dependencies between them.

Apache Airflow

tasks = []
for [1] in range(3):
    task = PythonOperator(task_id=f'task_[1]', python_callable=my_function, dag=dag)
    tasks.append(task)
for i in range(2):
    tasks[i].[2](tasks[i+1])

Drag options to blanks, or click blank then click option'

Ai

Bset_upstream

Cset_downstream

Dappend

Attempts:

3 left

💡 Hint

Common Mistakes

Using set_upstream instead of set_downstream for dependencies.

Not using the loop variable for unique task IDs.

The loop variable i is used to create unique task IDs and set_downstream sets the order of task execution.

5fill in blank

hard

Fill all three blanks to create a dictionary of tasks dynamically with conditions.

Apache Airflow

tasks = {f'task_[1]': PythonOperator(task_id=f'task_[1]', python_callable=my_function, dag=dag) for [2] in range(5) if [3] % 2 == 0}

Drag options to blanks, or click blank then click option'

Ai

Bj

Attempts:

3 left

💡 Hint

Common Mistakes

Using different variables inconsistently.

Not applying the modulo condition correctly.

The variable i is used for task IDs and loop iteration, and the condition filters even numbers.

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