Airflow 2.7.3(K8s)中DAG更新无法持久显示至UI的问题求助

问题现象 #

• 修改DAG引用的外部函数代码并推送到Git后，Airflow UI无法稳定展示更新：
  1. 除非重命名函数文件，否则UI不显示代码变更；
  2. 执行airflow dags reserialize可临时显示更新后的DAG；
  3. 约1小时后，UI自动回退至原始DAG版本，无额外操作或服务重启。

已尝试操作 #

• 推送Git代码变更
• 执行airflow dags reserialize命令
• 重命名函数文件
• 重启Airflow服务

仅重命名文件和临时执行reserialize有效，但不符合“代码推送后自动同步”的预期

复现步骤 #

1. 修改DAG依赖的外部函数文件代码；
2. 推送变更至Git仓库后刷新Airflow UI；
3. 执行airflow dags reserialize，此时UI可见更新后的DAG；
4. 等待约1小时，UI自动回退至原始DAG版本。

代码概览 #

主DAG文件 (perception-enrichments-regression-prototype.py) #

with DAG('perception-enrichments-regression-prototype',
         default_args=DEFAULT_ARGS,
         schedule_interval="@daily",
         description='perception-enrichments-prototype',
         max_active_runs=3,
         catchup=False,
         tags=["perception-enrichments"],
         is_paused_upon_creation=True,
         ) as dag:
    set_alert_email_in_dag(dag)

    init_dag_variables = PythonOperator(
        task_id="init-dag-variables3",
        python_callable=init_dag_variables_callable,
        provide_context=True,
        trigger_rule=TriggerRule.NONE_FAILED,
    )

    start = EmptyOperator(task_id='perception-enrichments-prototype-start')
    done = EmptyOperator(task_id='perception-enrichments-prototype-done')

    start >> init_dag_variables >> get_ada_enrich_prototype("{{ ti.xcom_pull(key='pit') }}", dag) >> done

外部函数文件 (groups/annotation_group/ada_enrich_prototype.py) #

from typing import Optional
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup

def get_ada_enrich_prototype(dag, pit: Optional[str] = None):
    with TaskGroup(group_id='ada-enrich-prototype') as ada_enrich_prototype:
        start = EmptyOperator(task_id='start-enrich-prototype123')
        start2 = EmptyOperator(task_id='start2-enrich-prototype123')
        start3 = BashOperator(task_id='start3-enrich-prototype123', bash_command='echo "Hello World"')
        done = EmptyOperator(task_id='done-enrich-prototype', trigger_rule="all_done")
        
        start >> start2 >> start3 >> done
    return ada_enrich_prototype

排查与解决建议 #

1. 调整DAG解析与序列化配置 #

Airflow 2.x默认序列化DAG到元数据库，Scheduler定期重新解析文件，1小时回退大概率和缓存/解析间隔有关：

• 缩短scheduler.min_file_process_interval（默认300秒），让Scheduler更频繁扫描DAG文件变更；
• 检查core.dag_file_processor_timeout，确保Scheduler能完成新代码的解析；
• 若临时验证，可设置core.dag_serialization_enabled = False禁用序列化缓存（生产环境不推荐），或缩短core.dag_serialization_timeout的缓存有效期。

2. 确保Git同步机制及时生效 #

K8s部署中若用GitSync/Argo CD同步DAG目录：

• 缩短Git同步间隔，确保最新代码快速同步到Airflow Pod；
• 同步完成后自动触发DAG重新解析：比如在CI/CD中调用airflow dags reserialize，或通过生成临时.airflowignore文件触发Scheduler重新扫描；
• 验证同步后的文件权限，确保Airflow进程有读取权限。

3. 检查TaskGroup生成逻辑 #

外部函数生成TaskGroup时，需确保：

• 避免硬编码重复的Task ID（当前代码无此问题，但更新后若Task结构未变化，Airflow可能复用旧序列化数据）；
• 依赖Airflow的自动解析机制，不要通过静态标识绑定TaskGroup，确保代码变更能被识别。

4. 验证Scheduler与Webserver的DAG目录一致性 #

K8s集群中需保证Scheduler和Webserver Pod的DAG目录完全同步：

• 检查两者的DAG存储卷挂载配置是否一致；
• 多Scheduler Pod场景下，确保所有Pod都能拉取到最新Git代码。

5. 自动化触发DAG更新 #

在Git推送后，通过CI/CD流程自动执行：

# K8s中触发指定DAG重新序列化示例
kubectl exec -it <airflow-scheduler-pod-name> -- airflow dags reserialize -d perception-enrichments-regression-prototype

内容的提问来源于stack exchange，提问作者אייל גל