Airflow DAG代码更新后任务仍运行旧代码的相关问题咨询

1. 延迟产生原因及缩短方案 #

产生原因 #

• Airflow 1.10.9本身的分层解析机制：Webserver、Scheduler、Worker三个组件各自独立扫描解析DAG文件，其中Webserver的默认扫描周期更短，所以会出现GUI已经显示更新后代码，但调度、执行层还未同步的情况。核心控制参数为min_file_process_interval（调度器DAG文件重解析间隔，默认300秒）、dag_dir_list_interval（DAG目录文件清单刷新间隔，默认300秒）。
• Cloud Composer的GCS同步机制：用户上传到Composer关联GCS桶的DAG文件，需要通过集群内置的同步进程分发到所有Scheduler、Worker节点的本地目录，该同步进程默认周期最长可达5分钟，两类延迟叠加后就会出现10分钟左右的旧代码运行窗口。

延迟缩短方案 #

• 调整Airflow核心配置：在Cloud Composer环境的「Airflow配置修改」页面，下调两个核心参数：min_file_process_interval可调整到30_{60秒，`dag_dir_list_interval`可调整到60}120秒。注意如果环境内DAG总数超过100，不建议设置过低，否则会大幅提升调度器负载，引发调度卡顿。
• 调整GCS同步间隔：在Composer环境变量中新增GCS_SYNC_INTERVAL_SECONDS参数，可下调到30~60秒，减少GCS到集群节点的文件同步延迟。

2. 确认任务代码已完成更新的方法 #

• 内置版本标记校验：在DAG代码中新增固定版本变量，比如DAG_VERSION = "20240520_v1"，将该变量写入DAG的doc_md属性，即可直接在Airflow GUI的DAG详情页看到当前解析的版本号；同时可以在DAG的第一个任务中加入版本号打印逻辑，运行时直接从任务日志确认实际执行的代码版本。
• 直接验证Worker节点文件：通过kubectl连接到Composer对应的GKE集群，执行kubectl exec -it <对应Worker Pod名称> -- cat /home/airflow/gcs/dags/<你的DAG文件名.py>，直接查看Worker节点本地存储的DAG文件内容是否为最新版本。
• 测试运行校验：每次更新DAG后，先手动触发一个带测试标记的DAG Run，仅运行前置的版本校验任务，确认版本正确后再触发正式业务任务。

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