如何在Cloud Build中添加DAG校验 仅同步合法DAG到Cloud Composer

步骤顺序说明 #

DAG校验步骤必须放在同步到Cloud Composer DAG存储桶之前：如果校验放在同步之后，即使校验不通过，有问题的DAG已经被同步到Composer的DAG目录，Composer会自动加载这些问题DAG触发错误告警，影响线上任务运行。前置校验可以在发现问题时直接终止Cloud Build流程，避免问题流到线上环境。

核心改造方案 #

不需要硬凑gsutil rsync的排除规则，更稳定的实现逻辑是：在校验完成后，仅将通过校验的DAG文件同步到临时目录，再将临时目录整体同步到GCS，既可以避开非DAG Python文件，也可以过滤掉校验不通过的无效DAG，完全满足需求。

具体实现步骤 #

1. 调整DAG校验脚本 #

先修改你现有校验脚本的两处兼容问题：

• 将self.dagbag = DagBag()改为self.dagbag = DagBag(dag_folder="./dags", include_examples=False)，指定DAG加载目录，排除Airflow示例DAG干扰
• Python3环境下将self.dagbag.dags.iteritems()改为self.dagbag.dags.items()，避免语法错误
• 可选扩展：测试通过后，将所有合法DAG的文件路径写入valid_dags.txt，方便后续步骤精准复制文件

调整后将脚本命名为test_dag_integrity.py存放在仓库根目录。

2. 完整cloudbuild.yaml配置 #

steps:
- name: ubuntu
  id: Initialization
  args: ['bash', '-c', "echo '$COMMIT_SHA' > REVISION.txt"]

- name: gcr.io/google.com/cloudsdktool/cloud-sdk:airflow-2.6.3
  id: DAGIntegrityCheck
  entrypoint: bash
  args:
  - '-c'
  - |
    # 可根据需求安装自定义DAG依赖
    pip install -r requirements.txt --quiet
    # 校验失败会直接抛出异常终止整个Cloud Build流程
    python test_dag_integrity.py

- name: ubuntu
  id: FilterValidDAGs
  entrypoint: bash
  args:
  - '-c'
  - |
    mkdir -p output_dags
    # 方案1：如果校验脚本已输出合法DAG列表valid_dags.txt，使用该方式精准复制
    # cat valid_dags.txt | xargs cp -t output_dags/
    # 方案2：如果明确知道合法DAG的存放规则，直接按规则筛选复制
    find . -path "./dags/*.py" -not -path "./utils/*" -not -path "./tests/*" -exec cp {} output_dags/ \;
    cp REVISION.txt output_dags/

- name: gcr.io/cloud-builders/gsutil
  id: Deployment
  args: ['rsync', '-r', 'output_dags/', '${_GCS_BUCKET}/dags/']

- name: gcr.io/cloud-builders/gsutil
  id: Check
  args: ['ls', '${_GCS_BUCKET}/dags/']

注意：请将DAG校验步骤的Airflow镜像版本替换为你Cloud Composer环境实际运行的Airflow版本，避免版本兼容问题导致校验通过但线上加载失败。

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