Airflow on Kubernetes：动态映射任务为何一段时间后停止排队？

• 拥有一个包含4000+动态映射任务的Airflow DAG，max_active_tasks设置为32，完整运行需数日
• DAG启动初期正常，Pod数量维持在32个上限，启停正常
• 运行数小时后，任务卡在scheduled状态，无活跃任务Pod，集群无任务运行
• 重启调度器可临时恢复，但需要稳定解决方案
• 调度器日志仅偶尔出现以下警告（正常运行时也会出现）：

  WARNING - Killing DAGFileProcessorProcess

  DETAIL: Key (dag_id)=(my_dag_name) already exists.
  [SQL: INSERT INTO serialized_dag (dag_id, fileloc, fileloc_hash, data, data_compressed, last_updated, dag_hash, processor_subdir) VALUES (%(dag_id)s, %(fileloc)s, %(fileloc_hash)s, %(data)s, %(data_compressed)s, %(last_updated)s, %(dag_hash)s, %(processor_subdir)s)]

  {manager.py:543} INFO - DAG my_dag_name is missing and will be deactivated.
  {manager.py:553} INFO - Deactivated 1 DAGs which are no longer present in file.
  {manager.py:557} INFO - Deleted DAG my_dag_name in serialized_dag table

• 已排除其他并行DAG干扰（仅运行此DAG时问题仍发生）

1. 调度器DAG文件处理器资源与超时问题 #

• 排查调度器的CPU/内存使用率，确认是否因资源耗尽导致DAGFileProcessorProcess频繁被OOM或系统杀死，进而中断DAG元数据同步
• 调整调度器参数dag_file_processor_timeout（默认5分钟），针对高负载动态映射DAG，延长处理器超时时间，避免因生成元数据耗时过长被强制终止
• 确认max_active_runs_per_dag参数设置，确保单个DAG的运行实例数限制不会干扰任务调度

2. Serialized DAG表的一致性冲突 #

• 检查数据库（PostgreSQL/MySQL）中serialized_dag表的锁状态，查看是否存在长时间未释放的锁，导致元数据写入冲突（对应日志中的主键重复警告）
• 临时禁用Serialized DAG（设置serialize_dag=False），验证问题是否消失：若禁用后正常，说明Serialized DAG机制在高负载下的同步存在问题，可调整min_serialized_dag_update_interval参数减少更新频率，缓解写入冲突
• 定期清理serialized_dag表中的冗余数据，避免元数据堆积导致的异常

3. 元数据存储的性能瓶颈 #

• 检查task_instance、xcom等核心表的数据量，4000+动态任务会产生大量元数据，若数据库膨胀会导致调度器查询任务状态的速度变慢，无法及时将scheduled任务转为queued
• 针对task_instance表的scheduled状态查询添加索引，优化慢查询性能
• 配置Airflow元数据清理机制（执行airflow db clean），设置合理的保留周期，定期清理过期任务实例、XCom等数据

4. 调度器任务调度逻辑阻塞 #

• 开启调度器DEBUG级日志（调整logging_level=DEBUG），捕捉任务调度的详细流程，定位是否存在循环阻塞或状态更新停滞的环节
• 确认全局参数parallelism是否设置过低，虽然初期正常，但长时间运行后可能因全局并行数限制导致任务无法调度
• 检查调度器的task_queued_timeout参数，确保任务不会因排队超时被标记为失败

5. Kubernetes Executor状态同步问题 #

• 查看Kubernetes Executor日志，检查是否存在与K8s API通信失败、Pod状态更新不及时的错误
• 调整kube_client_request_timeout参数，延长K8s API请求超时时间，避免因集群响应慢导致状态同步中断
• 检查Kubernetes集群节点状态，确认是否有节点NotReady、资源耗尽等情况，导致无法创建新任务Pod

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