Mapped tasks for parallel processing in Apache Airflow - Commands & Configuration

Introduction

Sometimes you need to run the same task many times with different inputs. Mapped tasks let you do this easily in Airflow by running many tasks in parallel without writing each one by hand.

When you want to process a list of files one by one but at the same time to save time.

When you have many similar jobs like sending emails to multiple users and want to run them all at once.

When you want to run the same data transformation on many datasets in parallel.

When you want to split a big job into smaller parts that can run independently.

When you want to avoid writing repetitive code for tasks that only differ by input values.

Config File - mapped_tasks_dag.py

mapped_tasks_dag.py

from airflow import DAG
from airflow.decorators import task
from airflow.utils.dates import days_ago

with DAG(dag_id='mapped_tasks_example', start_date=days_ago(1), schedule_interval=None, catchup=False) as dag:

    @task
    def process_item(item):
        print(f"Processing item: {item}")
        return f"Processed {item}"

    items = ['apple', 'banana', 'cherry']

    results = process_item.expand(item=items)

This DAG defines a simple task process_item that takes one input called item. The expand method runs this task for each item in the list ['apple', 'banana', 'cherry'] in parallel. This way, the same task runs three times with different inputs.

Commands

This command lists all the DAGs Airflow knows about, so you can check if your DAG is recognized.

Terminal

airflow dags list

Expected Output

dag_id          | filepath

----------------|---------------------------------

mapped_tasks_example | /usr/local/airflow/dags/mapped_tasks_dag.py

This command starts running the DAG named 'mapped_tasks_example' immediately.

Terminal

airflow dags trigger mapped_tasks_example

Expected Output

Created <DagRun mapped_tasks_example @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

This command shows all tasks in the DAG, including the mapped task.

Terminal

airflow tasks list mapped_tasks_example

Expected Output

process_item

This command runs the 'process_item' task for the date 2024-06-01 locally to test it.

Terminal

airflow tasks test mapped_tasks_example process_item 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {taskinstance.py:XXXX} INFO - Processing item: apple

[2024-06-01 12:00:00,001] {taskinstance.py:XXXX} INFO - Task succeeded

Key Concept

Mapped tasks let you run the same task many times in parallel with different inputs using the expand() method.

Common Mistakes

Trying to pass a non-iterable value to expand()

expand() expects a list or iterable to run tasks multiple times; a single value causes an error.

Always pass a list or iterable to expand(), even if it has only one item.

Not enabling the DAG file in the Airflow DAGs folder

If the DAG file is not in the correct folder or has syntax errors, Airflow won't detect it.

Place the DAG file in the Airflow DAGs folder and check for syntax errors before triggering.

Expecting mapped tasks to run sequentially

Mapped tasks run in parallel by design, so expecting sequential output can cause confusion.

Design your workflow assuming mapped tasks run independently and in parallel.

Summary

Define a task function with @task decorator that takes one input parameter.

Use the expand() method with a list of inputs to run the task multiple times in parallel.

Trigger the DAG and verify tasks run for each input item.

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