Pushing and pulling XCom values in Apache Airflow - Step-by-Step Execution

Process Flow - Pushing and pulling XCom values

Task A runs

↓

Push XCom value

↓

Task B waits

↓

Pull XCom value

↓

Task B uses value

This flow shows how one task pushes a value to XCom and another task pulls it later to use.

Execution Sample

Apache Airflow

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

def task_b(ti):
    val = ti.xcom_pull(key='my_key', task_ids='task_a')
    print(val)

Task A pushes 'hello' to XCom with key 'my_key'. Task B pulls it and prints.

Process Table

Step	Action	XCom State	Output
1	Task A starts	{}
2	Task A pushes key='my_key', value='hello'	{"task_a": {"my_key": "hello"}}
3	Task A finishes	{"task_a": {"my_key": "hello"}}
4	Task B starts	{"task_a": {"my_key": "hello"}}
5	Task B pulls key='my_key' from task_a	{"task_a": {"my_key": "hello"}}	hello
6	Task B prints pulled value	{"task_a": {"my_key": "hello"}}	hello
7	Task B finishes	{"task_a": {"my_key": "hello"}}

💡 Task B finishes after pulling and printing the XCom value.

Status Tracker

Variable	Start	After Step 2	After Step 5	Final
XCom	{}	{"task_a": {"my_key": "hello"}}	{"task_a": {"my_key": "hello"}}	{"task_a": {"my_key": "hello"}}
val (in task_b)	None	None	'hello'	'hello'

Key Moments - 2 Insights

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

Because XCom values are stored per task, Task B must specify which task's XCom to pull. See execution_table step 5 where task_b pulls from task_a.

What happens if Task A never pushes an XCom value?

Task B's pull will return None or empty, so the variable 'val' will be None. This is shown by the initial state in variable_tracker before step 5.

Visual Quiz - 3 Questions

Look at the execution table at step 2, what is the XCom state?

A {"task_b": {"my_key": "hello"}}

B {}

C {"task_a": {"my_key": "hello"}}

D None

💡 Hint

Check the 'XCom State' column at step 2 in the execution_table.

At which step does Task B get the value 'hello' from XCom?

A Step 5

B Step 3

C Step 4

D Step 7

💡 Hint

Look at the 'Output' column in execution_table where 'hello' appears.

If Task A pushed value 'world' instead of 'hello', what would Task B print?

A hello

B world

C None

D Error

💡 Hint

Refer to variable_tracker for 'val' after step 5 and imagine the pushed value changed.

Concept Snapshot

XCom lets tasks share small data.
Task pushes with ti.xcom_push(key, value).
Other tasks pull with ti.xcom_pull(key, task_ids).
Pull needs task_ids to find the right task's data.
XCom values persist between tasks in the same DAG run.

Full Transcript

In Airflow, tasks can share data using XComs. One task pushes a value with a key. Another task pulls that value by specifying the key and the task that pushed it. This example shows Task A pushing 'hello' with key 'my_key'. Task B pulls this value and prints it. The execution table traces each step: Task A starts, pushes the value, finishes; Task B starts, pulls the value, prints it, and finishes. The variable tracker shows how the XCom dictionary and the pulled variable 'val' change over time. Key points include the need to specify the task_id when pulling and what happens if no value is pushed. The quiz tests understanding of these steps and the data flow.

Practice

1. What is the main purpose of using xcom_push and xcom_pull in Airflow?

easy

A. To share small pieces of data between tasks in a workflow

B. To schedule tasks to run at specific times

C. To create new DAGs dynamically

D. To monitor task execution logs

Solution

1. Step 1: Understand the role of XComs

   XComs are designed to let tasks exchange small data pieces during workflow execution.

2. Step 2: Identify the functions' purpose

   xcom_push sends data, and xcom_pull retrieves it between tasks.

3. Final Answer:

   To share small pieces of data between tasks in a workflow -> Option A

4. Quick Check:

   Data sharing = A [OK]

Hint: Remember: push sends data, pull gets data [OK]

Common Mistakes:

• Confusing XComs with scheduling tasks
• Thinking XComs create DAGs
• Assuming XComs handle logs

2. Which of the following is the correct syntax to push a value '42' with key 'result' inside a task instance ti?

easy

A. xcom_push(ti, 'result', 42)

B. ti.push_xcom('result', 42)

C. ti.xcom_pull(key='result', value=42)

D. ti.xcom_push(key='result', value=42)

Solution

1. Step 1: Recall the method signature

   The correct method to push data is ti.xcom_push(key=..., value=...).

2. Step 2: Check each option

   ti.xcom_push(key='result', value=42) matches the correct syntax. Options A and B use wrong method names or argument order. ti.xcom_pull(key='result', value=42) uses xcom_pull which is for pulling, not pushing.

3. Final Answer:

   ti.xcom_push(key='result', value=42) -> Option D

4. Quick Check:

   Push syntax = C [OK]

Hint: Push uses ti.xcom_push(key=..., value=...) [OK]

Common Mistakes:

• Using wrong method names like push_xcom
• Confusing push with pull
• Incorrect argument order

3. Given the following code snippet inside a task:

ti.xcom_push(key='data', value=100)
value = ti.xcom_pull(key='data')

What will be the value of the variable value after execution?

medium

A. None

B. KeyError

C. 100

D. 'data'

Solution

1. Step 1: Understand the push operation

   The code pushes the integer 100 with key 'data' into XCom storage.

2. Step 2: Understand the pull operation

   The code pulls the value stored with key 'data', which is 100.

3. Final Answer:

   100 -> Option C

4. Quick Check:

   Push 100, pull 100 = D [OK]

Hint: Pull returns the value pushed with the same key [OK]

Common Mistakes:

• Expecting pull to return None if key exists
• Confusing key with value
• Assuming pull raises error if key exists

4. You wrote this code inside a task:

ti.xcom_push('result', 10)
value = ti.xcom_pull(key='result')

But value is always None. What is the likely problem?

medium

A. The xcom_push call is missing keyword arguments

B. The key 'result' does not exist in XCom

C. You must call xcom_pull before xcom_push

D. XComs cannot store integers

Solution

1. Step 1: Check the xcom_push syntax

   The method requires keyword arguments: key= and value=. Passing positional arguments causes failure.

2. Step 2: Understand the effect of wrong syntax

   Because the push failed silently, the key 'result' was never stored, so pull returns None.

3. Final Answer:

   The xcom_push call is missing keyword arguments -> Option A

4. Quick Check:

   Push needs key= and value= [OK]

Hint: Always use key= and value= in xcom_push [OK]

Common Mistakes:

• Using positional arguments in xcom_push
• Assuming pull before push order matters
• Thinking XComs can't store integers

5. You want to pass a dictionary from one task to another using XCom. Which code snippet correctly pushes and pulls the dictionary {'a': 1, 'b': 2}?

hard

A. ti.xcom_push('my_dict', {'a': 1, 'b': 2})
data = ti.xcom_pull('my_dict')

B. ti.xcom_push(key='my_dict', value={'a': 1, 'b': 2})
data = ti.xcom_pull(key='my_dict')

C. ti.xcom_push(value={'a': 1, 'b': 2}, key='my_dict')
data = ti.xcom_pull(key='my_dict')

D. ti.xcom_push(key='my_dict', value='{'a': 1, 'b': 2}')
data = ti.xcom_pull(key='my_dict')

Solution

1. Step 1: Verify correct xcom_push syntax with dictionary

   ti.xcom_push(key='my_dict', value={'a': 1, 'b': 2}) data = ti.xcom_pull(key='my_dict') uses correct keyword arguments key= and value= with a dictionary value.

2. Step 2: Check other options for errors

   ti.xcom_push(value={'a': 1, 'b': 2}, key='my_dict') data = ti.xcom_pull(key='my_dict') uses non-standard keyword argument order; ti.xcom_push('my_dict', {'a': 1, 'b': 2}) data = ti.xcom_pull('my_dict') uses positional args which is incorrect; ti.xcom_push(key='my_dict', value='{'a': 1, 'b': 2}') data = ti.xcom_pull(key='my_dict') passes dictionary as a string with wrong quotes causing syntax error.

3. Final Answer:

   ti.xcom_push(key='my_dict', value={'a': 1, 'b': 2}) data = ti.xcom_pull(key='my_dict') -> Option B

4. Quick Check:

   Push dict with key= and value= [OK]

Hint: Use key= and value= with dict, avoid string quotes errors [OK]

Common Mistakes:

• Using positional arguments in xcom_push
• Passing dict as a string with wrong quotes
• Mixing argument order incorrectly