3. DAG（有向无环图）原理

聊到任务调度，DAG 是个绕不开的概念。我刚开始接触调度系统时，觉得 DAG 就是个花哨的数学名词。后来踩了几个坑才明白——没有 DAG，任务调度就是一团乱麻。

说白了，DAG 就是一套规则，告诉系统「谁先干，谁后干，谁和谁能一起干」。今天咱们就把它的定义、应用场景，还有怎么构建和解析，一次性讲透。

3.1 DAG 的定义

DAG 的全称是 Directed Acyclic Graph，中文叫「有向无环图」。拆开来看：

• 有向（Directed）：每条边都有方向，表示依赖关系。A → B 意味着 A 完成后 B 才能开始。
• 无环（Acyclic）：图中不能有环路。你不能从 A 出发，绕一圈又回到 A。
• 图（Graph）：由节点（Vertex）和边（Edge）组成的数据结构。

举个例子，一个简单的 ETL 流程：

数据采集 → 数据清洗 → 数据转换 → 数据加载

这就是一个典型的 DAG。每个步骤是一个节点，箭头表示依赖方向。你想想看，如果数据还没采集完就开始清洗，那结果肯定是一团糟。

3.2 DAG 在任务调度中的应用

在实际调度系统中，DAG 解决了一个关键问题：如何确定任务的执行顺序。

我记得在做一个实时数仓项目时，有 200 多个任务需要调度。如果用手工配置依赖关系，光维护就够呛。DAG 帮我们做了三件事：

1. 依赖管理：明确每个任务的前置条件
2. 并行执行：没有依赖关系的任务可以同时跑
3. 异常处理：某个节点失败后，只影响下游依赖它的任务

来看一个实际的数据湖调度场景：

任务名称	依赖任务	执行内容
ods_log	无	采集原始日志到 ODS 层
ods_db	无	采集业务数据库到 ODS 层
dwd_clean	ods_log, ods_db	数据清洗与标准化
dws_agg	dwd_clean	轻度汇总聚合
ads_report	dws_agg	生成报表数据

这个 DAG 的结构很清晰：ods_log 和 ods_db 可以并行采集，等它们都完成后，dwd_clean 才开始清洗。清洗完再做聚合，最后出报表。

3.3 DAG 的构建与解析

构建 DAG 有两种常见方式：代码定义和配置文件定义。我个人的习惯是，小规模用代码，大规模用配置文件。

3.3.1 代码构建 DAG

以 Python 为例，用邻接表来表示 DAG：

class DAG:
    def __init__(self):
        self.nodes = {}  # 节点名 → 节点对象
        self.edges = {}  # 节点名 → 下游节点列表
    
    def add_node(self, name, task=None):
        self.nodes[name] = task
        if name not in self.edges:
            self.edges[name] = []
    
    def add_edge(self, from_node, to_node):
        if from_node not in self.nodes:
            self.add_node(from_node)
        if to_node not in self.nodes:
            self.add_node(to_node)
        self.edges[from_node].append(to_node)
    
    def get_upstream(self, node):
        """获取某个节点的所有上游依赖"""
        upstream = []
        for parent, children in self.edges.items():
            if node in children:
                upstream.append(parent)
        return upstream
    
    def topological_sort(self):
        """拓扑排序，确定执行顺序"""
        in_degree = {n: 0 for n in self.nodes}
        for parent, children in self.edges.items():
            for child in children:
                in_degree[child] += 1
        
        queue = [n for n, d in in_degree.items() if d == 0]
        result = []
        
        while queue:
            node = queue.pop(0)
            result.append(node)
            for child in self.edges.get(node, []):
                in_degree[child] -= 1
                if in_degree[child] == 0:
                    queue.append(child)
        
        if len(result) != len(self.nodes):
            raise ValueError("图中存在环，无法进行拓扑排序")
        return result

这段代码的核心是 拓扑排序。它通过计算每个节点的入度（有多少个上游依赖），从入度为 0 的节点开始，一层层往下推进。

3.3.2 配置文件构建 DAG

对于生产环境，我更推荐用 YAML 或 JSON 来定义 DAG。这样便于版本管理和自动化生成：

# dag_etl.yaml
dag:
  name: "data_lake_etl"
  schedule: "0 2 * * *"
  nodes:
    - name: ods_log
      command: "spark-submit ods_log.py"
    - name: ods_db
      command: "spark-submit ods_db.py"
    - name: dwd_clean
      command: "spark-submit dwd_clean.py"
      depends_on: [ods_log, ods_db]
    - name: dws_agg
      command: "spark-submit dws_agg.py"
      depends_on: [dwd_clean]
    - name: ads_report
      command: "spark-submit ads_report.py"
      depends_on: [dws_agg]

解析这个配置文件时，调度系统会：

1. 读取所有节点，建立节点索引
2. 根据 depends_on 字段构建边关系
3. 执行拓扑排序，生成执行计划
4. 按层级分发任务到执行器

3.3.3 DAG 的可视化

光看代码和配置还不够直观。下面这张图展示了我们刚才那个 ETL 流程的 DAG 结构：

这张图里，ods_log 和 ods_db 是并行节点，它们没有依赖关系。而 dwd_clean 必须等两个上游都完成后才能启动。这就是 DAG 的精髓——用图结构表达复杂的依赖关系。

3.4 DAG 的常见陷阱

做了这么多年调度系统，我总结了几条 DAG 设计时容易踩的坑：

• 环状依赖：A 依赖 B，B 依赖 C，C 又依赖 A。这种配置会让调度器死循环。
• 孤立节点：某个节点没有任何依赖，也没有被任何节点依赖。它会被调度，但可能不是你想要的行为。
• 层级过深：DAG 的深度（最长路径）太长，会导致整体执行时间被拉长。
• 扇出过大：一个节点下游有几十个节点，这个节点一旦失败，影响面太大。