DAG 文件是什么
Airflow 里的一个工作流就是一个 Python 文件，放在 DAGS_FOLDER 指向的目录下（默认 $AIRFLOW_HOME/dags），由调度器按 dag_dir_list_interval 周期扫描解析。文件里用上下文管理器声明 DAG 的标识、调度周期、起始日期与默认参数，任务则是文件中的算子实例。
解析期与运行期的区别
这是最容易踩的坑：调度器每隔几分钟就会导入一次 DAG 文件，此时文件顶层代码会真的执行。所以文件里只应该做声明——建 DAG、建算子、连依赖；任何查数据库、调接口、算变量的动作都要放进算子的可调用对象或模板里，运行时才执行。
基本信息
| 定义方式 | Python 文件，放在 dags 目录下被自动扫描 |
|---|---|
| 扫描间隔 | dag_dir_list_interval 控制，默认 300 秒 |
| 关键参数 | dag_id、schedule、start_date、catchup、tags、default_args |
| 解析失败 | 文件报错则该 DAG 在界面中不出现，日志在调度器日志里 |
| 生效方式 | 保存文件后等下一次扫描，无需重启调度器 |
关键参数读法
| 参数 | 含义 | 注意 |
|---|---|---|
| dag_id | 工作流唯一标识 | 改名等于新建，历史运行记录不会跟随 |
| schedule | 调度周期，cron 表达式或预设别名 | 旧版本参数名为 schedule_interval |
| start_date | 调度起点，第一个周期由此推算 | 改动会影响所有历史周期的判定 |
| catchup | 是否补跑 start_date 到当前之间的周期 | 新 DAG 建议设为 False，避免一次性拉起大量回填 |
| default_args | 重试次数、超时、所有者等公共参数 | 重试次数过大会掩盖真实故障 |
典型使用场景
- 日批链路：一个 DAG 描述当天从抽取到导出的完整流程
- 小时级任务：按小时周期触发，配合分区参数写入不同的数据分区
- 跨系统编排：把 SQL 任务、脚本任务与外部系统检查节点混编
- 局部重跑：出错后只重跑失败的任务子树，成功部分不动
- 按需触发的临时任务：用 schedule=None 定义，仅手工或接口触发
最小 DAG 示例
最小 DAG 示例
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
# 顶层只做声明：建 DAG、建任务、连依赖，不执行任何取数动作
dag = DAG(
dag_id="daily_orders",
start_date=datetime(2026, 1, 1),
schedule="0 2 * * *", # 每天 02:00 触发
catchup=False, # 不回填历史周期
default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
tags=["dw", "orders"],
)
extract = BashOperator(task_id="extract", bash_command="echo extract ok", dag=dag)
load = BashOperator(task_id="load", bash_command="echo load ok", dag=dag)
extract >> load # 上游成功后才调度下游模板注入业务日期
模板注入业务日期
# Airflow 会把命令中的模板变量在运行时替换为对应周期的日期
# ds 是周期开始日期，已格式化为 YYYY-MM-DD；写死在命令里会在补数时算错
bash_command='python /opt/etl/load.py --dt {{ ds }}'DAG 文件是「代码」而不是「配置」：建议纳入版本管理、区分开发与生产、改动前在测试环境解析一遍，避免一个语法错误让整个 DAG 从界面上消失。
在文件顶层写数据库连接或 HTTP 请求，解析期就会真的执行：轻则拖慢调度器扫描，重则每次扫描都产生一次真实的业务调用。