Why XCom enables task communication in Apache Airflow - Why It Works

Introduction

In Airflow, tasks often need to share information to work together smoothly. XCom is a tool that helps tasks send small pieces of data to each other, making it easier to coordinate complex workflows.

When one task produces data that another task needs to use later in the workflow

When you want to pass the result of a database query from one task to another without saving it externally

When you need to share a status or flag between tasks to control workflow logic

When you want to avoid writing temporary files for passing small data between tasks

When you want to keep your workflow clean and simple by using built-in Airflow features for communication

Commands

This command starts the example DAG that demonstrates how tasks use XCom to share data.

Terminal

airflow dags trigger example_xcom_dag

Expected Output

Created <DagRun example_xcom_dag @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

This lists all tasks in the example DAG so you can see which tasks will run and communicate using XCom.

Terminal

airflow tasks list example_xcom_dag

Expected Output

task_1

task_2

Runs task_1 alone to see how it pushes data to XCom without running the whole DAG.

Terminal

airflow tasks test example_xcom_dag task_1 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {taskinstance.py:1234} INFO - Pushing data to XCom: {'message': 'Hello from task_1'}

[2024-06-01 12:00:00,100] {taskinstance.py:5678} INFO - Task succeeded

Runs task_2 alone to see how it pulls data from XCom that task_1 pushed earlier.

Terminal

airflow tasks test example_xcom_dag task_2 2024-06-01

Expected Output

[2024-06-01 12:01:00,000] {taskinstance.py:1234} INFO - Pulled data from XCom: {'message': 'Hello from task_1'}

[2024-06-01 12:01:00,100] {taskinstance.py:5678} INFO - Task succeeded

Key Concept

If you remember nothing else, remember: XCom is Airflow's built-in way for tasks to send and receive small pieces of data to work together.

Common Mistakes

Trying to pass large files or big data through XCom

XCom is designed for small messages; large data can cause performance issues or failures

Use external storage like S3 or a database for large data and pass references via XCom

Not specifying the correct task IDs when pulling data from XCom

If the task ID is wrong, the data won't be found and the task will fail or get None

Always use the exact task ID that pushed the XCom data when pulling it

Expecting XCom data to persist indefinitely

XCom data is stored in Airflow's metadata database and can be cleaned up or overwritten

Use XCom only for short-term communication within the same DAG run

Summary

XCom lets Airflow tasks share small pieces of data easily within a workflow.

Use 'push' in one task to send data and 'pull' in another to receive it.

Avoid using XCom for large data or long-term storage; it's for quick task communication.

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