4、DAG调度模型：有向无环图在任务编排中的应用
聊到任务调度，绕不开的一个概念就是DAG。有向无环图，英文叫Directed Acyclic Graph。说白了，它就是一套任务编排的规则——任务之间有方向、有依赖，但不能有循环依赖。
我刚开始接触调度系统时，总觉得DAG就是个花哨的术语。后来在做一个实时数仓项目时，因为任务循环依赖导致整个管道死锁，数据延迟了整整两天。嗯，从那以后，我对DAG的理解就深刻多了。
4.1 什么是DAG？为什么它适合任务编排？
DAG由两部分组成：节点（Vertex）和边（Edge）。节点代表一个任务，边代表任务之间的依赖关系。方向从上游指向下游，表示上游任务完成后才能启动下游任务。
你想想看，现实中的数据处理流程，天然就是有向无环的：先抽取数据，再清洗，然后转换，最后加载。每一步都依赖上一步的结果。如果出现循环，比如A依赖B、B依赖C、C又依赖A，那整个流程就永远跑不完。
核心要点：DAG保证了任务执行的确定性和可终止性。只要图是无环的，就一定能找到一条从起点到终点的执行路径。
4.2 DAG的核心概念与术语
在实际项目中，我们经常用这几个术语来描述DAG。我建议你先把它们记牢，后面写代码时会频繁遇到。
| 术语 | 含义 | 举例 |
|---|---|---|
| 节点（Node） | 一个具体的任务单元 | 数据抽取任务、SQL计算任务 |
| 边（Edge） | 任务间的依赖关系 | 任务A完成后才能执行任务B |
| 入度（In-degree） | 指向该节点的边数 | 入度为0表示没有前置依赖 |
| 出度（Out-degree） | 从该节点出发的边数 | 出度为0表示没有下游任务 |
| 拓扑排序（Topological Sort） | 将DAG节点排成线性序列 | 保证所有依赖关系被满足 |
| 关键路径（Critical Path） | 耗时最长的路径 | 决定整个DAG的最短完成时间 |
我的习惯：在设计DAG时，我会先画出所有节点，再标注入度和出度。入度为0的节点就是起始任务，出度为0的就是终点任务。这样一眼就能看出整个流程的骨架。
4.3 DAG的构建与拓扑排序
构建DAG其实不复杂。核心就两步：定义节点，定义边。但真正难的是保证无环。
我曾经在一个项目中，因为配置错误，让任务C依赖了任务D，而任务D又间接依赖了任务C。结果调度器一直卡在等待状态，日志里全是"Deadlock detected"。排查了整整一个下午才发现是循环依赖。
下面是一个简单的DAG构建示例，用Python实现：
class DAGNode:
def __init__(self, task_id, name):
self.task_id = task_id
self.name = name
self.dependencies = [] # 上游任务列表
class DAG:
def __init__(self, dag_id):
self.dag_id = dag_id
self.nodes = {}
def add_node(self, node):
self.nodes[node.task_id] = node
def add_edge(self, from_id, to_id):
# from_id 是上游，to_id 是下游
if from_id in self.nodes and to_id in self.nodes:
self.nodes[to_id].dependencies.append(from_id)
def topological_sort(self):
# 使用Kahn算法进行拓扑排序
in_degree = {tid: 0 for tid in self.nodes}
for tid, node in self.nodes.items():
for dep in node.dependencies:
in_degree[tid] += 1
queue = [tid for tid, deg in in_degree.items() if deg == 0]
result = []
while queue:
tid = queue.pop(0)
result.append(tid)
for node in self.nodes.values():
if tid in node.dependencies:
in_degree[node.task_id] -= 1
if in_degree[node.task_id] == 0:
queue.append(node.task_id)
if len(result) != len(self.nodes):
raise ValueError("检测到循环依赖！DAG无效。")
return result
# 使用示例
dag = DAG("etl_pipeline")
dag.add_node(DAGNode("extract", "数据抽取"))
dag.add_node(DAGNode("clean", "数据清洗"))
dag.add_node(DAGNode("transform", "数据转换"))
dag.add_node(DAGNode("load", "数据加载"))
dag.add_edge("extract", "clean")
dag.add_edge("clean", "transform")
dag.add_edge("transform", "load")
print(dag.topological_sort())
# 输出: ['extract', 'clean', 'transform', 'load']避坑指南：我曾经在拓扑排序时忘记检查结果长度，导致循环依赖的任务被静默跳过。数据跑出来全是错的。记住：拓扑排序后一定要校验结果长度是否等于节点总数，否则就是有环。
4.4 DAG在任务调度中的执行策略
有了DAG结构，调度器怎么执行呢？常见的有三种策略：
- 串行执行：按拓扑排序的顺序，一个接一个跑。简单但慢。
- 并行执行：入度为0的任务同时启动，完成后更新下游入度，再启动新的可执行任务。效率高，但资源消耗大。
- 混合执行：设置并行度上限，在资源允许范围内尽量并行。这是生产环境最常用的方式。
我个人习惯用并行执行，但会配合资源池限制并发数。比如一个DAG有20个任务，我设置最大并行度为5，这样既不会撑爆集群，又能充分利用资源。
4.5 DAG与血缘关系的天然联系
这里要重点说一下DAG和血缘的关系。你想想看，DAG的边不就是数据流向吗？上游任务产出数据，下游任务消费数据。这种依赖关系，本质上就是数据血缘的体现。
我在做数据治理时，经常把DAG的边直接映射为血缘关系。比如任务A生成表t1，任务B读取t1并生成表t2。那么在血缘系统中，t1→t2这条线就自动建立起来了。
关键洞察：DAG是任务维度的依赖关系，血缘是数据维度的依赖关系。两者是同一枚硬币的两面。把DAG的边加上数据资产信息，就变成了完整的血缘图谱。
4.6 DAG调度模型的核心流程图
下面这张图展示了DAG从构建到执行再到血缘映射的完整流程。我建议你仔细看看，尤其是"循环检测"和"血缘映射"这两个环节，很多项目都在这里踩坑。
4.7 生产环境中的DAG最佳实践
最后，分享几个我在生产环境中总结的经验：
- 粒度控制：一个DAG的节点数建议控制在50个以内。太多会导致调度器压力大，排查问题也困难。
- 超时设置：每个节点都要设置超时时间。我曾经遇到过任务卡死，因为没有超时，整个DAG挂了三天。
- 重试机制：对于网络抖动等临时故障，设置2-3次重试。但幂等性要做好，否则重试会导致数据重复。
- 版本管理：DAG定义要纳入版本控制。我习惯把每个DAG的JSON配置放在Git里，方便回滚和审计。
- 血缘标注：在DAG的边上标注数据资产信息（表名、字段名），这样调度和治理就打通了。
我的建议：刚开始做DAG调度时，别追求大而全。先画一个小DAG，跑通拓扑排序和并行执行，再逐步加功能。一步到位往往容易出问题。
DAG调度模型，说白了就是给任务排个队，保证谁先谁后不出错。但真正用好它，需要理解拓扑排序的原理、并行执行的策略，以及如何把DAG的边变成血缘关系。嗯，这一块内容不少，但掌握了就是调度系统的基础能力。