Why XCom enables task communication in Apache Airflow - See It in Action

Why XCom Enables Task Communication in Airflow

📖 Scenario: You are working with Apache Airflow to automate workflows. Sometimes, one task needs to send information to another task. Airflow uses a feature called XCom (short for cross-communication) to pass small pieces of data between tasks.Imagine you are baking a cake with friends. One friend mixes the batter and passes it to another friend to bake. XCom is like passing the batter between friends so the next step can continue.

🎯 Goal: Build a simple Airflow DAG with two tasks where the first task sends a message using XCom, and the second task receives and prints that message.

📋 What You'll Learn

Create a DAG with two PythonOperator tasks

Use XCom to push a message from the first task

Use XCom to pull the message in the second task

Print the received message in the second task

💡 Why This Matters

🌍 Real World

In real workflows, tasks often need to share results or parameters. XCom makes this easy and automatic.

💼 Career

Understanding XCom is essential for Airflow users and DevOps engineers to build reliable, connected workflows.

1

Create the DAG and first task to push a message

Create a DAG called xcom_example_dag with default arguments. Define a Python function called push_message that uses kwargs['ti'].xcom_push(key='message', value='Hello from task 1') to send the message. Create a PythonOperator task called task_push that runs push_message inside the DAG.

Apache Airflow

# Create the DAG and the first task to push a message
# Your code here

Need a hint?

Remember to import DAG and PythonOperator. Use kwargs['ti'].xcom_push inside the push_message function.

2

Add the second task to pull the message

Define a Python function called pull_message that uses message = kwargs['ti'].xcom_pull(task_ids='task_push', key='message') to get the message. Create a PythonOperator task called task_pull that runs pull_message inside the same DAG.

Apache Airflow

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

def push_message(**kwargs):
    kwargs['ti'].xcom_push(key='message', value='Hello from task 1')

default_args = {
    'start_date': datetime(2024, 1, 1)
}

with DAG('xcom_example_dag', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    task_push = PythonOperator(
        task_id='task_push',
        python_callable=push_message
    )
    # Define the second task to pull the message
    # Your code here

Need a hint?

Use kwargs['ti'].xcom_pull to get the message from task_push. Then print it.

3

Set task dependencies to run pull after push

Set the task order so that task_pull runs after task_push by writing task_push >> task_pull inside the DAG.

Apache Airflow

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

def push_message(**kwargs):
    kwargs['ti'].xcom_push(key='message', value='Hello from task 1')

def pull_message(**kwargs):
    message = kwargs['ti'].xcom_pull(task_ids='task_push', key='message')
    print(f'Received message: {message}')

default_args = {
    'start_date': datetime(2024, 1, 1)
}

with DAG('xcom_example_dag', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    task_push = PythonOperator(
        task_id='task_push',
        python_callable=push_message
    )
    task_pull = PythonOperator(
        task_id='task_pull',
        python_callable=pull_message
    )

# Set task dependencies so task_pull runs after task_push
# Your code here

Need a hint?

Use the bitshift operator >> to set the order of tasks.

4

Run the DAG and observe the output

Run the DAG and observe the output of task_pull. It should print Received message: Hello from task 1.

Apache Airflow

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

def push_message(**kwargs):
    kwargs['ti'].xcom_push(key='message', value='Hello from task 1')

def pull_message(**kwargs):
    message = kwargs['ti'].xcom_pull(task_ids='task_push', key='message')
    print(f'Received message: {message}')

default_args = {
    'start_date': datetime(2024, 1, 1)
}

with DAG('xcom_example_dag', default_args=default_args, schedule_interval=None, catchup=False) as dag:
    task_push = PythonOperator(
        task_id='task_push',
        python_callable=push_message
    )
    task_pull = PythonOperator(
        task_id='task_pull',
        python_callable=pull_message
    )

    task_push >> task_pull

# Run the DAG and check the logs of task_pull for the message
# Your code here

Need a hint?

Run the DAG in Airflow UI or CLI and check the logs of task_pull to see the printed message.