Why XCom enables task communication in Apache Airflow - Visual Breakdown

Process Flow - Why XCom enables task communication

Task A runs

↓

Task A pushes XCom

↓

XCom stores data

↓

Task B starts

↓

Task B pulls XCom

↓

Task B uses data

↓

Task B completes

Task A runs and pushes data to XCom, which stores it. Task B later pulls this data from XCom to use it, enabling communication between tasks.

Execution Sample

Apache Airflow

def task_a(ti):
    ti.xcom_push(key='result', value='data from A')

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

Task A pushes data to XCom; Task B pulls and prints it.

Process Table

Step	Action	XCom State	Output
1	Task A runs	{}	No output
2	Task A pushes XCom with key='result', value='data from A'	{"result": "data from A"}	No output
3	Task B starts	{"result": "data from A"}	No output
4	Task B pulls XCom with key='result' from task_a	{"result": "data from A"}	"data from A"
5	Task B prints pulled data	{"result": "data from A"}	data from A
6	Task B completes	{"result": "data from A"}	No output

💡 Task B completes after successfully pulling and using data from XCom

Status Tracker

Variable	Start	After Step 2	After Step 4	Final
XCom	{}	{"result": "data from A"}	{"result": "data from A"}	{"result": "data from A"}
data (in task_b)	N/A	N/A	"data from A"	"data from A"

Key Moments - 2 Insights

Why does Task B need to specify task_ids='task_a' when pulling from XCom?

Because XCom stores data per task, Task B must specify which task's data it wants to pull. See execution_table step 4 where task_b pulls with task_ids='task_a'.

Can Task A push multiple values with the same key to XCom?

No, each key-task pair stores one value. Pushing again with the same key overwrites the previous value. This is why in execution_table step 2, the key 'result' holds one value.

Visual Quiz - 3 Questions

Look at the execution table, what is the XCom state after Task A pushes data?

A {"task_b": "data from A"}

B {}

C {"result": "data from A"}

D null

💡 Hint

Check execution_table row 2 under XCom State column

At which step does Task B retrieve the data from XCom?

A Step 4

B Step 5

C Step 3

D Step 6

💡 Hint

Look at execution_table rows and find when XCom pull happens

If Task A pushed data with a different key, what would Task B need to change?

A Change the task_ids parameter

B Change the key in xcom_pull to match the new key

C No change needed

D Change the print statement

💡 Hint

Refer to execution_table step 4 where key='result' is used to pull data

Concept Snapshot

XCom allows tasks to share data by pushing and pulling key-value pairs.
Task A pushes data with ti.xcom_push(key, value).
Task B pulls data with ti.xcom_pull(key, task_ids).
XCom stores data per task and key.
This enables communication between tasks in Airflow DAGs.

Full Transcript

In Airflow, tasks can communicate by using XComs. Task A runs and pushes data to XCom with a key and value. This data is stored in XCom associated with Task A. Later, Task B starts and pulls the data from XCom by specifying the key and the task id of Task A. Task B then uses this data, for example by printing it. This process allows tasks to share information and coordinate their work. The execution table shows each step: Task A pushing data, XCom storing it, Task B pulling it, and finally Task B using it. Variables like XCom and data in Task B change accordingly. Key points include that Task B must specify which task's data to pull, and that each key-task pair holds one value. This simple push-pull mechanism enables task communication in Airflow.

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