To make Airflow use your custom backend, configure the environment variable or update your airflow.cfg :
from airflow.decorators import dag, task from datetime import datetime import pandas as pd @dag(start_date=datetime(2026, 1, 1), schedule=None, catchup=False) def enterprise_data_pipeline(): @task def extract_user_demographics(): # Representing data extraction raw_data = "user_id": [101, 102], "country": ["US", "KR"] # If Custom Backend is active, this Dict/DataFrame securely saves to S3 return raw_data @task def process_demographics(demographics): # Airflow automatically resolves the XCom backend URI back into the raw object df = pd.DataFrame(demographics) processed_data = df.to_dict(orient="records") return processed_data # Setting up dependency seamlessly via Python function invocation user_data = extract_user_demographics() process_demographics(user_data) enterprise_data_pipeline() Use code with caution. Mixing TaskFlow with Traditional Operators
Airflow XCom is a powerful feature that enables tasks to communicate and share data. By understanding its benefits and exclusive use cases, you can unlock its full potential and create more dynamic and flexible workflows. By following best practices and using XCom judiciously, you can take your Airflow workflows to the next level.
def task_a(**context): context['ti'].xcom_push(key=f"result_context['ti'].task_id", value=100) airflow xcom exclusive
+-------------------+ Returns Object/Data +-----------------------+ | Upstream Task | --------------------------------> | Custom XCom Backend | +-------------------+ +-----------------------+ | +---------------------------+---------------------------+ | Serialize & Upload Payload | Save Metadata Pointer v v +-----------------------+ +-----------------------+ | Cloud Object Storage | | Airflow Metadata DB | | (S3 / GCS / Azure) | | (Stores JSON URI) | +-----------------------+ +-----------------------+ Architecture of a Custom Backend
“XComs let tasks exchange messages, allowing more nuanced forms of control and shared state. XComs look like a simple key‑value store, but are not intended for passing large amounts of data.”
For explicit control, use xcom_push with a meaningful key: To make Airflow use your custom backend, configure
To create an exclusive, secure cloud-backed XCom system, you override the BaseXCom class:
: XComs allow tasks to exchange messages, creating "shared state" within a specific DAG run.
@task def transform_data(data: dict) -> int: # data contains the dictionary returned by extract_data print(f"Processing data['path'] in data['format'] format") return 42 By following best practices and using XCom judiciously,
Unlike Variables, which are globally accessible to all DAGs, XComs are scoped exclusively to the DAG run in which they were created. The combination of dag_id , run_id , task_id , and map_index uniquely identifies each XCom. This means a downstream task can only pull an XCom from a known upstream task within the same DAG run—no accidental cross‑run contamination.
The xcom table is historically one of the fastest-growing tables in Airflow databases. Ensure your infrastructure team sets up a regular maintenance maintenance DAG using the airflow db clean command to prune historical XCom records corresponding to old DAG runs. Summary Reference Default Backend Custom Backend (Cloud) Storage Location Metadata Database Table ( xcom ) Cloud Storage Bucket (S3/GCS/Blob) Size Limitations Strict limits (dependent on DB column types) Virtually limitless (constrained by cloud provider limits) Performance Impact High DB I/O bottleneck at scale Minimal; database only stores URI strings Best Used For Task IDs, counters, status strings, short flags Dataframes, massive schemas, large metadata payloads
In Apache Airflow, tasks are isolated by design. They run on different workers, in separate processes, and often across distinct physical machines. This architectural isolation ensures scalability and resilience, but it introduces a fundamental challenge: How do orchestrators share data between independent tasks?
from airflow.utils.db import provide_session from airflow.models import XCom from datetime import datetime, timedelta @provide_session def cleanup_old_xcoms(session=None): oldest_date = datetime.utcnow() - timedelta(days=30) # Exclusively delete records older than 30 days to optimize DB performance session.query(XCom).filter(XCom.execution_date < oldest_date).delete() Use code with caution. Summary Checklist for Advanced XCom Architectures