Practical Guide to Using Airflow XComs for Inter‑Task Data Transfer

Problem: Sharing Data Between Tasks in Airflow

When building a DAG, it’s common to need a downstream task to consume the result of an upstream task—think of a data extraction step that produces a small result set, and a transformation step that consumes it. Airflow’s built‑in mechanism for this is XComs (short for “cross‑communication”). XComs let tasks push and pull arbitrary key‑value pairs without touching external storage. However, many teams misunderstand their limits and how to configure them for production workloads.

Thesis: Use XComs for Small, Structured Payloads and Fall Back to External Stores for Big Data

Airflow stores XCom values in the metadata database. By default the values are JSON‑serialized, which keeps the interface simple but also imposes a practical size ceiling. For payloads that stay under a few kilobytes, XComs are fast, reliable, and fully integrated with the UI and CLI. If you need to move megabytes or more, you should serialize a reference (e.g., an S3 key) and push that via XCom, letting the heavy lift happen in an external system.

Section 1 – Basic Push & Pull

Below is a minimal DAG that demonstrates pushing a dictionary from one task and pulling it in another. The example uses the PythonOperator and the op_kwargs argument to pass the result.

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

with DAG(
    "xcom_demo",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:

    def push_func(**kwargs):
        # Push a small JSON object
        kwargs["ti"].xcom_push(key="user_info", value={"name": "Alice", "age": 30})

    def pull_func(**kwargs):
        # Pull the value pushed earlier
        user = kwargs["ti"].xcom_pull(key="user_info", task_ids="push_task")
        print(f"Pulled user: {user}")

    push_task = PythonOperator(
        task_id="push_task",
        python_callable=push_func,
    )

    pull_task = PythonOperator(
        task_id="pull_task",
        python_callable=pull_func,
    )

    push_task >> pull_task

Run the DAG once and then inspect the XCom value in the Airflow UI under the XCom tab for push_task. You can also fetch it from the CLI:

airflow xcom get --dag-id xcom_demo --task-id push_task --key user_info

Section 2 – Custom XCom Serialization

When you need to store non‑JSON data (e.g., a NumPy array or a custom class), register a custom serializer. The serializer must implement serialize_value and deserialize_value methods.

# my_xcom_serializer.py
import json

class PickleSerializer:
    @staticmethod
    def serialize_value(value, **kwargs):
        return json.dumps(value).encode("utf-8")

    @staticmethod
    def deserialize_value(result, **kwargs):
        return json.loads(result)

Add the serializer to airflow.cfg:

[xcom]
serializer = my_xcom_serializer.PickleSerializer

After updating the config, restart the scheduler and workers. This change is global; all XCom pushes will now use the new serializer.

Section 3 – Performance Considerations

Because XComs live in the metadata database, large payloads can:

• Increase query latency for get_xcom calls.
• Grow the xcom table quickly, impacting backup and maintenance.
• Cause timeouts in the scheduler when pulling many large values.

Airflow recommends keeping XCom payloads under 4 KB for predictable performance. If you need to pass a file, upload it to S3, GCS, or Redis, and push only the URL or key.

Section 4 – Diagnostic Checklist

When an XCom pull returns None or the value looks truncated, verify:

1. Task Instance State: The upstream task must have succeeded. Failed tasks do not store XComs.
2. Correct Keys: XCom keys are case‑sensitive. Use the same string in push and pull.
3. Database Health: Inspect the xcom table size with a SQL client.

   SELECT COUNT(*) FROM xcom;
   

4. Serializer Consistency: If you changed the serializer, ensure all workers use the same version.

Trade‑Off: In‑Memory vs External Store

• In‑Memory (XCom) – Fast for small data, fully integrated, no extra cost. • External Store (S3, Redis) – Handles large objects, scales horizontally, but adds network latency and cost.

Choosing between them hinges on payload size, frequency of transfers, and the need for atomicity. For a typical ETL pipeline that moves a few hundred rows between extraction and transformation, XComs are sufficient. For a data lake ingestion that writes gigabytes of logs, an external store is preferable.

Actionable Closing

1. Keep XCom payloads <4 KB and use the UI or CLI to verify values. 2. For larger data, store the object externally and push a reference via XCom. 3. Monitor the xcom table size and scheduler logs for performance warnings. 4. If you need custom serialization, register it early and restart all Airflow services.

By following these guidelines, you’ll avoid common pitfalls and keep your DAGs efficient and maintainable.