Pushing and pulling XCom values in Apache Airflow - Mini Project: Build & Apply

Pushing and Pulling XCom Values in Airflow

📖 Scenario: You are working with Apache Airflow to automate a simple data pipeline. You want to pass information between tasks using XComs, which are a way for tasks to share small amounts of data.Imagine you have two tasks: one task generates a message, and the next task reads that message and prints it.

🎯 Goal: Build an Airflow DAG with two Python tasks. The first task pushes a message to XCom. The second task pulls that message from XCom and prints it.

📋 What You'll Learn

Create a DAG named xcom_example_dag with default arguments

Create a Python task called push_task that pushes the string 'Hello from push_task' to XCom

Create a Python task called pull_task that pulls the XCom value pushed by push_task and prints it

Set the task dependency so that push_task runs before pull_task

💡 Why This Matters

🌍 Real World

In real data pipelines, tasks often need to share small pieces of data like file paths, status messages, or computed values. XComs provide a simple way to pass this data between tasks.

💼 Career

Understanding XComs is essential for Airflow users and data engineers to build modular, maintainable workflows that communicate between steps effectively.

1

Create the DAG and push task

Create an Airflow DAG named xcom_example_dag with start_date set to datetime(2024, 1, 1) and schedule_interval set to @daily. Then create a Python function called push_function that returns the string 'Hello from push_task'. Finally, create a PythonOperator called push_task that uses push_function as its callable and belongs to the DAG.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# Create the DAG and push task below
# Your code here

Hint

Use with DAG(...) to create the DAG. Define a function that returns the message. Use PythonOperator with task_id and python_callable.

2

Create the pull task function

Inside the same DAG, create a Python function called pull_function that accepts ti as a parameter. Use ti.xcom_pull(task_ids='push_task') to get the message pushed by push_task and store it in a variable called message. Then print the message.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(dag_id='xcom_example_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily') as dag:
    def push_function():
        return 'Hello from push_task'

    push_task = PythonOperator(
        task_id='push_task',
        python_callable=push_function
    )

    # Create pull_function below
    # Your code here

Hint

Define a function with ti parameter. Use ti.xcom_pull(task_ids='push_task') to get the message. Print the message.

3

Create the pull task operator and set dependencies

Create a PythonOperator called pull_task that uses pull_function as its callable and belongs to the DAG. Then set the task dependency so that push_task runs before pull_task using the bitshift operator.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(dag_id='xcom_example_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily') as dag:
    def push_function():
        return 'Hello from push_task'

    push_task = PythonOperator(
        task_id='push_task',
        python_callable=push_function
    )

    def pull_function(ti):
        message = ti.xcom_pull(task_ids='push_task')
        print(message)

    # Create pull_task and set dependencies below
    # Your code here

Hint

Create a PythonOperator for pull_task with pull_function. Use push_task >> pull_task to set the order.

4

Run the DAG and verify output

Run the DAG and observe the logs of pull_task. The output should print the message Hello from push_task that was pushed by push_task.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(dag_id='xcom_example_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily') as dag:
    def push_function():
        return 'Hello from push_task'

    push_task = PythonOperator(
        task_id='push_task',
        python_callable=push_function
    )

    def pull_function(ti):
        message = ti.xcom_pull(task_ids='push_task')
        print(message)

    pull_task = PythonOperator(
        task_id='pull_task',
        python_callable=pull_function
    )

    push_task >> pull_task

# Run the DAG and check the logs of pull_task to see the output
# Your code here

Hint

Trigger the DAG run in Airflow UI or CLI. Check the logs of pull_task to see the printed message.

Practice

1. What is the main purpose of using xcom_push and xcom_pull in Airflow?

easy

A. To share small pieces of data between tasks in a workflow

B. To schedule tasks to run at specific times

C. To create new DAGs dynamically

D. To monitor task execution logs

Solution

1. Step 1: Understand the role of XComs

   XComs are designed to let tasks exchange small data pieces during workflow execution.

2. Step 2: Identify the functions' purpose

   xcom_push sends data, and xcom_pull retrieves it between tasks.

3. Final Answer:

   To share small pieces of data between tasks in a workflow -> Option A

4. Quick Check:

   Data sharing = A [OK]

Hint: Remember: push sends data, pull gets data [OK]

Common Mistakes:

• Confusing XComs with scheduling tasks
• Thinking XComs create DAGs
• Assuming XComs handle logs

2. Which of the following is the correct syntax to push a value '42' with key 'result' inside a task instance ti?

easy

A. xcom_push(ti, 'result', 42)

B. ti.push_xcom('result', 42)

C. ti.xcom_pull(key='result', value=42)

D. ti.xcom_push(key='result', value=42)

Solution

1. Step 1: Recall the method signature

   The correct method to push data is ti.xcom_push(key=..., value=...).

2. Step 2: Check each option

   ti.xcom_push(key='result', value=42) matches the correct syntax. Options A and B use wrong method names or argument order. ti.xcom_pull(key='result', value=42) uses xcom_pull which is for pulling, not pushing.

3. Final Answer:

   ti.xcom_push(key='result', value=42) -> Option D

4. Quick Check:

   Push syntax = C [OK]

Hint: Push uses ti.xcom_push(key=..., value=...) [OK]

Common Mistakes:

• Using wrong method names like push_xcom
• Confusing push with pull
• Incorrect argument order

3. Given the following code snippet inside a task:

ti.xcom_push(key='data', value=100)
value = ti.xcom_pull(key='data')

What will be the value of the variable value after execution?

medium

A. None

B. KeyError

C. 100

D. 'data'

Solution

1. Step 1: Understand the push operation

   The code pushes the integer 100 with key 'data' into XCom storage.

2. Step 2: Understand the pull operation

   The code pulls the value stored with key 'data', which is 100.

3. Final Answer:

   100 -> Option C

4. Quick Check:

   Push 100, pull 100 = D [OK]

Hint: Pull returns the value pushed with the same key [OK]

Common Mistakes:

• Expecting pull to return None if key exists
• Confusing key with value
• Assuming pull raises error if key exists

4. You wrote this code inside a task:

ti.xcom_push('result', 10)
value = ti.xcom_pull(key='result')

But value is always None. What is the likely problem?

medium

A. The xcom_push call is missing keyword arguments

B. The key 'result' does not exist in XCom

C. You must call xcom_pull before xcom_push

D. XComs cannot store integers

Solution

1. Step 1: Check the xcom_push syntax

   The method requires keyword arguments: key= and value=. Passing positional arguments causes failure.

2. Step 2: Understand the effect of wrong syntax

   Because the push failed silently, the key 'result' was never stored, so pull returns None.

3. Final Answer:

   The xcom_push call is missing keyword arguments -> Option A

4. Quick Check:

   Push needs key= and value= [OK]

Hint: Always use key= and value= in xcom_push [OK]

Common Mistakes:

• Using positional arguments in xcom_push
• Assuming pull before push order matters
• Thinking XComs can't store integers

5. You want to pass a dictionary from one task to another using XCom. Which code snippet correctly pushes and pulls the dictionary {'a': 1, 'b': 2}?

hard

A. ti.xcom_push('my_dict', {'a': 1, 'b': 2})
data = ti.xcom_pull('my_dict')

B. ti.xcom_push(key='my_dict', value={'a': 1, 'b': 2})
data = ti.xcom_pull(key='my_dict')

C. ti.xcom_push(value={'a': 1, 'b': 2}, key='my_dict')
data = ti.xcom_pull(key='my_dict')

D. ti.xcom_push(key='my_dict', value='{'a': 1, 'b': 2}')
data = ti.xcom_pull(key='my_dict')

Solution

1. Step 1: Verify correct xcom_push syntax with dictionary

   ti.xcom_push(key='my_dict', value={'a': 1, 'b': 2}) data = ti.xcom_pull(key='my_dict') uses correct keyword arguments key= and value= with a dictionary value.

2. Step 2: Check other options for errors

   ti.xcom_push(value={'a': 1, 'b': 2}, key='my_dict') data = ti.xcom_pull(key='my_dict') uses non-standard keyword argument order; ti.xcom_push('my_dict', {'a': 1, 'b': 2}) data = ti.xcom_pull('my_dict') uses positional args which is incorrect; ti.xcom_push(key='my_dict', value='{'a': 1, 'b': 2}') data = ti.xcom_pull(key='my_dict') passes dictionary as a string with wrong quotes causing syntax error.

3. Final Answer:

   ti.xcom_push(key='my_dict', value={'a': 1, 'b': 2}) data = ti.xcom_pull(key='my_dict') -> Option B

4. Quick Check:

   Push dict with key= and value= [OK]

Hint: Use key= and value= with dict, avoid string quotes errors [OK]

Common Mistakes:

• Using positional arguments in xcom_push
• Passing dict as a string with wrong quotes
• Mixing argument order incorrectly