Mapped tasks for parallel processing in Apache Airflow - Mini Project: Build & Apply

Mapped tasks for parallel processing in Airflow

📖 Scenario: You are working with Apache Airflow to automate data processing tasks. You want to run the same task on multiple data inputs at the same time to save time.This is like having a kitchen where you want to bake many cookies at once instead of one by one.

🎯 Goal: Build an Airflow DAG that uses mapped tasks to process a list of numbers in parallel. Each task will square a number from the list.

📋 What You'll Learn

Create a list of numbers to process

Define a simple Python function to square a number

Use Airflow's task mapping to run the function on each number in the list

Print the results of each squared number

💡 Why This Matters

🌍 Real World

Mapped tasks let you run many similar jobs at the same time, like processing many files or data chunks quickly.

💼 Career

Knowing how to use mapped tasks in Airflow is useful for data engineers and DevOps professionals automating workflows efficiently.

1

Create the list of numbers to process

Create a list called numbers with these exact values: [1, 2, 3, 4, 5]

Apache Airflow

# Create the list of numbers
# Your code here

Hint

Use square brackets to create a list and separate numbers with commas.

2

Define a Python function to square a number

Define a function called square_number that takes one argument n and returns n * n

Apache Airflow

numbers = [1, 2, 3, 4, 5]

# Define the function to square a number
# Your code here

Hint

Use def to define a function and return to send back the result.

3

Create an Airflow DAG with mapped tasks

Import DAG and task from airflow.decorators. Create a DAG called square_numbers_dag with start_date as 2023-01-01. Use the @task decorator on square_number. Use square_number.expand(n=numbers) to map the task over the list numbers.

Apache Airflow

numbers = [1, 2, 3, 4, 5]

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

# Create the DAG
# Your code here

Hint

Use with DAG(...) to create the DAG context. Decorate the function with @task. Use .expand() to map the task.

4

Print the results of the mapped tasks

Add a task called print_results decorated with @task that takes results as input and prints it. Call print_results(squared_results) inside the DAG. Run the DAG and write the output you see from print_results.

Apache Airflow

numbers = [1, 2, 3, 4, 5]

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

with DAG('square_numbers_dag', start_date=datetime(2023, 1, 1), schedule_interval=None, catchup=False) as dag:
    @task
    def square_number(n):
        return n * n

    squared_results = square_number.expand(n=numbers)

# Add print_results task and call it
# Your code here

Hint

The print output will show the list of squared numbers: [1, 4, 9, 16, 25]

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