3、任务依赖管理：DAG（有向无环图）在日志任务编排中的应用

日志任务不是孤岛。你想想看，一条日志从采集、解析、清洗、聚合到最终写入存储，中间要经过好几个环节。这些环节之间有严格的先后顺序——解析必须在采集之后，聚合必须在清洗之后。如果乱序执行，数据就全乱了。

我刚开始做日志平台的时候，用的是简单的「先来后到」队列。后来发现根本不行。有一次上游采集任务挂了，下游的解析任务还在傻等，结果整个管道堵死，数据延迟了整整两个小时。那次事故之后，我彻底转向了 DAG 编排。

为什么是 DAG，而不是别的？

说白了，DAG 就是一张有向无环图。有向，代表任务之间有明确的方向；无环，代表不会出现 A 依赖 B、B 又依赖 A 的死循环。

我见过不少团队用「状态机」或者「硬编码脚本」来管理依赖。结果呢？状态机一复杂就变成意大利面条，硬编码脚本改一次依赖要改三处代码。DAG 的好处在于：

• 依赖关系可视化——图结构天然适合展示上下游
• 无环保证——数学上杜绝死锁
• 并行度最大化——没有依赖关系的任务可以同时跑
• 容错粒度细——某个节点失败，只影响它的下游，不波及全链路

DAG 在日志任务中的典型结构

我们来看一个真实的日志处理 DAG。假设你有一个电商平台的访问日志，需要经过采集、解析、清洗、维度补全、聚合、写入六个阶段。画成 DAG 是这样的：

这张图里有个关键点：清洗节点之后，维度补全、聚合、写入三个任务没有相互依赖，所以可以并行执行。但聚合又依赖维度补全的结果，所以聚合必须等维度补全跑完才能启动。

我在项目中遇到过类似场景。当时有个日志管道，清洗之后要同时做「IP 地理信息补全」和「用户画像关联」。这两个任务互相独立，但后续的「地域聚合」必须等地理信息补全完成。用 DAG 表达这种「部分并行、部分串行」的关系，比硬编码优雅太多了。

DAG 的核心概念：节点、边、入度

要理解 DAG 调度，先记住三个基本概念：

概念	含义	日志场景举例
节点（Node）	一个具体的任务单元	日志解析任务、数据清洗任务
边（Edge）	任务之间的依赖关系	解析 → 清洗（解析完成后才能清洗）
入度（In-degree）	指向该节点的边的数量	聚合节点入度为 2（依赖清洗和维度补全）

调度器怎么知道哪个任务先跑？看入度。入度为 0 的节点就是「根任务」，没有前置依赖，可以立即执行。执行完后，把它从图中移除，更新下游节点的入度。如此反复，直到所有节点都跑完。

DAG 调度中的常见问题与避坑

DAG 听起来很美好，但落地时坑不少。我踩过的几个：

• 循环依赖检测——用户配置依赖时手一抖，A 依赖 B、B 依赖 C、C 又依赖 A。调度器一启动就死锁。我建议在提交 DAG 时做拓扑排序，如果排序失败（节点数少于总数），直接拒绝提交。
• 节点失败重试策略——日志任务经常因为网络抖动或资源不足失败。我曾经遇到一个场景：清洗节点重试了 5 次都失败，下游的聚合节点一直等，导致整个管道阻塞。后来我加了「失败阈值」和「跳过策略」——如果某个节点重试超过 3 次，就标记为失败，下游节点可以选择「跳过」或者「使用默认值」继续执行。
• 数据一致性——DAG 中某个节点失败后，已经执行成功的节点数据怎么处理？我个人的做法是：每个节点都设计成「幂等」的，失败重跑不会产生重复数据。日志数据天然带时间戳，配合去重机制，基本能保证一致性。

DAG 在日志平台中的落地实践

我参与过的一个日志平台，每天处理 50TB 以上的日志数据。DAG 调度器是我们自己写的，核心逻辑大概 300 行代码。给你看看简化版：

// 伪代码：DAG 调度器核心逻辑
class DagScheduler:
    def __init__(self, dag):
        self.graph = dag.nodes          // 节点列表
        self.edges = dag.edges          // 边列表
        self.in_degree = {}             // 入度表
        self.ready_queue = []           // 就绪队列

    def build_in_degree(self):
        for node in self.graph:
            self.in_degree[node] = 0
        for (from_node, to_node) in self.edges:
            self.in_degree[to_node] += 1

    def schedule(self):
        self.build_in_degree()
        // 初始化就绪队列
        for node, degree in self.in_degree.items():
            if degree == 0:
                self.ready_queue.append(node)

        while self.ready_queue:
            node = self.ready_queue.pop(0)
            execute(node)               // 执行任务
            // 更新下游入度
            for (from_n, to_n) in self.edges:
                if from_n == node:
                    self.in_degree[to_n] -= 1
                    if self.in_degree[to_n] == 0:
                        self.ready_queue.append(to_n)

        // 检查是否所有节点都执行了
        if any(degree > 0 for degree in self.in_degree.values()):
            raise Exception("存在循环依赖或未执行节点")

这段代码虽然简单，但核心思想都在里面了。实际生产环境还要加并发控制、失败重试、超时管理、监控告警。不过骨架就是这个。

DAG 与日志任务的生命周期管理

日志任务不是跑一次就完事的。每天都有新的日志产生，DAG 需要支持「周期调度」和「增量处理」。我习惯把 DAG 分成两类：

• 全量 DAG——用于初始化或数据修复，处理全部历史数据
• 增量 DAG——按时间窗口（比如每 5 分钟）处理新增数据

增量 DAG 的每个节点都要感知时间窗口。比如「解析节点」只解析最近 5 分钟的日志，「聚合节点」只聚合这个时间窗口内的数据。这样做的好处是：某个时间窗口失败了，只需要重跑那个窗口的 DAG，不影响其他窗口。

我记得有一次凌晨 3 点，某个数据源突然断连了 10 分钟。恢复后，我只需要重跑那 10 分钟的增量 DAG，而不是全量重跑。如果当时用的是全量 DAG，50TB 的数据重跑一遍，想想都头疼。

总结一下

DAG 在日志任务编排中的价值，说白了就是三句话：

• 依赖关系清晰可见，不会出现「改一处崩一片」
• 并行执行最大化，能同时跑的任务绝不排队
• 容错粒度精细，一个节点失败不影响全局

嗯，如果你现在还在用脚本硬编码任务依赖，我建议你试试 DAG。从一个小场景开始，比如把采集、解析、清洗三个任务用 DAG 串起来。跑通了之后，你会发现日志管道的稳定性会上一个台阶。