# Save this file in your airflow plugins/ directory as custom_backend.py import json import uuid from typing import Any from airflow.models.xcom import BaseXCom from airflow.providers.amazon.aws.hooks.s3 import S3Hook class S3XComBackend(BaseXCom): PREFIX = "s3://" BUCKET_NAME = "my-exclusive-airflow-xcom-bucket" @staticmethod def serialize(value: Any, **kwargs) -> str: """Serializes the object, saves to S3, and returns the S3 URI.""" s3_hook = S3Hook(aws_conn_id='aws_default') key = f"xcom/uuid.uuid4().json" # Convert object to string/json string_data = json.dumps(value) # Load string into S3 s3_hook.load_string( string_data=string_data, key=key, bucket_name=S3XComBackend.BUCKET_NAME, replace=True ) # This string URI is what actually saves to the Airflow DB return f"S3XComBackend.PREFIXS3XComBackend.BUCKET_NAME/key" @staticmethod def deserialize(result, **kwargs) -> Any: """Reads the S3 URI from the database and pulls the real data from S3.""" s3_hook = S3Hook(aws_conn_id='aws_default') s3_uri = result.get_value() # Strip the prefix to get the bucket and key path path = s3_uri.replace(S3XComBackend.PREFIX, "") bucket, key = path.split("/", 1) # Read from S3 file_content = s3_hook.read_key(key=key, bucket_name=bucket) return json.loads(file_content) Use code with caution.
To maintain a clean and professional Airflow environment, follow these exclusive patterns: Use the TaskFlow API (@task) airflow xcom exclusive
@task def extract_data(): # Dangerous: Automatically pushes a massive list to the metadata DB return ["row1", "row2", "row3", ... , "row1000000"] Use code with caution. # Save this file in your airflow plugins/
The Definitive Guide to Apache Airflow XComs: Deep Dive, Advanced Patterns, and Exclusive Optimization Techniques The Definitive Guide to Apache Airflow XComs: Deep