如何限制Apache Airflow中调度的DAG运行数量？解决队列过载问题

核心问题根源 #

Airflow 2.x默认开启catchup=True，当回溯历史数据时会一次性生成所有错过时间段的DAGRun实例，即使max_active_dag_runs限制了活跃运行数，剩余的DAGRun仍会进入排队状态，占据元数据库资源并导致调度器过载。

可行解决方案 #

1. 关闭自动追补，分批次手动回溯 #

将DAG的catchup设为False，避免一次性生成所有历史DAGRun，再通过命令行或UI分时间段触发回溯：

• 在DAG定义中添加：

  default_args={
      # 其他参数
  },
  catchup=False,
  schedule_interval="@hourly"
  

• 使用backfill命令分批次处理，例如先处理2022年1-3月：

  airflow dags backfill --dag-id your_sync_dag --start-date 2022-01-01 --end-date 2022-04-01 --max-active-runs 3
  

  通过--max-active-runs控制同时运行的回溯任务数，避免生成大量排队实例。

2. 限制调度器单次创建的DAGRun数量 #

修改Airflow配置文件（airflow.cfg）中的scheduler部分，设置max_dag_runs_to_create_per_loop参数，控制调度器每次循环生成的DAGRun数量：

[scheduler]
max_dag_runs_to_create_per_loop = 10  # 默认值为100，根据服务器性能调整

这个参数会限制调度器每次扫描DAG时创建的待执行DAGRun数量，避免短时间内生成大量排队任务。

3. 限制整个DAG的任务实例总数 #

使用max_active_tis_per_dag参数，限制该DAG所有运行中+排队的任务实例总数，从任务层面控制负载：
在DAG定义中添加：

default_args={
    # 其他参数
},
max_active_tis_per_dag = 50  # 根据服务器资源调整

这个参数会阻止超过阈值的任务实例进入排队状态，从根源上避免任务堆积。

4. 批量清理排队的DAGRun（无需直接操作数据库） #

如果已经生成大量排队任务，可通过Airflow命令行批量清理，替代手动操作元数据库：

• 删除指定DAG的所有排队状态DAGRun：

  airflow dags delete --dag-id your_sync_dag --state queued
  

• 重置回溯任务的DAGRun状态：

  airflow dags backfill --dag-id your_sync_dag --start-date 2022-01-01 --end-date 2022-12-31 --reset_dagruns
  

5. 使用动态调度控制回溯速率 #

通过自定义PythonOperator或TriggerDagRunOperator实现动态触发回溯任务，例如每次只触发前一天的同步任务，完成后再触发下一天的，完全避免批量排队：

from airflow.operators.python import PythonOperator
from airflow.models import DagRun
from datetime import datetime, timedelta

def trigger_next_backfill(**context):
    current_end_date = context['dag_run'].conf.get('end_date', datetime(2022,1,1))
    next_end_date = current_end_date + timedelta(days=1)
    if next_end_date <= datetime(2022,12,31):
        context['dag'].create_dagrun(
            run_id=f"backfill_{next_end_date.strftime('%Y%m%d')}",
            execution_date=next_end_date - timedelta(hours=1),
            conf={'end_date': next_end_date}
        )

with DAG(
    'your_sync_dag',
    start_date=datetime(2022,1,1),
    catchup=False,
    schedule_interval=None
) as dag:
    sync_task = # 你的同步任务定义
    trigger_next = PythonOperator(
        task_id='trigger_next_backfill',
        python_callable=trigger_next_backfill,
        provide_context=True
    )
    sync_task >> trigger_next

内容的提问来源于stack exchange，提问作者Mateus Leão