Mapped tasks for parallel processing in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

💻 Command Output

intermediate

Output of mapped task execution in Airflow

Given the following Airflow DAG snippet using task mapping, what will be the output logs of the mapped task?

Apache Airflow

from airflow import DAG
from airflow.decorators import task
from datetime import datetime

with DAG('mapped_task_example', start_date=datetime(2023, 1, 1), schedule_interval='@daily', catchup=False) as dag:
    @task
    def multiply_by_two(number):
        return number * 2

    numbers = [1, 2, 3]
    results = multiply_by_two.expand(number=numbers)

A Task multiply_by_two fails due to unsupported input type

B Task multiply_by_two runs once with input [1, 2, 3] and outputs [2, 4, 6]

C Task multiply_by_two runs 3 times but outputs are all 0

D Task multiply_by_two runs 3 times with inputs 1, 2, 3 and outputs 2, 4, 6 respectively

💡 Hint

Mapped tasks run once per item in the input list, producing one output per run.

In Airflow, using expand() on a task creates multiple task instances, each running with one item from the input list. Here, multiply_by_two runs 3 times with inputs 1, 2, and 3, producing outputs 2, 4, and 6 respectively.

🧠 Conceptual

intermediate

Understanding task mapping behavior in Airflow

Which statement correctly describes how Airflow handles mapped tasks when the input list is empty?

A The mapped task does not run at all

B The mapped task runs infinitely until stopped

C The mapped task runs once with a None input

D The mapped task raises a runtime error

💡 Hint

Think about what happens if there is nothing to process.

If the input list to a mapped task is empty, Airflow skips running that task entirely because there are no items to map over.

❓ Troubleshoot

advanced

Troubleshooting mapped task failure due to XCom size limit

You have a mapped task that returns a very large list as output, but the task fails with an XCom size limit error. What is the best way to fix this issue?

A Split the output into smaller chunks and map over those chunks instead

B Increase the XCom size limit in Airflow configuration

C Disable XCom for the task

D Return the large list as a string instead of a list

💡 Hint

Think about how to reduce the size of data passed between tasks.

Airflow has a limit on the size of data stored in XComs. Splitting the large output into smaller chunks and mapping over those chunks reduces the size of each XCom, avoiding the limit.

🔀 Workflow

advanced

Designing a workflow with mapped tasks and dependencies

You want to create an Airflow DAG where a mapped task processes a list of files, and after all mapped tasks finish, a final task aggregates the results. Which approach correctly ensures the final task runs only after all mapped tasks complete?

A Set the final task downstream of the mapped task using final_task >> mapped_task

B Set the final task downstream of the mapped task using final_task.set_downstream(mapped_task)

C Set the final task downstream of the mapped task using mapped_task >> final_task

D Set the final task downstream of the mapped task using final_task.set_upstream(mapped_task)

💡 Hint

Remember the direction of dependencies in Airflow DAGs.

In Airflow, the operator >> sets the left task upstream of the right task. So mapped_task >> final_task means final_task runs after all mapped tasks complete.

✅ Best Practice

expert

Best practice for handling dynamic task mapping with large input lists

When using Airflow's mapped tasks with a very large input list (thousands of items), what is the best practice to avoid scheduler overload and improve performance?

A Map over the entire large list at once to maximize parallelism

B Use batch processing by splitting the input list into smaller chunks and map over each chunk sequentially

C Disable task retries to reduce scheduler load

D Increase the number of scheduler workers without changing the DAG

💡 Hint

Think about balancing parallelism and system limits.

Mapping over thousands of items at once can overwhelm the scheduler. Splitting the input into smaller batches and processing them sequentially reduces load and improves stability.

Practice

1. What is the main purpose of using mapped tasks in Airflow?

easy

A. To run many similar tasks in parallel using a list of inputs

B. To schedule tasks only once per day

C. To create a single task that runs sequentially

D. To manually trigger tasks one by one

Solution

1. Step 1: Understand mapped tasks concept

   Mapped tasks allow running multiple similar jobs at the same time using different inputs.

2. Step 2: Compare options with concept

   Only To run many similar tasks in parallel using a list of inputs describes running many similar tasks in parallel using a list of inputs, which matches mapped tasks.

3. Final Answer:

   To run many similar tasks in parallel using a list of inputs -> Option A

4. Quick Check:

   Mapped tasks = parallel jobs with input list [OK]

Hint: Mapped tasks run many jobs at once using input lists [OK]

Common Mistakes:

• Thinking mapped tasks run sequentially
• Confusing mapped tasks with scheduling frequency
• Assuming mapped tasks require manual triggers

2. Which of the following is the correct syntax to map a task with a list of inputs in Airflow?

easy

A. task.execute(inputs=[1, 2, 3])

B. task.map(input_list=[1, 2, 3])

C. task.run_parallel([1, 2, 3])

D. task.expand(input_list=[1, 2, 3])

Solution

1. Step 1: Recall the method to map tasks

   In Airflow, .expand() is used to map tasks with a list of inputs.

2. Step 2: Match syntax with options

   Only task.expand(input_list=[1, 2, 3]) uses task.expand(input_list=[1, 2, 3]), which is correct syntax.

3. Final Answer:

   task.expand(input_list=[1, 2, 3]) -> Option D

4. Quick Check:

   Use .expand() to map tasks [OK]

Hint: Use .expand() method with input list to map tasks [OK]

Common Mistakes:

• Using .map() instead of .expand()
• Calling .run_parallel() which does not exist
• Using .execute() with inputs incorrectly

3. Given the following code snippet, what will be the output of the mapped task?

def greet(name):
    return f"Hello, {name}!"

mapped_task = greet.expand(name=["Alice", "Bob", "Carol"])

medium

A. ["Hello, Alice!", "Hello, Bob!", "Hello, Carol!"]

B. "Hello, ['Alice', 'Bob', 'Carol']!"

C. ["Hello, Alice!", "Hello, Alice!", "Hello, Alice!"]

D. Error: greet function cannot be expanded

Solution

1. Step 1: Understand .expand() behavior

   The .expand() method runs the function for each item in the list separately, producing a list of results.

2. Step 2: Apply greet function to each name

   For names "Alice", "Bob", and "Carol", greet returns "Hello, Alice!", "Hello, Bob!", and "Hello, Carol!" respectively.

3. Final Answer:

   ["Hello, Alice!", "Hello, Bob!", "Hello, Carol!"] -> Option A

4. Quick Check:

   .expand() applies function to each input [OK]

Hint: Mapped tasks output list of results for each input [OK]

Common Mistakes:

• Thinking .expand() passes whole list as one input
• Expecting repeated output for first input only
• Assuming .expand() causes an error on functions

4. Identify the error in this Airflow mapped task code:

def process(item):
    return item * 2

mapped_task = process.expand(items=[1, 2, 3])

medium

A. The function cannot return a value

B. The argument name in expand() should match the function parameter

C. expand() cannot be used with lists

D. The function must be decorated with @task

Solution

1. Step 1: Check function parameter and expand argument

   The function parameter is named item, but expand uses items. They must match.

2. Step 2: Understand expand() argument matching

   expand() requires the argument name to match the function parameter exactly to map inputs correctly.

3. Final Answer:

   The argument name in expand() should match the function parameter -> Option B

4. Quick Check:

   expand() argument name = function parameter [OK]

Hint: expand() argument name must match function parameter exactly [OK]

Common Mistakes:

• Using different names for expand() argument and function parameter
• Thinking function cannot return values
• Believing expand() can't use lists
• Forgetting @task decorator (not always required)

5. You want to process a list of user IDs with a mapped task, but only for IDs greater than 10. Which code correctly applies this filter before mapping?

def process_user(user_id):
    return f"Processed {user_id}"

user_ids = [5, 12, 7, 20, 15]

# Which line correctly maps filtered user_ids?

hard

A. mapped_task = process_user.expand(user_id=user_ids if user_ids > 10 else [])

B. mapped_task = process_user.expand(user_id=user_ids.filter(lambda x: x > 10))

C. mapped_task = process_user.expand(user_id=[uid for uid in user_ids if uid > 10])

D. mapped_task = process_user.expand(user_id=user_ids.map(lambda x: x > 10))

Solution

1. Step 1: Filter user_ids list correctly

   Use a list comprehension to select IDs greater than 10: [uid for uid in user_ids if uid > 10].

2. Step 2: Pass filtered list to expand()

   Pass the filtered list as the argument to expand() matching the function parameter name.

3. Final Answer:

   mapped_task = process_user.expand(user_id=[uid for uid in user_ids if uid > 10]) -> Option C

4. Quick Check:

   Filter with list comprehension before expand() [OK]

Hint: Filter list with comprehension before .expand() [OK]

Common Mistakes:

• Using .filter() or .map() which are not list methods
• Trying to compare list directly to number
• Passing boolean list instead of filtered values