Pushing and pulling XCom values in Apache Airflow - Commands & Configuration

Introduction

Sometimes tasks in Airflow need to share small pieces of data. XComs let tasks send and receive these values easily during a workflow run.

When you want one task to send a result to another task later in the workflow.

When you need to pass a status or a small data value between tasks without using external storage.

When you want to coordinate tasks by sharing simple messages or flags.

When debugging and you want to check what data a task sent to another.

When you want to avoid writing temporary files or databases for small data exchange.

Config File - xcom_example_dag.py

xcom_example_dag.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

def push_function(ti):
    ti.xcom_push(key='sample_key', value='Hello from push task')

def pull_function(ti):
    pulled_value = ti.xcom_pull(key='sample_key', task_ids='push_task')
    print(f'Pulled XCom value: {pulled_value}')

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

dag = DAG(
    'xcom_push_pull_example',
    default_args=default_args,
    schedule_interval=None,
    catchup=False
)

push_task = PythonOperator(
    task_id='push_task',
    python_callable=push_function,
    dag=dag
)

pull_task = PythonOperator(
    task_id='pull_task',
    python_callable=pull_function,
    dag=dag
)

push_task >> pull_task

This DAG defines two Python tasks. The first task push_task uses ti.xcom_push to send a value with a key. The second task pull_task uses ti.xcom_pull to get that value by key and task ID, then prints it. The tasks run in order so the pull happens after the push.

Commands

This command starts the DAG run to execute the tasks that push and pull XCom values.

Terminal

airflow dags trigger xcom_push_pull_example

Expected Output

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

→

None - No flags needed for basic trigger

Check the logs of the push task to confirm it ran and pushed the XCom value.

Terminal

airflow tasks logs xcom_push_pull_example push_task

Expected Output

Running <TaskInstance: xcom_push_pull_example.push_task manual__2024-06-01T12:00:00+00:00 [running]>

[2024-06-01 12:00:01,000] INFO - Pushed XCom key='sample_key' value='Hello from push task'

[2024-06-01 12:00:02,000] INFO - Task succeeded

Check the logs of the pull task to see the pulled XCom value printed.

Terminal

airflow tasks logs xcom_push_pull_example pull_task

Expected Output

Running <TaskInstance: xcom_push_pull_example.pull_task manual__2024-06-01T12:00:00+00:00 [running]>

Pulled XCom value: Hello from push task

[2024-06-01 12:00:03,000] INFO - Task succeeded

Key Concept

If you remember nothing else from this pattern, remember: XComs let tasks share small data by pushing with a key and pulling by that key and task ID.

Common Mistakes

Trying to pull an XCom value before the push task has run.

The pull task will get None or fail because the value does not exist yet.

Set task dependencies so the push task runs before the pull task.

Using different keys or task IDs when pushing and pulling XCom values.

The pull will not find the value and return None.

Use the exact same key and correct task ID when pulling the XCom.

Trying to pass large data or complex objects via XCom.

XCom is designed for small data; large data can cause performance issues or errors.

Use external storage like databases or files for large data, and only use XCom for small messages.

Summary

Define a PythonOperator task that pushes a value using ti.xcom_push with a key.

Define another PythonOperator task that pulls the value using ti.xcom_pull with the same key and task ID.

Trigger the DAG and check task logs to verify the push and pull of XCom values.

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