Airflow 2.7.3中API调用DAG无法并行执行问题求助

针对你遇到的DAG并行运行问题，结合Airflow 2.7.x版本的特性调整，可从以下几个方面排查解决：

• 显式设置DAG级别的concurrency参数
  仅设置max_active_runs只能控制DAG的并行运行实例数，而concurrency参数控制同一个DAG中可同时执行的任务实例总数。Airflow 2.7.x的默认concurrency值可能与2.2.x不同，需显式配置：

  from datetime import datetime
  from airflow import DAG
  
  with DAG('DAG-21',
           schedule_interval=None,
           start_date=datetime(2022,10,5),
           max_active_runs=10,
           concurrency=10) as dag:
      # 任务定义...
  

• 检查全局并行配置
  确认airflow.cfg中的全局参数是否限制了并行能力：

  • core.parallelism：Airflow实例允许的最大任务实例并行数，默认32，若任务总量超过该值会触发排队
  • core.dag_concurrency：单个DAG默认的任务并行数，默认16，若DAG未显式设置concurrency会继承此值
    调整后需重启Airflow服务生效。

• 排查任务池（Pool）限制
  若任务指定了自定义pool，需检查该池的槽位（slots）数量：

  1. 登录Airflow UI，进入Admin -> Pools
  2. 查看目标池的槽数，若槽数不足需增加，或改为使用默认池（默认槽数128）

• 确保API触发时生成唯一运行实例
  调用API触发DAG时，避免手动指定重复的run_id，让Airflow自动生成唯一ID，否则会被判定为同一运行实例而排队：

  curl -X POST "http://<airflow-host>:<port>/api/v1/dags/DAG-21/dagRuns" \
       -H "Content-Type: application/json" \
       -H "Authorization: Bearer <your-token>" \
       -d '{"conf": {}}'
  

• 检查调度器配置与日志
  查看调度器日志，确认是否存在调度延迟或资源不足的提示，必要时调整调度器参数（如max_tis_per_query、parsing_processes）提升处理能力。

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