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

Need a 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

Need a 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

Need a 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

Need a hint?

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