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

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

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

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

Hint

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

Practice

1. What is the main purpose of XCom in Airflow?

easy

A. To store large files generated by tasks

B. To schedule tasks at specific times

C. To monitor the health of Airflow workers

D. To enable tasks to share small pieces of data between each other

Solution

1. Step 1: Understand XCom's role in Airflow

   XCom is designed to let tasks exchange small data easily during a workflow.

2. Step 2: Compare options with XCom's purpose

   Scheduling tasks, monitoring workers, or storing large files are not XCom's functions.

3. Final Answer:

   To enable tasks to share small pieces of data between each other -> Option D

4. Quick Check:

   XCom = task data sharing [OK]

Hint: XCom is for small data sharing between tasks [OK]

Common Mistakes:

• Confusing XCom with scheduling features
• Thinking XCom stores large files
• Mixing XCom with monitoring tools

2. Which Airflow method is used to send data from one task to another using XCom?

easy

A. xcom_pull()

B. xcom_push()

C. send_data()

D. push_xcom()

Solution

1. Step 1: Identify the method to send data in XCom

   The method to send data is called xcom_push().

2. Step 2: Verify other options

   xcom_pull() is for receiving data, and the others are not valid Airflow methods.

3. Final Answer:

   xcom_push() -> Option B

4. Quick Check:

   Send data = xcom_push() [OK]

Hint: Push data with xcom_push(), pull with xcom_pull() [OK]

Common Mistakes:

• Confusing xcom_push() with xcom_pull()
• Using non-existent methods like send_data()
• Mixing method names

3. Given the following code snippet in Airflow tasks:

def task_a(ti):
    ti.xcom_push(key='result', value=42)

def task_b(ti):
    value = ti.xcom_pull(key='result', task_ids='task_a')
    print(value)

What will task_b print when executed?

medium

A. 42

B. None

C. task_a

D. Error

Solution

1. Step 1: Understand xcom_push and xcom_pull usage

   task_a pushes the value 42 with key 'result'. task_b pulls this value using the same key and task_id.

2. Step 2: Determine the printed output

   Since task_b pulls the value 42 correctly, it will print 42.

3. Final Answer:

   42 -> Option A

4. Quick Check:

   Push 42, pull 42 = print 42 [OK]

Hint: Pull what was pushed with matching key and task_id [OK]

Common Mistakes:

• Expecting None if keys don't match
• Confusing task_ids parameter
• Assuming print outputs task name

4. In the code below, why does task_b print None instead of the expected value?

def task_a(ti):
    ti.xcom_push(key='data', value='hello')

def task_b(ti):
    value = ti.xcom_pull(key='result', task_ids='task_a')
    print(value)

medium

A. xcom_pull requires a different method signature

B. The task_ids parameter is incorrect

C. The key used in xcom_pull does not match the key used in xcom_push

D. xcom_push cannot send strings

Solution

1. Step 1: Compare keys used in push and pull

   task_a pushes with key 'data', but task_b pulls with key 'result'. This mismatch causes None to be returned.

2. Step 2: Verify other parameters

   task_ids is correct, xcom_push supports strings, and method signature is valid.

3. Final Answer:

   The key used in xcom_pull does not match the key used in xcom_push -> Option C

4. Quick Check:

   Matching keys needed for XCom data [OK]

Hint: Keys must match exactly in push and pull [OK]

Common Mistakes:

• Ignoring key name mismatch
• Assuming xcom_push can't send strings
• Changing task_ids unnecessarily

5. You want to pass a list of numbers from task_a to task_b using XCom. Which approach correctly enables this communication?

hard

A. In task_a: ti.xcom_push(key='numbers', value=[1, 2, 3])In task_b: nums = ti.xcom_pull(key='numbers', task_ids='task_a')

B. In task_a: ti.xcom_push(key='numbers', value='1,2,3')In task_b: nums = ti.xcom_pull(key='numbers', task_ids='task_b')

C. In task_a: ti.xcom_push(key='data', value=[1, 2, 3])In task_b: nums = ti.xcom_pull(key='numbers', task_ids='task_a')

D. In task_a: ti.xcom_push(value=[1, 2, 3])In task_b: nums = ti.xcom_pull(key='numbers', task_ids='task_a')

Solution

1. Step 1: Check key and value consistency

   In task_a: ti.xcom_push(key='numbers', value=[1, 2, 3])
   In task_b: nums = ti.xcom_pull(key='numbers', task_ids='task_a') uses the same key 'numbers' in push and pull, and passes a list correctly.

2. Step 2: Verify task_ids and value types

   Other options fail: one pushes a string '1,2,3' instead of list and pulls from wrong task_id 'task_b'; another mismatches keys 'data' vs 'numbers'; another omits key in push (defaults to 'return_value') but uses key='numbers' in pull.

3. Final Answer:

   In task_a: ti.xcom_push(key='numbers', value=[1, 2, 3])
   In task_b: nums = ti.xcom_pull(key='numbers', task_ids='task_a') -> Option A

4. Quick Check:

   Matching keys and task_ids with correct value type [OK]

Hint: Use same key and task_id, push list directly [OK]

Common Mistakes:

• Mismatching keys between push and pull
• Pulling from wrong task_id
• Omitting key parameter in push or pull