Mapped tasks for parallel processing in Apache Airflow - Step-by-Step Execution

Process Flow - Mapped tasks for parallel processing

Define base task

↓

Create list of inputs

↓

Map base task over inputs

↓

Run tasks in parallel

↓

Collect results

↓

Complete DAG

This flow shows how a single task is defined and then mapped over multiple inputs to run many tasks in parallel, speeding up processing.

Execution Sample

Apache Airflow

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

with DAG('mapped_tasks_dag', start_date=datetime(2024,1,1)) as dag:
    @task
    def process_item(item):
        return item * 2

    items = [1, 2, 3]
    results = process_item.expand(item=items)

This Airflow DAG defines a task that doubles a number, then runs that task in parallel for each number in the list.

Process Table

Step	Action	Input	Task Instance	Output	Notes
1	Define task function	None	process_item	Function ready	Task process_item is defined
2	Create input list	None	N/A	[1, 2, 3]	List of items to process
3	Map task over inputs	[1, 2, 3]	process_item[1]	Scheduled	Task instance for item=1 created
4	Map task over inputs	[1, 2, 3]	process_item[2]	Scheduled	Task instance for item=2 created
5	Map task over inputs	[1, 2, 3]	process_item[3]	Scheduled	Task instance for item=3 created
6	Run task instance	1	process_item[1]	2	Item 1 doubled to 2
7	Run task instance	2	process_item[2]	4	Item 2 doubled to 4
8	Run task instance	3	process_item[3]	6	Item 3 doubled to 6
9	Collect results	N/A	N/A	[2, 4, 6]	All mapped tasks completed
10	DAG complete	N/A	N/A	Success	All tasks finished successfully

💡 All mapped task instances finished, DAG run completes successfully

Status Tracker

Variable	Start	After Step 2	After Step 3-5	After Step 6-8	Final
items	None	[1, 2, 3]	[1, 2, 3]	[1, 2, 3]	[1, 2, 3]
results	None	None	None	[2, 4, 6]	[2, 4, 6]

Key Moments - 3 Insights

Why do we see multiple task instances with the same task name but different inputs?

Because the task is mapped over a list of inputs, Airflow creates separate task instances for each input to run them in parallel, as shown in steps 3 to 5.

How does Airflow know when all mapped tasks are done?

Airflow tracks each task instance separately and considers the mapped task complete only when all instances finish successfully, as seen in step 9.

What happens if one mapped task instance fails?

The DAG run will be marked as failed or retry depending on configuration, because all mapped tasks must succeed for the DAG to complete, implied by the exit note.

Visual Quiz - 3 Questions

Look at the execution table, what is the output of process_item[2] at step 7?

A 6

B 2

C 4

D None

💡 Hint

Check row with Step 7 in execution_table under Output column

At which step are all mapped task instances scheduled but not yet run?

A Step 5

B Step 2

C Step 6

D Step 9

💡 Hint

Look at steps 3 to 5 in execution_table where tasks are scheduled

If the input list changed to [1, 2, 3, 4], how would the execution table change?

A The DAG would complete earlier

B There would be an extra mapped task instance for item=4

C The output of process_item[3] would be 8

D No change in task instances

💡 Hint

Mapping creates one task instance per input item, see steps 3-5

Concept Snapshot

Mapped tasks let you run the same task code on many inputs at once.
Define a task function, create a list of inputs, then use expand() to map.
Airflow runs each mapped task instance in parallel.
All instances must finish for the DAG to succeed.
This speeds up processing by using parallelism.

Full Transcript

Mapped tasks in Airflow allow running the same task multiple times with different inputs in parallel. First, you define a base task function. Then, you create a list of inputs. Using the expand() method, Airflow creates separate task instances for each input. These run at the same time, speeding up the workflow. The DAG completes only after all mapped tasks finish successfully. This visual trace showed each step from defining the task, scheduling mapped instances, running them, and collecting results.

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