Cleanup Airflow Database. Xcom Table – Docker

With every question, please provide details (versions, stack traces, etc.) about what you tried, and what didn’t work. Saying "it doesn’t work" doesn’t tell anything and makes it difficult to help. See https://stackoverflow.com/help/how-to-ask for how to write a good question.

If you have direct access to the metastore, you could execute this query:

DELETE FROM xcom;

The XCom table contains a timestamp which you can use to preserve recent XComs. For example, this query deletes all XComs older than 14 days:

DELETE FROM xcom WHERE current_date - "timestamp"::date > 14;

If you do not have direct access to the metastore, you could create a DAG to clean up objects in the metastore. Whether you run this on a schedule to clean objects periodically, or without a schedule and run it whenever you want to clean up the database is up to yourself:

import datetime

from airflow import DAG
from airflow.models import XCom
from airflow.operators.python import PythonOperator
from airflow.utils.session import provide_session

with DAG(dag_id="cleanup_xcoms", schedule_interval=None, start_date=datetime.datetime(2022, 1, 1)) as dag:

    @provide_session
    def _delete_xcoms(session=None):
        num_rows_deleted = 0

        try:
            num_rows_deleted = session.query(XCom).delete()
            session.commit()
        except:
            session.rollback()

        print(f"Deleted {num_rows_deleted} XCom rows")

    delete_xcoms = PythonOperator(task_id="delete_xcoms", python_callable=_delete_xcoms)

The XCom object has several attributes which you could filter on, for example dag_id:

session.query(XCom).filter(XCom.dag_id == "mydag123").delete()