Automatically put large Airflow XCOM values in S3
Closed, ResolvedPublic
Actions

Assigned To

Authored By

	brouberol
	Jan 27 2026, 12:26 PM

Project Tags

Referenced Files

	F71618852: Screenshot 2026-01-28 at 12.08.40.png
	Jan 28 2026, 12:09 PM

	F71618839: Screenshot 2026-01-28 at 12.04.00.png
	Jan 28 2026, 12:04 PM

	F71618824: Screenshot 2026-01-28 at 11.58.19.png
	Jan 28 2026, 11:58 AM

	F71618808: Screenshot 2026-01-28 at 11.50.09.png
	Jan 28 2026, 11:50 AM

	F71615302: Screenshot 2026-01-27 at 17.47.39.png
	Jan 27 2026, 5:48 PM

	F71615303: Screenshot 2026-01-27 at 17.47.45.png
	Jan 27 2026, 5:48 PM

	F71615216: Screenshot 2026-01-27 at 17.16.55.png
	Jan 27 2026, 5:21 PM

	F71615217: Screenshot 2026-01-27 at 17.17.16.png
	Jan 27 2026, 5:21 PM

View All 9 Files

Subscribers

Aklapper

brouberol

Description

As seen in T414953, the refine Airflow DAGs generate large (as in multiple MBs) XCOM values that are stored in database. When leaf tasks in that DAGs run, they all pull the XCOM value generated by the root task, which causes a high amount of traffic between the scheduler and the database.

We want to configure airflow to store large XCOM values in S3 instead of the database, to ensure that these values are stored out of the database, to alleviate this particular issue.

While the issue specific to refine is being worked on in T414953, this would act as a safety net for all Airflow instances as well.

Cf https://airflow.apache.org/docs/apache-airflow/2.10.5/core-concepts/xcoms.html#object-storage-xcom-backend

Details

Related Changes in Gerrit:

	Subject	Author	Repo	Branch	Lines +/-
	airflow: store XCOMs>256KB in s3 to alleviate load on the database	Brouberol	operations/deployment-charts	master	+2 -0
	airflow: store large XCOMs in s3 to alleviate load on the database	Brouberol	operations/deployment-charts	master	+10 -2

Customize query in gerrit

Related Changes in GitLab:

	Title	Reference	Author	Source Branch	Dest Branch
	Add apache-airflow-providers-amazon[s3fs] to be able to store large XCOMs in s3	repos/data-engineering/airflow-dags!1954	brouberol	T415661-s3fs	main

Customize query in GitLab

Related Objects
Search...

Status	Subtype	Assigned	Task
Resolved		Antoine_Quhen	T411988 Airflow main performance instance optimization
Resolved	BUG REPORT	JAllemandou	T412003 Airflow-main scheduler loop sometimes slows down markedly
Resolved		brouberol	T415661 Automatically put large Airflow XCOM values in S3
Resolved		brouberol	T415777 Schedule the regular cleanup of large XCOM values from S3
Resolved		brouberol	T416821 Implement a way to fetch the content of an XCOM when stored in S3

Event Timeline

brouberol created this task.Jan 27 2026, 12:26 PM

brouberol triaged this task as Medium priority.

Restricted Application added a subscriber: Aklapper. · View Herald TranscriptJan 27 2026, 12:26 PM

brouberol merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1954

Add apache-airflow-providers-amazon[s3fs] to be able to store large XCOMs in s3

brouberol changed the task status from Open to In Progress.Jan 27 2026, 4:57 PM

brouberol moved this task from Backlog - project to In Progress on the Data-Platform-SRE (2026.01.23 - 2026.02.13) board.

brouberol edited projects, added: Data-Platform-SRE (2026.01.23 - 2026.02.13); removed: Data-Platform-SRE.

I created a devenv with the following DAGs:

from airflow.decorators import task, dag

@dag(schedule=None, catchup=False)
def generate_small_xcom():
    @task
    def generate_xcom():
        return {
            "task_id": "generate_xcom",
            "value": "small"
        }

    generate_xcom()

@dag(schedule=None, catchup=False)
def generate_large_xcom():
    @task
    def generate_xcom():
        return {
            "task_id": "generate_xcom",
            "value": "a" * 550 * 1024 # large enough to be stored in S3
        }

    generate_xcom()

generate_small_xcom()
generate_large_xcom()

The generate_small_xcom inserted an XCOM with the value in DB, whereas the second one inserted an XCOM containing a reference to the actual value, stored in S3.

Screenshot 2026-01-27 at 17.17.00.png (2,970×908 px, 351 KB)

Screenshot 2026-01-27 at 17.17.16.png (2,978×858 px, 293 KB)

Screenshot 2026-01-27 at 17.16.55.png (2,954×864 px, 196 KB)

brouberol@stat1008:~$ s3cmd --access_key=$access_key --secret_key=$secret_key --host=rgw.eqiad.dpe.anycast.wmnet --region=dpe --host-bucket=no ls s3://logs.airflow-dev.dse-k8s-eqiad/dev-brouberol/xcoms/generate_large_xcom/manual__2026-01-27T17:14:10.584142+00:00/generate_xcom/
2026-01-27 17:14          622  s3://logs.airflow-dev.dse-k8s-eqiad/dev-brouberol/xcoms/generate_large_xcom/manual__2026-01-27T17:14:10.584142+00:00/generate_xcom/3d297179-460f-43cb-81a9-c7a17f895a48.gz

The last thing we need to make sure of is whether Airflow gets the actual value automatically. Reading the source code, it should, but I want to perform a real-life test.

I changed the DAG to add tasks pulling the XCOM, either using the taskflow API or the explicit task_instance.xcom_pull call:

from airflow.decorators import task, dag
from airflow.operators.python import PythonOperator

@dag(schedule=None, catchup=False)
def generate_small_xcom():
    @task
    def generate_xcom():
        return {
            "task_id": "generate_xcom",
            "value": "small"
        }

    generate_xcom()

@dag(schedule=None, catchup=False)
def generate_large_xcom():
    @task
    def generate_xcom() -> str:
        return "a" * 550 * 1024 # large enough to be stored in S3

    @task
    def pull_xcom(value: str):
        print(type(value))
        print(len(value))

    def fetch_xcom_manually(**kwargs):
        ti = kwargs['ti']
        value = ti.xcom_pull(task_ids='generate_xcom')
        print(type(value))
        print(len(value))

    manual_xcom_pull = PythonOperator(task_id="manual_xcom_pull", python_callable=fetch_xcom_manually)

    value = generate_xcom()
    pull_xcom(value) >> manual_xcom_pull

generate_small_xcom()
generate_large_xcom()

Everything seems to have worked transparently, we don't need to manually get the file content from S3.

Screenshot 2026-01-27 at 17.47.39.png (2,960×890 px, 327 KB)

Screenshot 2026-01-27 at 17.47.45.png (2,928×946 px, 386 KB)

Change #1233776 had a related patch set uploaded (by Brouberol; author: Brouberol):

[operations/deployment-charts@master] airflow: store large XCOMs in s3 to alleviate load on the database

https://gerrit.wikimedia.org/r/1233776

gerritbot added a project: Patch-For-Review.Jan 27 2026, 5:50 PM

brouberol moved this task from In Progress to Needs Review on the Data-Platform-SRE (2026.01.23 - 2026.02.13) board.Jan 27 2026, 5:50 PM

brouberol mentioned this in T414953: Refine generates very large XCOM values.Jan 27 2026, 5:58 PM

brouberol updated the task description. (Show Details)Jan 27 2026, 8:49 PM

Change #1233776 merged by Brouberol:

[operations/deployment-charts@master] airflow: store large XCOMs in s3 to alleviate load on the database

https://gerrit.wikimedia.org/r/1233776

I've deployed the change to airflow-main.

The refine job ran, and the value were still stored in DB, which makes me believe that the XCOM size was under 0.5MB.

>>> data = {
... "hdfs_input_processed_flags":"hdfs:///wmf/data/raw/eventlogging_legacy/eventlogging_CentralNoticeBannerHistory/year=2026/month=01/day=28/hour=07/_PROCESSED",
... "hdfs_source_paths":"/wmf/data/raw/eventlogging_legacy/eventlogging_CentralNoticeBannerHistory/year=2026/month=01/day=28/hour=07",
... "hdfs_source_sensor_paths":"hdfs:///wmf/data/raw/eventlogging_legacy/eventlogging_CentralNoticeBannerHistory/year=2026/month=01/day=28/hour=07/,hdfs:///wmf/data/raw/eventlogging_legacy/eventlogging_CentralNoticeBannerHistory/year=2026/month=01/day=2\
8/hour=08/,hdfs:///wmf/data/raw/eventlogging_legacy/eventlogging_CentralNoticeBannerHistory/year=2026/month=01/day=28/hour=09/",
... "hive_partition_columns":"year:LONG,month:LONG,day:LONG,hour:LONG",
... "hive_partition_paths":"year=2026/month=1/day=28/hour=7",
... "hive_pre_partitions":[{
...     "hive_table":"event.centralnoticebannerhistory",
...     "ignore_missing_input_paths":False,
...     "schema_uri":"/analytics/legacy/centralnoticebannerhistory/latest",
...     "sensor_timeout":21600,
...     "spark_refine_job_scale":"small",
...     "spark_refine_job_scale_params":{
...         "driver_cores":1,
...         "driver_memory":"3G",
...         "master":"local[1]",
...         "skein_memory":"4096",
...         "skein_vcores":"2",
...         "spark.executor.memoryOverhead":"384MB",
...     }
... }],
... "stream":"eventlogging_CentralNoticeBannerHistory",
... "table_format":"hive",
... "table_location":"hdfs://analytics-hadoop/wmf/data/event/centralnoticebannerhistory",
... "transform_functions":"org.wikimedia.analytics.refinery.job.refine.filter_allowed_domains,org.wikimedia.analytics.refinery.job.refine.remove_canary_events,org.wikimedia.analytics.refinery.job.refine.deduplicate,org.wikimedia.analytics.refinery.job.r\
efine.geocode_ip,org.wikimedia.analytics.refinery.job.refine.parse_user_agent,org.wikimedia.analytics.refinery.job.refine.add_is_wmf_domain,org.wikimedia.analytics.refinery.job.refine.add_normalized_host,org.wikimedia.analytics.refinery.job.refine.norma\
lizeFieldNamesAndWidenTypes",
... }
>>> sys.getsizeof(pickle.dumps(data)) * 211
408074

I've changed the threshold to 0.25MB instead.

brouberol@stat1008:~$ s3cmd --access_key=$access_key --secret_key=$secret_key --host=rgw.eqiad.dpe.anycast.wmnet --region=dpe --host-bucket=no ls s3://logs.airflow-main.dse-k8s-eqiad/xcoms/refine_to_hive_hourly/
                          DIR  s3://logs.airflow-main.dse-k8s-eqiad/xcoms/refine_to_hive_hourly/scheduled__2026-01-28T08:00:00+00:00/
                          DIR  s3://logs.airflow-main.dse-k8s-eqiad/xcoms/refine_to_hive_hourly/scheduled__2026-01-28T09:00:00+00:00/

It worked!

Now that we're storing / loading XCOMs in/from s3, we should see an effect on the network usage of the associated host.

brouberol@deploy2002:~$ k get cluster
NAME                      AGE    INSTANCES   READY   STATUS                     PRIMARY
postgresql-airflow-main   338d   2           2       Cluster in healthy state   postgresql-airflow-main-1
brouberol@deploy2002:~$ k get pod -o wide | grep post | grep -v pooler
postgresql-airflow-main-1                                         1/1     Running             0          24h     10.67.28.121   dse-k8s-worker1014.eqiad.wmnet   <none>           <none>
postgresql-airflow-main-3                                         1/1     Running             0          24h     10.67.28.15    dse-k8s-worker1005.eqiad.wmnet   <none>           <none>

And indeed!