生产环境的DAG版本管理——代码即调度

# 推荐的Git工作流
# 1. 开发在feature分支写/改DAG
# 2. 提PR到main分支
# 3. PR触发CI：语法检查+单元测试+DAG解析验证
# 4. Code Review通过后合并
# 5. CD自动部署：同步DAG文件到Airflow服务器
#
# Git仓库结构
# airflow-dags/
# ├── dags/
# │   ├── daily_warehouse.py
# │   ├── hourly_metrics.py
# │   └── ...
# ├── plugins/
# ├── tests/
# │   ├── test_dag_integrity.py
# │   └── test_sql_syntax.py
# ├── .github/workflows/ci.yml
# └── requirements.txt

# tests/test_dag_integrity.py
import pytest
from airflow.models import DagBag
 
def test_no_import_errors():
    dag_bag = DagBag(include_examples=False)
    assert len(dag_bag.import_errors) == 0, dag_bag.import_errors
 
def test_no_cycles():
    dag_bag = DagBag(include_examples=False)
    for dag_id, dag in dag_bag.dags.items():
        assert not dag.test_cycle(), f'{dag_id} has a cycle!'
 
def test_all_tasks_have_owner():
    dag_bag = DagBag(include_examples=False)
    for dag_id, dag in dag_bag.dags.items():
        for task in dag.tasks:
            assert task.owner != 'airflow', (
                f'{dag_id}.{task.task_id} missing owner')

# 方式一：Git Sync（最常用）
# Airflow调度器所在服务器定期git pull
# 或用sidecar容器持续同步
# Kubernetes: git-sync initContainer
 
# 方式二：S3同步
# CI/CD把DAG文件上传到S3
# Airflow配置从S3读取DAG
# AWS MWAA就是这种方式
 
# 方式三：Docker镜像
# 把DAG文件打包进Airflow Docker镜像
# 每次发布构建新镜像、滚动更新
# 最安全但发布周期最长
 
# 推荐：小团队用Git Sync，大团队用Docker镜像

灰度发布DAG

# 新DAG上线的灰度策略
# 1. 先部署DAG但设paused=True（不自动执行）
dag = DAG('new_etl', is_paused_upon_creation=True, ...)
 
# 2. 手动触发一次测试Run验证
# 3. 检查结果正确后取消pause
# 4. 观察1-3天确认稳定
 
# 修改已有DAG的灰度策略
# 1. 创建shadow DAG（new_etl_v2）和原DAG并行
# 2. shadow DAG跑1周对比结果
# 3. 确认一致后切换（rename DAG ID或改下游依赖）
# 4. 保留原DAG1周可回滚

实战经验：从手动部署到CI/CD

某短视频平台的DAG管理演进：

阶段一（混乱期）：DAG文件直接在服务器上vim编辑。没有版本管理、没有Review、没有测试。经常出现'谁改了这个DAG''为什么昨天的修复消失了'的问题。

阶段二（Git管理）：DAG文件放Git仓库，git pull部署。解决了版本管理问题但还是手动部署。

阶段三（CI/CD）：GitHub Actions做CI（语法检查+DAG解析+单元测试），合并后自动同步到Airflow服务器。发布时间从'找运维手动部署30分钟'变成'合并PR后2分钟自动生效'。

面试高频追问

追问一：DAG文件更新后Airflow多久生效？

取决于min_file_process_interval配置（默认30秒）。Scheduler每30秒扫描一次DAG文件目录，检测到变化后重新解析。如果DAG文件很多扫描一轮可能要几分钟。可以用airflow dags reserialize命令立即触发解析。

追问二：怎么做DAG的回滚？

Git revert+重新部署。因为DAG文件在Git中有完整历史，回滚就是git revert到上一个版本然后触发CD部署。如果是Docker镜像方式，回滚就是切换到上一个镜像版本。

追问三：DataWorks怎么做版本管理？

DataWorks内置了版本管理——每次提交节点变更会生成一个版本号，可以查看历史版本和回滚。但不如Git灵活——没有分支、没有PR Review流程。大型团队建议在DataWorks之外用Git管理SQL文件，DataWorks只做执行。

总结

DAG版本管理讲三层：① Git管理（DAG文件和应用代码一样走Git，禁止服务器直接修改）；② CI/CD（语法检查+DAG解析验证+自动部署，发布从30分钟变成2分钟）；③ 灰度发布（新DAG先paused+手动测试，改DAG用shadow并行验证）。能讲出从vim直接编辑到CI/CD的演进过程，面试官觉得你有工程化的调度管理思维。

下一篇聊Executor选型：Local/Celery/Kubernetes怎么选？

你们DAG怎么管理和部署的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长