Mapped tasks for parallel processing in Apache Airflow - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to map the task over the list of numbers.

Apache Airflow

from airflow.decorators import task

@task
 def square_number(number):
     return number [1] 2

results = square_number.expand(number=[1, 2, 3, 4])

Drag options to blanks, or click blank then click option'

A**

B//

C+

D*

Attempts:

3 left

💡 Hint

Common Mistakes

Using multiplication (*) instead of power operator (**).

Using addition (+) which adds instead of squares.

The ** operator is used to raise a number to a power. Here, it squares the number.

2fill in blank

medium

Complete the code to define a mapped task that prints each item.

Apache Airflow

from airflow.decorators import task

@task
 def print_item(item):
     print(item)

print_item.expand(item=[1])

Drag options to blanks, or click blank then click option'

A["apple", "banana", "cherry"]

B"hello"

C42

Drange(5)

Attempts:

3 left

💡 Hint

Common Mistakes

Passing a single string instead of a list.

Passing a number which is not iterable.

The expand method expects an iterable. A list of strings is a good example.

3fill in blank

hard

Fix the error in the mapped task call to correctly map over the list.

Apache Airflow

from airflow.decorators import task

@task
 def multiply_by_two(x):
     return x * 2

results = multiply_by_two.expand(x=[1])

Drag options to blanks, or click blank then click option'

A"123"

B5

C[1, 2, 3]

Drange(1)

Attempts:

3 left

💡 Hint

Common Mistakes

Passing a single integer instead of a list.

Passing a string which is iterable but not intended here.

The expand method requires an iterable like a list to map the task over multiple inputs.

4fill in blank

hard

Fill both blanks to create a mapped task that filters even numbers and squares them.

Apache Airflow

from airflow.decorators import task

@task
 def process_number(n):
     if n [1] 2 == 0:
         return n[2]2

results = process_number.expand(n=[1, 2, 3, 4, 5])

Drag options to blanks, or click blank then click option'

A%

B**

C==

D+

Attempts:

3 left

💡 Hint

Common Mistakes

Using addition (+) instead of power to square.

Using equality (==) instead of modulo (%) to check evenness.

The modulo operator % checks if a number is even. The power operator ** squares the number.

5fill in blank

hard

Fill all three blanks to create a mapped task that returns a dictionary with uppercase keys and values greater than zero.

Apache Airflow

from airflow.decorators import task

@task
 def filter_and_transform(data):
     return { [1]: [2] for [3], v in data.items() if v > 0 }

results = filter_and_transform.expand(data=[{'a': 1, 'b': -1}, {'c': 3, 'd': 0}])

Drag options to blanks, or click blank then click option'

Ak.upper()

Bv

Ck

Ditem

Attempts:

3 left

💡 Hint

Common Mistakes

Using 'item' instead of 'k' for the key variable.

Not converting keys to uppercase.

Mixing up keys and values in the comprehension.

The dictionary comprehension uses k.upper() for keys, v for values, and k as the key variable in the loop.

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