Custom XCom Backends
Why large XCom payloads hurt the metadata database, and how object-storage-backed XComs keep Airflow passing references instead of blobs.
XCom is a mailbox, not a warehouse. When the bookshop’s enrich_titles task hands a fact table to load_warehouse, the honest question is what actually travels between them. Most of the time it should be a short string — a row count, a partition key, an S3 URI — and the data itself should have moved through a system built to hold data. The trouble starts the day someone returns a pandas DataFrame from a task and Airflow, being obliging, serializes the whole thing into the metadata database. It works in the demo. It works for the first month. Then the DataFrame grows, the xcom table balloons, the UI’s “XCom” tab times out trying to render a 40MB cell, and a Postgres VACUUM becomes a production event. A custom XCom backend is how you keep the mailbox behavior — the value flows through Airflow’s dependency graph like any other XCom — while moving the payload somewhere that was designed to carry it.
What XCom actually stores
By default in Airflow 3.x, an XCom value is serialized to JSON and written to a single row in the xcom table of the metadata database. That default matters for two reasons. First, JSON is deliberately conservative: it round-trips dicts, lists, strings, numbers, and booleans cleanly, and it refuses anything it can’t represent, which is a feature — it stops you from smuggling a live database handle or a NumPy array through XCom by accident. Second, JSON has no size ceiling of its own, but the column it lands in does, and every push is a write to the same shared, transactional database that the scheduler is using to decide what runs next. XCom was designed for coordination signals measured in bytes, and every property of the default backend follows from that assumption.
The 3.x detail that changes the mechanics — and that most 2.x-era writing gets wrong — is who does the writing. In Airflow 2.x a worker held a direct connection to the metadata database and inserted the XCom row itself. In 3.x, the Task Execution API brokers all of it. A worker no longer touches the metadata DB; when your task calls ti.xcom_push(...) or returns a value, the Task SDK serializes it and sends it to the api-server over the Execution API, and the api-server persists it. This is the security and isolation boundary that let Airflow 3 split the api-server from the workers in the first place — a worker running arbitrary DAG code never gets database credentials.
A custom XCom backend lives on both sides of that boundary, and understanding where each hook runs is the whole game. serialize_value runs in the worker, inside the task process, before the value crosses the Execution API. deserialize_value also runs in the worker, when a downstream task pulls the value back. And purge runs when the XCom is cleared, so it’s where you delete whatever you offloaded. Notice what is not on that list: nothing your backend writes ever runs on the api-server. The UI displays the stored row as-is. Get that mental model right and everything else is detail.
The mechanism: subclassing BaseXCom
A custom backend is a subclass of airflow.sdk.BaseXCom that overrides three methods. The contract is small:
from airflow.sdk import BaseXCom
class MyXComBackend(BaseXCom):
@staticmethod
def serialize_value(value, *, key=None, task_id=None,
dag_id=None, run_id=None, map_index=None, **kwargs):
# Runs in the WORKER. Turn `value` into something the base
# class can persist through the Execution API.
...
@staticmethod
def deserialize_value(result):
# Runs in the WORKER. `result` is the stored row; return the
# original object a downstream task expects.
...
@classmethod
def purge(cls, xcom, *args):
# Runs when an XCom is CLEARED. Delete the object you wrote
# in serialize_value — this is your garbage collector.
...Two things about that signature are easy to miss and expensive to get wrong.
serialize_value is a @staticmethod and it receives the full identity of the XCom — dag_id, run_id, task_id, key, map_index — as keyword arguments. That identity is the reason a custom backend can do anything interesting: it’s exactly the coordinate you need to build a stable, unique storage path for the payload. Accept **kwargs defensively; Airflow has added parameters to this signature across minor versions, and a backend that hard-codes the argument list will break on upgrade.
The return value of serialize_value is whatever eventually gets stored via the base class. The pattern that keeps the metadata database small is: write the big payload to object storage inside serialize_value, and return only a tiny reference (a URI string, or a small JSON envelope). The base class serializes that reference to JSON and hands it to the Execution API. The xcom row now holds a pointer measured in bytes, not the DataFrame.
purge is the hook people don’t know exists, and it’s the one that keeps your bucket from becoming a landfill. When Airflow clears an XCom — you clear a task, you clear a DAG run, a retention job sweeps old runs — it deletes the xcom row. It has no idea that row pointed at a five-megabyte Parquet file in S3. purge is where you tell it: Airflow calls it with the XCom being removed, and your job is to delete the object you wrote. Skip it and every cleared run leaves its payload behind forever, and you find out from the storage bill.
A note if you’re porting a 2.x backend. You may be looking for orm_deserialize_value — the hook that returned a cheap stand-in so the UI didn’t have to download the payload just to render a table cell. It no longer exists in Airflow 3, and the reason is the architecture change this chapter opened with. In 2.x the api-server read XComs straight out of the metadata DB, so it needed its own deserialization path and its own way of not being slow. In 3.x the worker is the only thing that ever calls deserialize_value, and the UI simply displays what’s stored in the row — which, in an offloading backend, is already the small reference string you returned from serialize_value. The cheap render isn’t a hook you have to remember any more; it falls out of the boundary for free. One less thing to get wrong, and a good example of the Task Execution API earning its keep.
A worked S3 backend for the bookshop
Here is a complete backend that offloads any large payload to S3 and leaves a URI behind. It’s written for the bookshop pipeline, where a task might legitimately produce a Parquet-sized frame that a downstream task consumes.
# include/xcom/s3_backend.py
import json
import os
import uuid
import pandas as pd
from airflow.sdk import BaseXCom
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
# A marker so we can tell "we offloaded this" from an ordinary small XCom.
_OFFLOAD_PREFIX = "s3xcom://"
class S3XComBackend(BaseXCom):
BUCKET = os.environ["XCOM_S3_BUCKET"]
AWS_CONN_ID = os.environ.get("XCOM_S3_CONN_ID", "aws_default")
# Only offload things worth offloading; small values stay inline.
THRESHOLD_BYTES = 4096
@staticmethod
def _hook():
return S3Hook(aws_conn_id=S3XComBackend.AWS_CONN_ID)
@staticmethod
def _key_for(dag_id, run_id, task_id, key, map_index):
# A path that is unique per (dag, run, task, key, mapped index),
# so a re-run or a mapped fan-out never collides with a sibling.
suffix = "" if map_index in (None, -1) else f"/{map_index}"
salt = uuid.uuid4().hex[:8]
return f"{dag_id}/{run_id}/{task_id}/{key}{suffix}/{salt}.bin"
@staticmethod
def serialize_value(value, *, key=None, task_id=None, dag_id=None,
run_id=None, map_index=None, **kwargs):
# Decide the wire format based on the payload's type.
if isinstance(value, pd.DataFrame):
body = value.to_parquet() # bytes
fmt = "parquet"
else:
try:
body = json.dumps(value).encode("utf-8")
fmt = "json"
except (TypeError, ValueError):
# Not JSON-serializable and not a frame: let the base
# class refuse it, exactly as the default backend would.
return BaseXCom.serialize_value(
value, key=key, task_id=task_id, dag_id=dag_id,
run_id=run_id, map_index=map_index, **kwargs)
# Below the threshold, keep it inline — object storage isn't free
# and tiny values belong in the DB.
if len(body) < S3XComBackend.THRESHOLD_BYTES and fmt == "json":
return BaseXCom.serialize_value(
value, key=key, task_id=task_id, dag_id=dag_id,
run_id=run_id, map_index=map_index, **kwargs)
s3_key = S3XComBackend._key_for(dag_id, run_id, task_id, key, map_index)
S3XComBackend._hook().load_bytes(
body, key=s3_key, bucket_name=S3XComBackend.BUCKET, replace=True)
# This tiny envelope is what actually lands in the metadata DB.
reference = {
"uri": f"{_OFFLOAD_PREFIX}{S3XComBackend.BUCKET}/{s3_key}",
"format": fmt,
"bytes": len(body),
}
return BaseXCom.serialize_value(
reference, key=key, task_id=task_id, dag_id=dag_id,
run_id=run_id, map_index=map_index, **kwargs)
@staticmethod
def deserialize_value(result):
raw = BaseXCom.deserialize_value(result)
if not (isinstance(raw, dict)
and str(raw.get("uri", "")).startswith(_OFFLOAD_PREFIX)):
# A small inline value that was never offloaded — return as-is.
return raw
path = raw["uri"][len(_OFFLOAD_PREFIX):]
bucket, _, s3_key = path.partition("/")
body = S3XComBackend._hook().get_key(
key=s3_key, bucket_name=bucket).get()["Body"].read()
if raw["format"] == "parquet":
import io
return pd.read_parquet(io.BytesIO(body))
return json.loads(body.decode("utf-8"))
@classmethod
def purge(cls, xcom, *args):
# Called when this XCom is CLEARED. Delete what we offloaded,
# or the bucket grows forever.
raw = cls.deserialize_value(xcom) if xcom.value else None
uri = raw.get("uri") if isinstance(raw, dict) else None
if uri and str(uri).startswith(_OFFLOAD_PREFIX):
bucket, key = _split(uri)
cls._hook().delete_objects(bucket=bucket, keys=[key])Note what the UI shows for an offloaded value with no extra work from you: the small JSON envelope you returned from serialize_value — the URI, the byte count, the format. That’s exactly the cheap render you want, and you get it because the api-server displays the stored row and never calls deserialize_value. In Airflow 2 you had to implement a separate hook to achieve that; in Airflow 3 the worker/api-server boundary hands it to you.
Wire it up with a single environment variable, in the Astro project’s .env or your deployment’s config:
AIRFLOW__CORE__XCOM_BACKEND=include.xcom.s3_backend.S3XComBackend
XCOM_S3_BUCKET=bookshop-xcom
XCOM_S3_CONN_ID=aws_defaultRestart the scheduler, api-server, and workers so all three load the class. Now a bookshop task can return frame and downstream tasks receive a DataFrame, while the metadata database only ever sees s3xcom://bookshop-xcom/etl_bookshop/manual__2026-06-14.../enrich_titles/return_value/3f9a1c22.bin.
Trace the flow once, end to end, and the 3.x model clicks: enrich_titles runs in a worker, serialize_value writes the Parquet bytes to S3 from the worker, and only the URI envelope crosses the Task Execution API to the api-server, which stores that string. When load_warehouse starts, its worker pulls the XCom, deserialize_value reads the bytes back from S3, and reconstitutes the frame. The metadata database never held the data, and the worker never held database credentials. The custom backend slots cleanly into the brokered model precisely because the payload takes a different road than the coordination signal — the api-server brokers the pointer; S3 carries the freight.
Cleanup: purge is the hook
Airflow deletes the xcom rows when it clears a DAG run’s XComs, and — via purge — it will delete your S3 object too, but only if you implement it. Left unimplemented, it has no idea your row pointed at an S3 object — that byte-blob is now orphaned storage, quietly accruing cost. There is no built-in reaper for offloaded payloads, and pretending otherwise is how a “cheap” backend turns into a surprise bill.
Two defenses, used together. First, an S3 lifecycle policy on the bucket: expire objects after N days. Because the key path starts with {dag_id}/{run_id}/, you can even scope lifecycle rules per DAG if some pipelines need longer retention than others. This is the durable safety net — it runs whether or not Airflow is healthy. Second, if you need prompt cleanup tied to run lifecycle, a small teardown task (the setup/teardown pattern from the previous chapter) or an on_success_callback can delete the run’s prefix once downstream consumers are done. Lifecycle policy is the floor; explicit teardown is the optimization. Never rely on the explicit path alone — a failed run skips its own teardown, and those are exactly the runs whose debris you’ll forget.
Serialization: JSON, pickle, or Parquet
The backend above branches on type deliberately, because the right wire format depends on what you’re moving:
- JSON is the safe default for structured, non-tabular values — dicts, lists, config-shaped payloads. It’s portable, human-readable in S3, and version-stable. Its limits are its virtues: no dates, no NumPy scalars, no arbitrary objects.
- Parquet is the right call for DataFrames. It’s columnar, compressed, typed, and — critically — it preserves dtypes across the round trip in a way
to_json()does not. If tabular data is your reason for building a backend at all, Parquet is why the backend pays off. - Pickle will serialize almost anything, and that is exactly why you should reach for it last. Pickle is a code-execution format: deserializing a malicious or corrupted pickle can run arbitrary code, and pickled objects are brittle across Python and library versions. Reserve it for objects you genuinely can’t express otherwise, and never point a pickle-based backend at storage that untrusted parties can write to.
The through-line: prefer the most constrained format that fits the payload. A backend that quietly falls back to pickle for everything is a backend that will one day deserialize something it shouldn’t.
Global versus per-task, and testing
AIRFLOW__CORE__XCOM_BACKEND is global — one backend for the whole deployment. There’s no per-task override knob, so the backend has to behave sanely for every XCom in the system, including the tiny ones the scheduler and mapped tasks pass around constantly. That’s the entire reason the worked example keeps a size threshold and an inline fallback: a global backend that offloads a three-character partition key to S3 would generate a storm of pointless network calls and make the whole instance slower. The discipline for a global backend is “offload only what’s worth offloading, pass everything else straight through to BaseXCom.”
If you genuinely need per-task behavior — say only the bookshop’s ETL DAG should offload, and everything else stays inline — the identity kwargs are your lever. serialize_value receives dag_id and task_id, so you can gate offloading on them: branch to the S3 path only when dag_id is in an allowlist, and fall through to BaseXCom.serialize_value otherwise. That keeps a single deployment-wide class while letting policy vary by pipeline. It’s a code branch, not configuration, which is a fair criticism — but it’s honest about the constraint, and it’s far safer than the tempting alternative of raising the global threshold so high that the DAGs which should offload silently stop doing so. Treat the size threshold and any DAG allowlist as the backend’s public policy surface, document them next to the class, and resist the urge to bury magic numbers in the branch logic.
Because the class is plain Python with static methods, you can unit-test it without a running Airflow. Round-trip a value through serialize_value and deserialize_value against a mocked or local S3 (moto, or MinIO in the Astro project’s docker-compose.override.yml) and assert you get an equal object back. Test the branches explicitly, including purge — a large DataFrame goes to Parquet in S3, a small dict stays inline, a non-serializable object raises the same error the default backend would. Test purge too: assert the object is gone afterwards, because a broken purge is completely silent — the DAG stays green and the bucket grows forever.
Here’s what that test file actually looks like with moto standing in for S3. The moto fixture gives you a real-enough S3 API in-process, so the same code path that runs in production runs in the test — no monkeypatching the hook:
# tests/xcom/test_s3_backend.py
import io
import boto3
import pandas as pd
import pytest
from moto import mock_aws
from include.xcom.s3_backend import S3XComBackend, _OFFLOAD_PREFIX
@pytest.fixture
def s3_bucket(monkeypatch):
with mock_aws():
boto3.client("s3", region_name="us-east-1").create_bucket(
Bucket="bookshop-xcom")
monkeypatch.setenv("XCOM_S3_BUCKET", "bookshop-xcom")
yield
def _coords():
# The identity Airflow would pass to serialize_value.
return dict(key="return_value", task_id="enrich_titles",
dag_id="etl_bookshop", run_id="manual__2026-06-14T00:00",
map_index=-1)
def test_large_frame_offloads_and_round_trips(s3_bucket):
frame = pd.DataFrame({"isbn": range(10_000), "title": ["x"] * 10_000})
stored = S3XComBackend.serialize_value(frame, **_coords())
# What lands in the metadata DB is a tiny envelope, not the frame.
assert _OFFLOAD_PREFIX in stored
assert len(stored) < 512
back = S3XComBackend.deserialize_value({"value": stored})
pd.testing.assert_frame_equal(back, frame)
def test_small_dict_stays_inline(s3_bucket):
stored = S3XComBackend.serialize_value({"rows": 42}, **_coords())
assert _OFFLOAD_PREFIX not in stored # never touched S3
assert S3XComBackend.deserialize_value({"value": stored}) == {"rows": 42}
def test_purge_deletes_the_offloaded_object(s3_bucket):
frame = pd.DataFrame({"isbn": range(10_000)})
stored = S3XComBackend.serialize_value(frame, **_coords())
uri = json.loads(stored)["uri"]
bucket, key = _split(uri)
assert S3XComBackend._hook().check_for_key(key, bucket) # it's there
S3XComBackend.purge(SimpleNamespace(value=stored))
assert not S3XComBackend._hook().check_for_key(key, bucket) # and now it isn'tThat last test is the one worth having, because the failure it guards against is invisible. A backend with a broken purge works perfectly — tasks push, tasks pull, the DAG is green — and quietly leaks an object per cleared XCom until someone notices the bucket has grown to a terabyte of orphaned Parquet. Nothing else in your pipeline will ever tell you. (SimpleNamespace(value=…) stands in for the XCom record Airflow passes; the real object carries more, but purge only reads the stored value.)
The exact shape of the stored result handed to deserialize_value is an internal detail; against a real Airflow you’d push through a live task instance. These tests catch the failures that are miserable to debug in a live deployment, where a serialization bug surfaces as a task that pushes fine and a downstream task that mysteriously can’t read.
When to skip the backend entirely
A custom XCom backend is the right tool when you want transparent handoff — tasks return and receive real Python objects and neither one knows storage is involved. But it’s not the only tool, and often not the best one.
If your tasks are explicitly moving files or datasets, Airflow 3.x ships airflow.io, the ObjectStorage abstraction, and using it directly is frequently cleaner than hiding S3 behind XCom. You get a Path-like API over S3, GCS, or Azure through ObjectStoragePath, and your DAG says what it does — it writes a Parquet file to a known location and passes the path. The data movement is visible in the code and in the graph, not tucked inside a serialization hook.
Here’s the same bookshop handoff written the explicit way. Note what crosses XCom: a plain string, the path — exactly the small signal XCom was built for:
from airflow.sdk import ObjectStoragePath
from airflow.sdk import dag, task
BASE = ObjectStoragePath("s3://bookshop-xcom", conn_id="aws_default")
@dag(schedule="@daily", catchup=False)
def etl_bookshop_explicit():
@task
def enrich_titles(**context) -> str:
frame = build_enriched_frame() # a big DataFrame
target = BASE / context["run_id"] / "enriched.parquet"
with target.open("wb") as f:
frame.to_parquet(f)
return str(target) # only the path travels
@task
def load_warehouse(path: str):
src = ObjectStoragePath(path, conn_id="aws_default")
with src.open("rb") as f:
frame = pd.read_parquet(f)
write_to_warehouse(frame)
load_warehouse(enrich_titles())
etl_bookshop_explicit()Because ObjectStoragePath supports the same brokered path resolution and the familiar /-join, .open(), .unlink(), and .rmdir() operations, cleanup is a first-class line of code (target.unlink() in a teardown task) rather than an out-of-band lifecycle policy you hope fires. The trade is explicitness for uniformity: every task that touches the file has to know about the path convention, whereas the custom backend hides all of it behind an ordinary return. Reach for ObjectStorage when the file is the interface between tasks; reach for a custom XCom backend when you’d rather the return-value ergonomics stay invisible and uniform across many tasks.
And step back one more level: if a downstream task needs a table, the most honest handoff is a table. Write it to the warehouse in the upstream task, pass the table name or partition through XCom, and let the downstream task read it. This is exactly the boundary where the bookshop pipeline hands off to dbt — Airflow schedules the load, the warehouse holds the data, and what crosses between them is a name, not a payload. A custom XCom backend is for the cases that fall between “a tiny signal” and “a materialized table” — a real object that’s too big for the metadata database but not something you want to model as a warehouse artifact. That band is narrower than people expect, which is the real lesson.
Final thoughts
The metadata database is Airflow’s nervous system, not its cargo hold. Everything in this chapter follows from taking that seriously: XCom exists to carry the small signals that coordinate a DAG — ids, counts, keys, URIs — and the moment a payload outgrows a log line, the answer is to move the payload through storage that was built for it and let Airflow carry only the reference. A custom BaseXCom subclass makes that offload transparent, and Airflow 3.x’s Task Execution API makes it safe, because the worker that runs your serialization code never holds the keys to the database it’s protecting. But the sharpest version of the lesson is the one that tells you not to build the backend: if downstream tasks need a table, write a table; if they need a file, use ObjectStorage and pass the path; and if they need a Python object larger than a log line, ask one more time whether XCom is the abstraction you actually want. The best custom backend is the one you reach for only after you’ve ruled out the simpler handoff — and then it’s the difference between a pipeline that scales and a VACUUM that pages you.
Comments