当前位置：首页 > news >正文

关键词解释：DAG 系统（Directed Acyclic Graph，有向无环图）

news 2025/11/8 7:38:27

看到这个名字我是很懵的什么是DAG？系统？查了一下才知道是有向无环图
从零开始学习 DAG 系统（Directed Acyclic Graph，有向无环图）。我会按照「由浅入深、理论+实践」的思路，分阶段带你掌握核心概念、应用场景和动手能力。

🧭 学习路线总览

阶段	目标	内容
第1步	理解基础概念	什么是图？什么是 DAG？数学定义与可视化
第2步	掌握关键性质	拓扑排序、路径、入度/出度、无环性判断
第3步	应用场景入门	工作流调度、数据处理、区块链等典型用例
第4步	动手实践	用 Python 构建简单 DAG，实现拓扑排序
第5步	深入系统案例	学习 Airflow / Spark / IOTA 中的 DAG 实现
第6步	扩展与思考	DAG 的局限性、与其他模型对比、前沿研究

✅ 第1步：理解 DAG 是什么？

1.1 图（Graph）基础

图 = 节点（顶点） + 边
有向图：边有方向（A → B ≠ B → A）
无环：不能从一个节点出发，沿着箭头走回自己

1.2 DAG 定义

DAG（Directed Acyclic Graph） 是一个有向图，且不存在任何环路。

✅ 例子：

A → B → C  ↘     ↗  D

这是一个 DAG：所有路径都是“向前”的，没有循环。

❌ 反例：

A → B → C  ↑    ↓  ← D ←

存在环（B→C→D→B），所以不是 DAG。

✅ 第2步：DAG 的关键性质

2.1 拓扑排序（Topological Sort）

将 DAG 的节点排成线性序列，使得所有依赖关系都满足“前驱在前”。
用途：任务调度、编译顺序、课程安排等。

📌 举例：
任务：洗衣服（W）、晾衣服（D）、买洗衣液（B）
依赖：W 依赖 B；D 依赖 W
DAG：B → W → D
拓扑排序结果：[B, W, D]

2.2 入度（In-degree）与出度（Out-degree）

入度：指向该节点的边数
出度：从该节点指出的边数
入度为 0 的节点通常是“起点”

2.3 如何判断一个图是不是 DAG？

方法1：尝试做拓扑排序，若成功 → 是 DAG
方法2：DFS 检测是否有回边（back edge）

✅ 第3步：DAG 的典型应用场景

领域	应用	DAG 的作用
任务调度	Airflow, Prefect	定义任务依赖，自动并行执行
大数据处理	Apache Spark	优化计算流程，避免重复计算
区块链	IOTA, Nano	替代链式结构，支持高并发交易
软件工程	Makefile, npm install	管理构建或依赖安装顺序
AI/ML	TensorFlow 1.x（静态图）	定义计算图，优化执行

💡 核心思想：“谁先做，谁后做” + “能并行就并行”

✅ 第4步：动手实践 —— 用 Python 实现一个简单 DAG

目标：构建 DAG 并输出拓扑排序

from collections import defaultdict, dequeclass DAG:def __init__(self):self.graph = defaultdict(list)self.in_degree = defaultdict(int)def add_edge(self, u, v):"""添加边 u -> v"""self.graph[u].append(v)self.in_degree[v] += 1if u not in self.in_degree:self.in_degree[u] = 0def topological_sort(self):# 初始化队列：所有入度为0的节点queue = deque([node for node in self.in_degree if self.in_degree[node] == 0])result = []while queue:node = queue.popleft()result.append(node)for neighbor in self.graph[node]:self.in_degree[neighbor] -= 1if self.in_degree[neighbor] == 0:queue.append(neighbor)# 检查是否有环if len(result) != len(self.in_degree):raise ValueError("图中存在环，不是 DAG！")return result# 示例使用
dag = DAG()
dag.add_edge("买洗衣液", "洗衣服")
dag.add_edge("洗衣服", "晾衣服")
dag.add_edge("买衣架", "晾衣服")print("拓扑排序结果：", dag.topological_sort())

输出：

拓扑排序结果： ['买洗衣液', '买衣架', '洗衣服', '晾衣服']

✅ 你已经实现了一个最简 DAG 调度器！

✅ 第5步：深入真实系统（选学）

5.1 Apache Airflow（工作流调度）

用 Python 代码定义 DAG：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetimewith DAG('my_dag', start_date=datetime(2025, 1, 1)) as dag:task1 = BashOperator(task_id='task1', bash_command='echo "Hello"')task2 = BashOperator(task_id='task2', bash_command='sleep 5')task1 >> task2  # task1 必须在 task2 之前完成