Dynamic task generation with loops in Apache Airflow - Time & Space Complexity

Time Complexity: Dynamic task generation with loops

O(n)

Understanding Time Complexity

When we create tasks dynamically in Airflow using loops, the number of tasks depends on the loop size.

We want to understand how the total work grows as we add more tasks.

Scenario Under Consideration

Analyze the time complexity of the following code snippet.

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

def generate_tasks(dag, task_count):
    for i in range(task_count):
        BashOperator(
            task_id=f'task_{i}',
            bash_command='echo Hello',
            dag=dag
        )

with DAG('example_dag', start_date=datetime(2024, 1, 1)) as dag:
    generate_tasks(dag, 5)

This code creates a number of BashOperator tasks inside a DAG using a loop.

Identify Repeating Operations

Identify the loops, recursion, array traversals that repeat.

• Primary operation: Loop creating tasks
• How many times: Equal to task_count times

How Execution Grows With Input

Each new task adds one more operation to create and register it.

Input Size (n)	Approx. Operations
10	10 task creations
100	100 task creations
1000	1000 task creations

Pattern observation: The work grows directly with the number of tasks.

Final Time Complexity

Time Complexity: O(n)

This means the time to create tasks grows linearly as you add more tasks.

Common Mistake

[X] Wrong: "Creating tasks in a loop is instant no matter how many tasks there are."

[OK] Correct: Each task creation takes some time, so more tasks mean more total time.

Interview Connect

Understanding how loops affect task creation helps you explain how Airflow scales with many tasks.

Self-Check

"What if we nested loops to create tasks inside tasks? How would the time complexity change?"

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