Airflow 不构建工作流，只可靠执行用户用 Python 显式定义的依赖逻辑；企业级能力取决于 DAG 代码设计：职责原子化、连接外置、动态分支、初始化逻辑移入 execute()。
直接说结论：Airflow 本身不“构建”复杂工作流，它只负责**可靠地执行你用 Python 显式定义出来的依赖逻辑**。所谓“企业级”，核心不在 Airflow 多高级，而在于你如何用 DAG 文件把任务边界划清、状态管住、失败兜住、数据传准——这些全靠代码设计，不是靠配置开关。
为什么不能直接写 PythonOperator 堆满整个 DAG？
常见错误是把所有业务逻辑塞进一个 PythonOperator：读数据库、调 API、写文件、发邮件全在一个函数里。结果一出错根本不知道卡在哪步，重跑要全量执行，日志里全是混合输出，调试成本爆炸。
真正可行的做法是按「原子职责」拆分 Task：
- 每个
PythonOperator只做一件事：比如fetch_orders_from_mysql()或validate_daily_file_exists() - 用
BashOperator执行带副作用的命令（如清理临时目录），避免在 Python 函数里混用os.system - 对需要强一致性的操作（如写入数仓），优先用
PostgresOperator或BigQueryInsertJobOperator，而不是手写psycopg2连接 - 所有跨 Task 的中间数据，小量用
XCom（比如文件路径、校验和），大量必须存 S3/HDFS 并只传 URI
default_args 不只是设重试次数，它是 DAG 的契约起点
很多人把 default_args 当成可选配置，其实它决定了整个 DAG 的健壮基线。漏掉关键项会导致任务看似运行成功，实则埋雷：
- 必须显式设
'retries': 2和'retry_delay': timedelta(minutes=3)，否则网络抖动就让任务永久失败 -
'sla': timedelta(hours=1)要和实际 SLA 对齐，不是随便填；触发后会进SLA Miss视图，但不会自动终止任务 -
'email_on_failure'建议配成团队邮箱或 Slack webhook，别只写个人邮箱——人休假时告警就失效了 -
'depends_on_past': True要慎用：它要求前序日期任务成功才触发当前任务，适合严格串行场景（如逐日补数），但会阻断并行修复
DAG 文件里绝对不能出现硬编码的连接信息或路径
看到 host='10.1.2.3', port=5432, password='xxx' 或 /tmp/data.csv 这种写法，基本可以判定这个 DAG 过不了生产评审。
正确做法是全部外置：
- 数据库连接统一走 Airflow UI 的
Admin → Connections页面创建，Task 中用conn_id='postgres_prod'引用 - S3 路径用
Variable.get('s3_raw_bucket')，变量值在 UI 的Admin → Variables里维护 - 敏感参数（如 API token）绝不能出现在 DAG 文件或
default_args里，必须通过 Connection 的extra字段 JSON 存储，并在 Operator 中用hook.get_connection(conn_id).extra_dejson.get('api_token')安全读取 - 所有路径拼接用
os.path.join(Variable.get('base_path'), 'daily', '{{ ds }}')，利用 Jinja 模板注入日期，避免手动字符串格式化
复杂依赖不是靠多写几个 task1 >> task2 >> task3 实现的
真实企业场景里，任务间关系远不止线性链。比如：“只有当订单表和用户表都更新完成，才触发合并计算；但如果用户表延迟超 2 小时，就跳过它，只用订单表跑降级逻辑”。这种动态分支，PythonOperator 必须返回明确信号，再配合 BranchPythonOperator 或 ShortCircuitOperator 控制流向。
示例片段：
def decide_branch(**context):
# 检查上游两个表的元数据
order_updated = context['dag_run'].conf.get('order_updated', False)
user_updated = context['dag_run'].conf.get('user_updated', False)
if user_updated:
return ['merge_task']
else:
return ['degrade_task']
<p>branch_task = BranchPythonOperator(
task_id='branch_check',
python_callable=decide_branch,
dag=dag
)
</p>
注意：BranchPythonOperator 返回的是 task_id 列表，不是布尔值；下游必须用 trigger_rule='none_failed_or_skipped' 接收分支结果，否则未被选中的路径会卡在 upstream_failed 状态。
最易被忽略的一点：DAG 文件每次被调度器解析时都会重新执行顶层代码。所以不要在 DAG 文件里做耗时操作（如初始化大模型、建立长连接），所有初始化逻辑必须封装进 Operator 的 execute() 方法内——那里才是任务真正运行的地方。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
