Airflow 2.8.2可删除元数据库对象及Postgres清理咨询

我通过Docker部署了Airflow 2.8.2，需要清理其关联的Postgres数据库以释放硬盘空间。已知不能直接删除表中数据，已实现简易清理DAG代码，同时找到旧版本Airflow的数据库清理代码但不适用于当前版本，整理了候选数据库对象列表并配置了部分清理规则，现明确Airflow 2.8.2中可安全删除的元数据库对象。

现有清理实现 #

简易DagRun清理DAG代码 #

from airflow import settings
from airflow.models import DagRun
from sqlalchemy.orm import load_only
from airflow.operators.python import PythonOperator

session = settings.Session()
def remove_old_dag_runs():
    query = session.query(
            DagRun # 指定要清理的资源元数据
        ).options(
            load_only(
                DagRun.execution_date # 仅加载执行日期列以提升性能
            )
        )
    query = query.filter(
            DagRun.execution_date < '2022-02-25 12:23:38.462662+00'
        )
    query.delete(synchronize_session=False)
    session.commit()

remove_old_dag_runs_task = PythonOperator(
    task_id='remove_old_dag_runs',
    python_callable=remove_old_dag_runs,
    dag=dag
)

当前已配置的清理规则 #

from airflow.models import DagRun, TaskInstance, XCom, TaskReschedule

DATABASE_OBJECTS = {
    'DagRun': {
        "airflow_db_model": DagRun # 对应数据库模型类
        , "age_check_column": DagRun.execution_date # 用于日期过滤的字段
        , "keep_last": True # 是否保留最后一次运行记录
        , "keep_last_filters": [DagRun.external_trigger.is_(False)] # 额外过滤条件
        , "keep_last_group_by": DagRun.dag_id
        , 'age_in_days': 14 # 保留天数
    }
    , 'TaskInstance': {
        "airflow_db_model": TaskInstance
        , "age_check_column": TaskInstance.execution_date
        , "keep_last": False
        , "keep_last_filters": None
        , "keep_last_group_by": None
        , 'age_in_days': 14
    }
    , XCom.__name__: {
        "airflow_db_model": XCom
        , "age_check_column": XCom.execution_date
        , "keep_last": False
        , "keep_last_filters": None
        , "keep_last_group_by": None
        , 'age_in_days': 3
    }
    , 'TaskReschedule': {
        "airflow_db_model": TaskReschedule
        , "age_check_column": TaskReschedule.execution_date
        , "keep_last": False
        , "keep_last_filters": None
        , "keep_last_group_by": None
        , 'age_in_days': 14
    }
}

DagRun清理SQL查询 #

delete
from    airflow.dag_run
where   execution_date not in (select max(execution_date) from dag_run where external_trigger = false group by dag_id)
        and execution_date <= current_date - interval '14 days'

候选数据库对象列表 #

airflow.models 候选对象 #

• DagModel, DagRun, Log, XCom, SlaMiss, TaskInstance, Variable
• TaskReschedule, TaskFail, RenderedTaskInstanceFields, ImportError

airflow.jobs 候选对象 #

• BaseJob（来自airflow.jobs.base_job）
• SchedulerJob（来自airflow.jobs.scheduler_job）
• LocalTaskJob（来自airflow.jobs.local_task_job）

可安全清理的元数据库对象 #

以下对象的旧数据可安全清理，建议根据业务需求设置合理的保留天数：

可直接按规则清理的对象 #

• DagRun：任务流运行记录，保留最近非外部触发的运行记录后，清理超过保留天数的旧记录，不会影响当前DAG运行。
• TaskInstance：任务实例执行记录，旧记录仅用于历史追溯，清理后不影响现有任务调度。
• XCom：任务间传递的临时数据，旧数据无业务价值，可设置较短保留周期（如3天）。
• TaskReschedule：任务重调度历史记录，清理旧记录无负面影响。
• Log：Airflow系统与任务日志，若不需要长期历史日志排查，可清理旧日志；若需保留排查依据，可延长保留天数。
• TaskFail：任务失败记录，旧失败记录仅用于统计或追溯，清理后不影响当前运行。
• RenderedTaskInstanceFields：任务渲染后的字段内容，旧数据无复用价值，可安全清理。
• SlaMiss：SLA超时记录，旧记录仅用于历史合规或统计，可清理。
• ImportError：DAG导入错误记录，旧错误记录已无参考意义，可清理。

需谨慎操作的对象 #

• DagModel：存储DAG的核心元数据（如定义、启用状态），绝对不能删除，否则对应DAG会从Airflow中消失。
• Variable：全局配置变量，仅能删除确认已废弃的变量，否则会导致依赖该变量的DAG运行失败。
• BaseJob/SchedulerJob/LocalTaskJob：作业执行记录（如调度器运行、本地任务作业），建议保留最近的记录（如7-14天），不要全部删除，避免排查历史作业问题无据可查。

内容的提问来源于stack exchange，提问作者John Doe