Airflow 任务依赖与调度最佳实践DAG 设计精要一、DAG 不是流程图从依赖本质说起很多团队刚接触 Airflow 时把它当成一张普通的流程图来画。结果很快就踩坑。DAG 的核心不是「画得好看」而是「依赖关系正确且可重放」。这是调度系统的命门。如果依赖写错下游任务会在上游数据还没就绪时就启动。脏数据就这样流向下游。循环依赖则更隐蔽。Airflow 在解析阶段就会拒绝成环的 DAG但定位成本高。还有一些团队喜欢把所有任务塞进一个巨型 DAG。一旦某节点失败整图都要重算。合理的做法是按业务边界拆分 DAG。让每个 DAG 内聚、外部通过数据集或传感器解耦。本文聚焦三件事DAG 该怎么设计、依赖如何管理、重试与告警怎样才不误伤。可观测性设计这个环节常被低估。DAG 跑通只是开始能看懂才关键。日志要结构化关键节点要打点。否则一次失败排障同学要在海量日志里捞线索。SLA 也要提前约定。哪些任务必须在几点前完成超时即触发值班而非等下游投诉。这些工程习惯比任何炫技的算子都更能决定调度系统在凌晨是否把你叫醒。二、依赖管理与触发链路从上游到下游依赖管理要区分「硬依赖」与「软依赖」并善用 Dataset 做跨 DAG 的数据驱动触发。硬依赖是最常见的一类。任务 B 必须等任务 A 成功才允许开始执行。这是强约束。软依赖则更灵活。比如任务 B 等上游数据到达即触发不必关心上游任务是否成功。Airflow 2.4 之后引入的 Dataset让跨 DAG 触发变得声明式不再依赖人工计时错位。序列化任务的状态也要管好。失败要有重试重试耗尽要有告警告警要能直达负责人。在 Data-aware 调度模式下上下游 DAG 通过数据集实现衔接。上游任务完成事实表写入后会产出特定的 Dataset 对象。该对象作为触发条件会自动激活下游的指标汇总 DAG 与质量校验 DAG。若指标任务执行失败系统会执行重试策略若质检任务失败则直接触发告警通知负责人。依赖要尽量窄。一个任务依赖的上游越少失败时的爆炸半径就越小排查也越快。跨 DAG 触发优先用 Dataset而不是用长轮询传感器。后者既耗资源又容易误判超时。三、生产级 DAG 与重试告警代码下面给出一段生产级 DAG 定义。它包含重试策略、超时、告警回调与空数据兜底。代码强调把失败处理显式化避免任务静默成功却产出了空结果误导下游。from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.datasets import Dataset from airflow.utils.email import send_email def _on_failure(context): 任务失败回调发送告警邮件附上执行上下文便于快速定位。 dag_id context.get(dag).dag_id task_id context.get(task).task_id ts context.get(ts) send_email( to[oncallcorp.com], subjectf[Airflow 失败] {dag_id}.{task_id}, htmlfp任务在 {ts} 失败请查看日志。/p, ) def extract_orders(**kwargs): 抽取订单数据空结果时显式抛错避免下游误信空表为正常。 rows _query_source() # 假设返回列表 if not rows: raise ValueError(源表返回空疑似上游延迟主动失败以便重试) return len(rows) def _query_source(): # 占位真实场景替换为数据库/数仓查询返回行列表 raise NotImplementedError(请实现 _query_source 以对接真实数据源) with DAG( dag_idorders_summary_v2, start_datedatetime(2026, 7, 1), scheduledaily, catchupFalse, # 禁止历史补跑避免突发负载 default_args{ retries: 3, # 失败重试三次 retry_delay: timedelta(minutes5), execution_timeout: timedelta(hours1), # 超时熔断防挂死 on_failure_callback: _on_failure, }, tags[dwd, orders], ) as dag: t_extract PythonOperator( task_idextract_orders, python_callableextract_orders, outlets[Dataset(dataset://fact_orders)], # 声明产出数据集 ) t_extract重试次数要克制。无限重试会卡住调度器并掩盖真实故障三到五次通常足够。执行超时务必设置。否则一个慢查询可能长期占用 worker拖垮整批任务的并发。四、边界条件与权衡何时该信、何时该拦Airflow 调度也有边界。第一个边界是「任务粒度」。过细会产生海量小任务。海量小任务会压垮调度器元数据库让 UI 卡顿、解析变慢整体稳定性下降。第二个边界是「跨系统依赖」。Airflow 擅长编排不擅长做重型数据计算本身。第三个边界是「时间敏感型触发」。纯定时在上下游波动大时容易错位或空跑。在 Trade-offs 上我们主张「声明式 Dataset 优先于隐式计时」。让数据说话。但也别盲目追求细粒度解耦。拆得太碎可观测性反而变差排障成本直线上升。适用场景批处理编排、跨系统任务依赖、需要重试与可观测性的 ETL 链路。禁用场景毫秒级实时流处理、把 Airflow 当计算引擎硬算 TB 级数据、无监控告警。还要注意版本演进带来的范式变化。Dataset 触发虽好但要求调度器版本足够新。老旧集群若不支持数据感知调度应继续用传感器或外部编排做跨 DAG 衔接。不要为了追新特性而强行升级。稳定压倒一切调度系统是下游所有数据的节拍器。最后提醒一点catchup 默认开启时新 DAG 上线会疯狂补跑历史务必显式关闭。五、总结Airflow 的精髓在依赖正确与可重放而不在流程图是否花哨。这是调度设计的根本。依赖要分清硬软跨 DAG 优先用 Dataset 声明式触发避免长轮询传感器的资源浪费。生产 DAG 必须显式配置重试、超时与失败告警并对空结果主动失败以防误导下游。粒度要平衡太粗爆炸半径大太细压垮调度器。让数据驱动而不是让计时猜测。
