Airflow的DAG写了500行没人看得懂——DAG设计最佳实践
本文目录
先说结论
好的DAG设计遵循四个原则：① 任务粒度适中（一个Task做一件事，不要太粗也不要太细）；② 命名规范清晰（看名字就知道做什么）；③ 参数化配置（不硬编码日期、路径、表名）；④ 模板复用（相似任务用工厂函数或模板生成）。
先把这段讲顺；面试官追问时，再展开下面的原理和细节。
详细讲解
DAG写得好团队维护成本低，写得差就是灾难。在某短视频平台接手一个同事离职后留下的DAG文件——500行Python代码，30+个Task全部写在一个文件里，变量名是task1/task2/task3，没有注释没有文档。我花了整整两天才理清逻辑。后来我制定了一套DAG开发规范，新人接手时间从2天降到2小时。
原则一：任务粒度——一个Task做一件事
# 太粗（一个Task做所有事情）
# task: run_entire_etl
# 问题：失败了不知道哪一步出错，重跑要从头来
# 太细（每个SQL一个Task）
# task1: create_temp_table
# task2: insert_into_temp
# task3: update_stats
# task4: drop_temp_table
# 问题：Task太多调度开销大，DAG图看不清
# 适中（按逻辑步骤划分）
# task: ods_load_events (加载原始数据)
# task: dwd_clean_events (清洗明细)
# task: dws_aggregate_daily (聚合汇总)
# task: ads_export_report (输出报表)
# 每个Task是一个完整的逻辑单元，失败可独立重跑原则二：命名规范
| 维度 | 规范 | 好例子 | 坏例子 |
|---|---|---|---|
| DAG ID | 业务_频率 | daily_warehouse_etl | dag1 |
| Task ID | 层_操作_表名 | ods_load_events | task1 |
| 文件名 | 和DAG ID一致 | daily_warehouse_etl.py | my_dag.py |
| 变量 | 语义化 | events_df, user_count | df1, x |
# 命名模板
# DAG: {业务域}_{频率}_{描述}
# 例: ads_daily_revenue_report
# user_hourly_active_stats
# Task: {层}_{动作}_{对象}
# 例: ods_load_events
# dwd_clean_transactions
# dws_agg_daily_revenue
# ads_export_dashboard
# check_data_quality原则三：参数化——不硬编码
# 错误：硬编码日期
sql = "SELECT * FROM events WHERE dt = '2024-01-15'"
# 重跑其他日期要改代码！
# 正确：用Airflow模板变量
sql = "SELECT * FROM events WHERE dt = '{{ ds }}'"
# {{ ds }} 自动替换为逻辑执行日期
# 重跑任何日期都不用改代码
# 常用Airflow模板变量
# {{ ds }} 逻辑日期 2024-01-15
# {{ ds_nodash }} 20240115
# {{ prev_ds }} 前一天 2024-01-14
# {{ next_ds }} 后一天 2024-01-16
# {{ execution_date }} 完整时间戳
# 表名也参数化
TABLE_CONFIG = {
'events': {'source': 'ods_events', 'target': 'dwd_events'},
'users': {'source': 'ods_users', 'target': 'dwd_users'},
}
# 用循环批量生成Task，不要手写每个表的Task原则四：模板复用——工厂函数生成Task
# 工厂函数：批量生成同类型Task
def create_ods_load_task(table_name, source_path, dag):
return PythonOperator(
task_id=f'ods_load_{table_name}',
python_callable=load_ods,
op_kwargs={
'table': table_name,
'source': f's3://data/{source_path}/dt={{{{ ds }}}}/',
'target': f'ods_{table_name}',
},
dag=dag)
# 10张表只需要10行配置，不需要10段重复代码
tables = ['events','users','devices','orders','payments',
'sessions','clicks','installs','reviews','refunds']
ods_tasks = [create_ods_load_task(t, t, dag) for t in tables]
# 对比：不用工厂函数，10张表要写300+行重复代码
# 用工厂函数，10张表只写20行（配置+循环）DAG文件的组织结构
# 推荐的目录结构
# dags/
# ├── daily_warehouse/
# │ ├── dag.py # DAG定义（依赖关系）
# │ ├── tasks/
# │ │ ├── ods_tasks.py # ODS层Task定义
# │ │ ├── dwd_tasks.py # DWD层Task定义
# │ │ └── dws_tasks.py # DWS层Task定义
# │ ├── sql/
# │ │ ├── dwd_events.sql # SQL文件单独存放
# │ │ └── dws_revenue.sql
# │ └── config.yaml # 表配置、参数
# ├── hourly_realtime/
# │ └── ...
# └── common/
# ├── operators.py # 自定义Operator
# ├── alerts.py # 告警函数
# └── utils.py # 工具函数实战经验：DAG开发规范的效果
在某短视频平台制定DAG开发规范后：
新人接手时间：从2天降到2小时。命名规范+注释+文档让新人看名字就懂每个Task做什么。
代码量：10张表的ODS加载从300行降到50行（工厂函数）。新增一张表只需在配置里加一行。
Bug率：参数化后消除了硬编码日期导致的重跑错误。之前每月有2-3次因为忘改日期导致的数据事故，规范后降到0。
面试高频追问
追问一：一个DAG文件应该包含多少个Task？
经验值10-30个Task。少于5个说明粒度太粗可以拆成更细的步骤。超过50个说明DAG太复杂应该拆成多个子DAG（SubDag或TaskGroup）。Airflow 2.0+的TaskGroup可以把相关Task分组折叠，让DAG图更清晰。
追问二：SQL应该内联还是放外部文件？
短SQL（5行以内）可以内联在Task定义中。长SQL（10行以上）放外部.sql文件——方便Code Review、版本对比和SQL工具检查。Airflow的TemplatedBigQueryOperator等支持直接引用.sql文件。
追问三：怎么做DAG的Code Review？
和普通代码一样走Git PR流程。Review重点：① 命名是否规范；② 依赖关系是否正确；③ 是否幂等（INSERT OVERWRITE）；④ 是否参数化（不硬编码日期）；⑤ 是否有重试和告警配置。我在出海硬件品牌建了一个DAG Review CheckList，每个PR必须逐项勾选。
总结
DAG设计讲四原则：① 粒度适中（一个Task一件事，10-30个Task/DAG）；② 命名规范（层_操作_表名，看名字知功能）；③ 参数化（用{{ ds }}不硬编码，用配置文件不硬编码表名）；④ 模板复用（工厂函数批量生成Task，10张表50行搞定不用300行）。能讲出DAG规范制定后新人接手从2天降到2小时的效果，面试官觉得你有团队级的工程治理能力。
下一篇聊任务超时和熔断：一个任务卡住了后面30个全堵了怎么办？
你们DAG有开发规范吗？评论区聊聊～
关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长