Airflow（Postgres）数据库定期管理及Web服务卡顿问题咨询

一、Airflow内置的数据库清理机制 #

Airflow提供了内置的数据库清理命令airflow db clean，可自动清理旧的dag_run、log、task_instance等表的数据，无需手动编写基础删除脚本。

使用方式 #

1. 手动执行命令：直接在终端运行，指定保留数据的天数（示例保留30天以内的数据）：

   airflow db clean --days 30 --verbose
   

2. 定期自动清理：创建DAG调度该命令，实现每日/每周自动清理。示例DAG如下：

   from airflow import DAG
   from airflow.operators.bash import BashOperator
   from datetime import datetime, timedelta
   
   default_args = {
       'owner': 'airflow',
       'start_date': datetime(2024, 1, 1),
       'retries': 1,
       'retry_delay': timedelta(minutes=5)
   }
   
   with DAG(
       'automated_db_cleanup',
       default_args=default_args,
       schedule_interval='0 2 * * *',  # 每日凌晨2点执行
       catchup=False
   ) as dag:
       clean_db_task = BashOperator(
           task_id='clean_stale_db_records',
           bash_command='airflow db clean --days 30 --verbose'
       )
   

3. 配置默认保留天数：在airflow.cfg中设置core.max_db_retention_days参数，后续执行airflow db clean时会默认使用该天数，无需每次手动指定。

二、自定义清理逻辑（针对特定表） #

如果内置命令的清理范围不符合需求，比如仅需清理dag_run和log表，可通过PythonOperator编写自定义清理逻辑：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta

def cleanup_dag_run_and_log():
    # 初始化Postgres连接
    pg_hook = PostgresHook(postgres_conn_id='your_postgres_connection_id')
    conn = pg_hook.get_conn()
    cursor = conn.cursor()

    # 删除30天前的dag_run数据
    cursor.execute("""
        DELETE FROM dag_run 
        WHERE execution_date < NOW() - INTERVAL '30 days';
    """)

    # 删除30天前的log数据
    cursor.execute("""
        DELETE FROM log 
        WHERE date < NOW() - INTERVAL '30 days';
    """)

    conn.commit()
    cursor.close()
    conn.close()

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'custom_db_cleanup',
    default_args=default_args,
    schedule_interval='0 3 * * *',  # 每日凌晨3点执行
    catchup=False
) as dag:
    custom_clean_task = PythonOperator(
        task_id='clean_dag_run_log_tables',
        python_callable=cleanup_dag_run_and_log
    )

注意：替换your_postgres_connection_id为Airflow中配置的Postgres连接ID；执行删除操作前建议先通过SELECT语句验证清理范围，避免误删数据。

三、数据累积导致Web服务器卡顿的原因 #

Airflow Web UI的多个核心页面（如DAG运行历史、任务日志页）依赖dag_run和log表的数据：

• 查询性能瓶颈：当表中数据量过大时，即使是分页查询，SQL语句的WHERE过滤、排序操作需要扫描大量数据，导致查询延迟升高，进而拖慢Web服务器响应。如果这些表的execution_date（dag_run）、date（log）字段未建立索引，查询效率会进一步降低。
• 资源竞争：大表查询会占用数据库大量CPU、内存资源，导致数据库整体响应变慢，Web服务器的所有数据库请求都会受到影响。

打开新窗口时，Web服务器不会一次性获取全部数据，而是采用分页加载，但总数据量过大时，分页查询的底层逻辑仍需遍历大量数据来定位目标页，导致页面加载卡顿。

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