Sharing data between tasks effectively in Apache Airflow - Mini Project: Build & Apply

Sharing data between tasks effectively

📖 Scenario: You are building a simple Airflow workflow to process daily sales data. The first task fetches sales numbers, and the second task calculates the total sales. You want to share the sales data between these tasks efficiently.

🎯 Goal: Build an Airflow DAG where one task pushes sales data, and the next task pulls that data to calculate the total sales.

📋 What You'll Learn

Create a Python dictionary called sales_data with exact entries: 'Monday': 100, 'Tuesday': 150, 'Wednesday': 200

Create a task called push_sales that pushes sales_data to XCom

Create a task called pull_and_sum_sales that pulls sales_data from XCom and calculates the total sales

Print the total sales in the pull_and_sum_sales task

💡 Why This Matters

🌍 Real World

Sharing data between tasks is common in workflows where one step produces data that the next step needs to use, like processing sales, logs, or user data.

💼 Career

Understanding XCom in Airflow is essential for building reliable data pipelines and workflows in many DevOps and data engineering roles.

1

Create the sales data dictionary

Create a Python dictionary called sales_data with these exact entries: 'Monday': 100, 'Tuesday': 150, 'Wednesday': 200

Apache Airflow

# Create the sales_data dictionary with exact entries
# Your code here

Hint

Use curly braces {} to create a dictionary with keys and values.

2

Create the push_sales task to push data to XCom

Create a Python function called push_sales that takes **kwargs and pushes the sales_data dictionary to XCom using kwargs['ti'].xcom_push(key='sales', value=sales_data)

Apache Airflow

sales_data = {'Monday': 100, 'Tuesday': 150, 'Wednesday': 200}

# Define the push_sales function to push sales_data to XCom
# Your code here

Hint

Use kwargs['ti'].xcom_push to send data to the next task.

3

Create the pull_and_sum_sales task to pull data and calculate total

Create a Python function called pull_and_sum_sales that takes **kwargs, pulls sales_data from XCom using kwargs['ti'].xcom_pull(key='sales', task_ids='push_sales'), calculates the total sales by summing the values, and stores it in a variable called total_sales

Apache Airflow

sales_data = {'Monday': 100, 'Tuesday': 150, 'Wednesday': 200}

def push_sales(**kwargs):
    kwargs['ti'].xcom_push(key='sales', value=sales_data)

# Define pull_and_sum_sales function to pull sales_data and calculate total_sales
# Your code here

Hint

Use kwargs['ti'].xcom_pull to get data from the previous task, then sum the dictionary values.

4

Print the total sales in pull_and_sum_sales task

Add a print statement inside the pull_and_sum_sales function to display the text Total sales: {total_sales} using an f-string

Apache Airflow

sales_data = {'Monday': 100, 'Tuesday': 150, 'Wednesday': 200}

def push_sales(**kwargs):
    kwargs['ti'].xcom_push(key='sales', value=sales_data)

def pull_and_sum_sales(**kwargs):
    sales = kwargs['ti'].xcom_pull(key='sales', task_ids='push_sales')
    total_sales = sum(sales.values())
    # Print the total sales below
    # Your code here

Hint

Use print(f"Total sales: {total_sales}") to show the result.

Practice

1. What is the primary purpose of using Airflow Variables when sharing data between tasks?

easy

A. To execute tasks in parallel

B. To monitor task execution logs

C. To schedule DAG runs automatically

D. To store small pieces of data accessible across different tasks

Solution

1. Step 1: Understand Airflow Variables role

   Airflow Variables are designed to hold small data values that can be shared between tasks.

2. Step 2: Identify the correct use case

   Sharing data between tasks requires a storage mechanism accessible by all tasks, which Airflow Variables provide.

3. Final Answer:

   To store small pieces of data accessible across different tasks -> Option D

4. Quick Check:

   Airflow Variables = data sharing [OK]

Hint: Variables store shared data between tasks easily [OK]

Common Mistakes:

• Confusing Variables with task scheduling
• Thinking Variables execute tasks
• Using Variables for logging

2. Which of the following is the correct way to set an Airflow Variable named my_var with value 123 inside a PythonOperator task?

easy

A. Variable.set('my_var', 123)

B. Variable.set(my_var, '123')

C. Variable.get('my_var', 123)

D. Variable.create('my_var', 123)

Solution

1. Step 1: Recall Variable.set syntax

   The correct method to set a variable is Variable.set(key, value) where key is a string and value is the data.

2. Step 2: Check option correctness

   Variable.set('my_var', 123) uses Variable.set with key 'my_var' and value 123 correctly as a string key and integer value.

3. Final Answer:

   Variable.set('my_var', 123) -> Option A

4. Quick Check:

   Set variable = Variable.set(key, value) [OK]

Hint: Use Variable.set('key', value) to store data [OK]

Common Mistakes:

• Using Variable.get instead of set to store data
• Not quoting the variable name key
• Using a non-existent method like Variable.create

3. Given the following code snippet in two tasks, what will be printed by task_b?

from airflow.models import Variable

def task_a():
    Variable.set('shared_key', 'hello')

def task_b():
    value = Variable.get('shared_key')
    print(value)

medium

A. None

B. hello

C. shared_key

D. Error: Variable not found

Solution

1. Step 1: Analyze task_a behavior

   task_a sets the Variable 'shared_key' with value 'hello'.

2. Step 2: Analyze task_b behavior

   task_b retrieves the Variable 'shared_key' and prints its value, which is 'hello'.

3. Final Answer:

   hello -> Option B

4. Quick Check:

   Set then get Variable = 'hello' [OK]

Hint: Get returns the stored value by key [OK]

Common Mistakes:

• Expecting the key name instead of value
• Assuming Variable.get returns None if set
• Forgetting to set the Variable before getting

4. You wrote this code in task_b to get a Variable set by task_a:

value = Variable.get('my_var')

But you get an error: KeyError: 'Variable my_var does not exist'. What is the best fix?

medium

A. Set a default value in Variable.get like Variable.get('my_var', default_var='default')

B. Change Variable.get to Variable.set

C. Remove the Variable.get call

D. Restart Airflow scheduler

Solution

1. Step 1: Understand the error cause

   The error means 'my_var' was not set before getting it, so Variable.get fails.

2. Step 2: Use default value to avoid error

   Variable.get supports a default_var parameter to return if key missing, preventing KeyError.

3. Final Answer:

   Set a default value in Variable.get like Variable.get('my_var', default_var='default') -> Option A

4. Quick Check:

   Use default_var to avoid missing key error [OK]

Hint: Add default_var to Variable.get to prevent errors [OK]

Common Mistakes:

• Using Variable.set instead of get to read data
• Ignoring missing Variable errors
• Restarting scheduler without fixing code

5. You want to share a list of user IDs between two tasks in Airflow. Which approach is best to share this data effectively?

hard

A. Write the list to a local file and read it in the next task

B. Pass the list directly as a Python variable between tasks

C. Use XComs to push and pull the list between tasks

D. Serialize the list as JSON string and store it in an Airflow Variable

Solution

1. Step 1: Understand data sharing options

   Airflow Variables are for small strings; passing Python variables directly is not possible between tasks.

2. Step 2: Identify best practice for complex data

   XComs are designed to share data like lists between tasks efficiently and safely.

3. Final Answer:

   Use XComs to push and pull the list between tasks -> Option C

4. Quick Check:

   XComs = best for task-to-task data sharing [OK]

Hint: Use XComs for complex data sharing between tasks [OK]

Common Mistakes:

• Trying to pass Python variables directly
• Using Variables for large or complex data
• Using local files which break portability