Mapped tasks for parallel processing in Apache Airflow - Cheat Sheet & Quick Revision

Recall & Review

beginner

What is the main purpose of mapped tasks in Airflow?

Mapped tasks allow you to run the same task multiple times in parallel with different inputs, making workflows more efficient and scalable.

beginner

How do you define a mapped task in Airflow using Python?

You use the .expand() method on a task to pass a list of inputs, which creates multiple task instances running in parallel.

intermediate

What happens if the input list for a mapped task is empty?

No task instances are created, so the mapped task effectively skips execution.

intermediate

Can mapped tasks in Airflow handle dynamic input sizes at runtime?

Yes, mapped tasks can dynamically adjust the number of parallel runs based on the input list size determined at runtime.

beginner

What is a key benefit of using mapped tasks over manually creating multiple similar tasks?

Mapped tasks reduce code duplication and make workflows easier to maintain by automatically generating parallel task instances.

Which Airflow method is used to create mapped tasks for parallel processing?

A .repeat()

B .map()

C .parallelize()

D .expand()

The .expand() method is used to pass a list of inputs to a task, creating multiple parallel task instances.

What happens if you pass an empty list to a mapped task's .expand() method?

A No task instances are created

B The task runs once with no input

C The task fails with an error

D The task runs infinitely

If the input list is empty, no task instances are created and the mapped task is skipped.

Mapped tasks in Airflow help to:

A Run the same task multiple times in parallel

B Reduce the number of tasks in a DAG

C Run tasks sequentially

D Automatically retry failed tasks

Mapped tasks run the same task multiple times in parallel with different inputs.

Which of the following is NOT a benefit of mapped tasks?

A Reduce code duplication

B Automatically fix task errors

C Handle dynamic input sizes

D Improve workflow scalability

Mapped tasks do not automatically fix errors; they help with parallelism and code simplicity.

In Airflow, mapped tasks are best suited for:

A Tasks that must run once

B Tasks that require manual triggering

C Tasks with varying inputs that can run in parallel

D Tasks that depend on external APIs only

Mapped tasks are ideal for running the same task multiple times with different inputs in parallel.

Explain how mapped tasks improve parallel processing in Airflow and give an example of when to use them.

Think about running the same job many times with different data.

Describe what happens internally when you use the .expand() method on a task in Airflow.

Focus on how Airflow creates multiple tasks from one.

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