Airflow XComs

XComs (short for "cross-communications") let one task share a small piece of data with another task in the same DAG run. Without XComs, each task runs in isolation and cannot see what another task produced.

Why Tasks Need to Communicate

Scenario: A file download pipeline
──────────────────────────────────────────────
Task 1: Download a file → gets the filename "sales_2024_01_15.csv"
Task 2: Process that file → needs to know the filename

Problem: Task 2 does not know what Task 1 named the file.
Solution: Task 1 pushes the filename as an XCom.
          Task 2 pulls that filename from XCom.
──────────────────────────────────────────────

How XComs Work

XCom Flow:
┌───────────────────────────────────────────────────┐
│                                                   │
│  Task 1                   Task 2                  │
│  ────────              ────────────               │
│  Runs first            Runs second                │
│  Returns "hello"  →→→  Pulls "hello" from XCom    │
│  (push to DB)          (pull from DB)             │
│                                                   │
│         Airflow Database (XCom store)             │
│         ┌────────────────────────────┐            │
│         │ key="return_value"         │            │
│         │ value="hello"              │            │
│         │ dag_id="my_dag"            │            │
│         │ task_id="task_1"           │            │
│         │ run_id="run_2024_01_15"    │            │
│         └────────────────────────────┘            │
└───────────────────────────────────────────────────┘

Push and Pull: The Two XCom Operations

Push (Sending Data)

A task pushes data by returning a value from its function (automatic push) or by calling xcom_push explicitly.

Automatic Push (Return Value)

def get_filename():
    filename = "sales_2024_01_15.csv"
    return filename  # ← Airflow automatically pushes this to XCom

Airflow stores the return value in XCom with the key "return_value".

Manual Push (xcom_push)

def get_multiple_values(**context):
    ti = context["ti"]  # ti = task instance
    ti.xcom_push(key="filename",    value="sales_2024_01_15.csv")
    ti.xcom_push(key="record_count", value=1523)

Manual push lets you store multiple named values from a single task.

Pull (Receiving Data)

def process_file(**context):
    ti = context["ti"]

    # Pull the return value from get_filename task
    filename = ti.xcom_pull(task_ids="get_filename")

    # Pull a specific named key from get_multiple_values task
    record_count = ti.xcom_pull(task_ids="get_multiple_values", key="record_count")

    print(f"Processing {filename} with {record_count} records")

Full XCom Example

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def download_file(**context):
    # Simulates downloading a file and returning its name
    filename = "sales_2024_01_15.csv"
    print(f"Downloaded: {filename}")
    return filename  # auto-pushed to XCom

def process_file(**context):
    ti = context["ti"]
    filename = ti.xcom_pull(task_ids="download_file")
    print(f"Processing file: {filename}")

def send_report(**context):
    ti = context["ti"]
    filename = ti.xcom_pull(task_ids="download_file")
    print(f"Sending report based on: {filename}")

with DAG("xcom_demo", start_date=datetime(2024, 1, 1),
         schedule="@daily", catchup=False) as dag:

    t1 = PythonOperator(task_id="download_file",  python_callable=download_file)
    t2 = PythonOperator(task_id="process_file",   python_callable=process_file)
    t3 = PythonOperator(task_id="send_report",    python_callable=send_report)

    t1 >> t2 >> t3

Data Flow:
[download_file] → XCom DB (filename="sales_2024_01_15.csv")
                       ↓
              [process_file] ← pulls filename
                       ↓
              [send_report]  ← pulls filename

Viewing XComs in the UI

Go to Admin → XComs. You see a table of all stored XCom values with their DAG ID, task ID, run ID, key, and value. This helps you debug — you can verify exactly what one task passed to the next.

XComs and the Jinja Templating Shortcut

You can pull XComs directly inside templated fields using Jinja syntax — no Python code needed:

from airflow.operators.bash import BashOperator

process = BashOperator(
    task_id="process",
    bash_command="echo Processing {{ ti.xcom_pull(task_ids='download_file') }}",
)

XCom Limitations

XComs store data in Airflow's database. They are designed for small values — filenames, IDs, status codes, short strings, small numbers.

For large data, write it to a shared storage location (S3, GCS, a database) and pass only the file path or record ID via XCom.

Custom XCom Backend

In Airflow 2.0+, you can replace the default database XCom storage with a custom backend — for example, storing XCom values in S3 or GCS instead of the Airflow database. This removes the size limitation for large data values. Set up a custom backend in airflow.cfg:

[core]
xcom_backend = airflow.providers.amazon.aws.xcom_backends.s3.S3XComBackend

XComs in TaskFlow API

The TaskFlow API (covered in a later topic) handles XCom push and pull automatically through function return values and parameters — no manual xcom_push or xcom_pull calls needed.