下游任务等了2小时上游还没跑完——依赖管理和优先级

任务调度手册第 3 题约 7 分钟读完

先说结论

依赖管理有三个层面：① 任务内依赖（同一个DAG内的Task间依赖，用>>或set_upstream配置）；② 跨DAG依赖（不同DAG之间的依赖，用ExternalTaskSensor或数据就绪检测）；③ 数据依赖（下游任务需要的数据是否已经产出，比如Hive分区是否存在）。最可靠的是数据依赖——不依赖任务的执行状态而是检查数据是否真的到了。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

依赖管理是调度系统最核心的能力——配错了依赖，要么下游读到旧数据（依赖缺失），要么下游一直等不到触发（依赖配多了）。在某短视频平台用Airflow时，DWS层的日报任务等了2小时不触发——排查发现依赖了一个已经废弃的DWD任务（那个任务因为上游数据源下线一直失败），但没人去清理这个依赖。一个废弃任务堵住了整条链路。

强依赖 vs 弱依赖

依赖类型	含义	失败影响	适用场景
强依赖	上游必须成功才能执行下游	上游失败则下游不跑	DWD依赖ODS、DWS依赖DWD
弱依赖	上游失败也不影响下游执行	上游失败下游照跑	非核心辅助数据、可降级的维度
自依赖	当前周期依赖上一周期的自己	昨天没跑完今天不跑	增量ETL、累积快照表

# Airflow中的依赖配置
 
# 强依赖（默认——上游成功才触发下游）
ods_task >> dwd_task  # ODS成功后才跑DWD
 
# 弱依赖（用TriggerRule实现）
from airflow.utils.trigger_rule import TriggerRule
dws_task = PythonOperator(
    task_id='dws_daily',
    python_callable=build_dws,
    trigger_rule=TriggerRule.ALL_DONE  # 上游全部完成（不管成功失败）
    # 其他选项：
    # ALL_SUCCESS（默认）：所有上游成功
    # ALL_DONE：所有上游完成（含失败）
    # ONE_SUCCESS：至少一个上游成功
    # NONE_FAILED：没有上游失败（允许跳过）
)
 
# 自依赖（wait_for_downstream + depends_on_past）
daily_task = PythonOperator(
    task_id='daily_incremental',
    depends_on_past=True,  # 昨天的这个Task必须成功今天才能跑
)

跨DAG依赖——最容易出问题的地方

# 场景：DAG_A产出DWD表，DAG_B消费DWD表产出DWS
# DAG_B需要等DAG_A完成才能跑
 
# 方式一：ExternalTaskSensor（监听另一个DAG的Task状态）
from airflow.sensors.external_task import ExternalTaskSensor
wait_for_dwd = ExternalTaskSensor(
    task_id='wait_for_dwd_orders',
    external_dag_id='dag_a_dwd',
    external_task_id='dwd_orders',
    execution_date_fn=lambda dt: dt,  # 等同一天的执行
    timeout=7200,  # 最多等2小时
    mode='reschedule',  # 释放Worker去做其他事
)
wait_for_dwd >> dws_task
 
# 方式二：数据就绪检测（更可靠——第8篇详讲）
# 不依赖DAG状态，而是检查数据分区是否存在
# 更解耦，DAG_A重构不影响DAG_B

ExternalTaskSensor的坑：① execution_date必须精确匹配——如果两个DAG的schedule不一样（一个daily一个hourly），需要用execution_date_fn手动映射，容易配错；② 如果被依赖的DAG改了task_id，Sensor就找不到了——没有人通知你；③ 占Worker资源——Sensor在等的时候也占一个Worker Slot（mode='reschedule'可以缓解但不彻底）。

任务优先级——关键任务先跑

# Airflow支持给Task设优先级
# priority_weight越大越优先
 
# 核心报表（老板每天早上看的）——最高优先级
ads_boss_report = PythonOperator(
    task_id='ads_boss_report',
    priority_weight=10,  # 高优先级
    pool='default_pool')
 
# 普通ETL——默认优先级
dwd_logs = PythonOperator(
    task_id='dwd_logs',
    priority_weight=1,   # 默认优先级
    pool='default_pool')
 
# 当Worker不够时，优先执行priority_weight高的Task
# 保证核心报表按时产出
 
# Pool（资源池）：限制同时执行的Task数量
# 防止某类任务（如Spark作业）占满所有Worker

实战踩坑：废弃任务堵住整条链路

在某短视频平台，有一个DWD任务（dwd_third_party）依赖一个第三方数据源。后来第三方合作终止了，数据源不再更新，这个DWD任务每天都失败。但没人去清理依赖关系——DWS层还依赖着这个已经废弃的DWD任务。

结果：DWS层的日报任务每天等dwd_third_party成功才触发，但它永远不会成功——日报就一直不跑。运营每天早上看不到报表，以为是'数据延迟'。排查了才发现是一个废弃任务堵住了链路。

修复：① 把dwd_third_party从DWS的依赖中移除（或改为弱依赖ALL_DONE）；② 把废弃任务设为skip状态，不再占用调度资源；③ 建立依赖审计机制——每月Review一次所有跨DAG依赖，清理废弃链路。

教训：依赖关系需要定期维护——业务变化了依赖也要跟着变。废弃的上游任务如果不清理，就像管道里的堵塞物，迟早堵住下游。

面试高频追问

追问一：怎么解决循环依赖？

DAG不允许循环依赖——如果发现A依赖B、B又依赖A，说明设计有问题。解决方法：① 拆分——把A拆成A1和A2，A1不依赖B，B依赖A1，A2依赖B；② 合并——如果A和B确实需要交互，合并成一个Task内部处理；③ 引入中间表——A先写中间表，B读中间表，A再读B的输出。

追问二：depends_on_past和wait_for_downstream有什么区别？

depends_on_past：当前Task等上一周期的同一个Task成功。wait_for_downstream：当前Task等上一周期的同一个Task及其所有下游都成功。wait_for_downstream更严格——它确保上一周期的整条链路都完成了才开始新周期。对于需要严格有序的增量ETL用wait_for_downstream更安全。

追问三：ExternalTaskSensor和数据就绪检测哪个好？

数据就绪检测更好——它检查的是'数据产出了没有'（比如Hive分区是否存在）而不是'任务跑没跑'。两者的区别：ExternalTaskSensor依赖任务状态（任务可能成功但数据没写对），数据就绪检测依赖数据状态（数据真的到了才触发）。数据就绪检测更解耦——上游DAG怎么重构不影响下游，只要数据产出就行。

总结

依赖管理面试讲三层：① 三种依赖类型（强依赖=必须成功、弱依赖=完成即可、自依赖=等上一周期）；② 跨DAG依赖（ExternalTaskSensor vs 数据就绪检测，后者更解耦更可靠）；③ 优先级和资源池（priority_weight保证核心任务优先、Pool限制并发）。能讲出废弃任务堵住链路的踩坑和依赖审计机制，面试官会觉得你有生产级的调度运维经验。

下一篇聊失败处理：凌晨3点ETL失败了早上才发现——重试策略和告警怎么配？

你们DAG的依赖关系复杂吗？踩过什么依赖的坑？评论区聊聊～

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

刷题进度保存在当前浏览器