Python 数据管线调度引擎：DAG 编排代替 Cron 硬编码-拓冰建站

Python 数据管线调度引擎：DAG 编排代替 Cron 硬编码

一、Cron 长成了无人能看懂的怪物

数据团队最怕的事情之一：凌晨 Cron 任务失败了。
一个数据仓库有 40 个定时任务，依赖关系像蜘蛛网。
任务 A 依赖任务 B 和 C 的输出。
B 又依赖 D 和 E。
这个依赖链写在 Shell 脚本里，靠 sleep 估算等待时间。

真实事故：任务 B 因上游数据延迟跑了 3 小时。
任务 A 的 sleep 只有 2 小时。
A 提前启动，读到了不完整的数据。
最终报表数据偏差 15%。
排查花了一天，修复只用了五分钟。

Cron 的原罪是两样。
第一，没有原生依赖支持。
第二，没有失败传播机制。
上游任务失败，下游应该自动跳过，但 Cron 不知道。

二、DAG 调度引擎的核心抽象

DAG（有向无环图）是描述任务依赖的自然模型。
每个节点是一个任务，每条边表示一个依赖。
调度引擎的核心职责是：按拓扑顺序执行，同时最大化并行度。

flowchart TB A[数据源抽取] --> B[数据清洗] A --> C[维度表同步] B --> D[事实表计算] C --> D D --> E[聚合报表] D --> F[数据质量检查] E --> G[邮件推送] F -->|通过| G F -->|失败| H[告警 + 阻断下游]

调度引擎的关键特性：
拓扑排序决定执行顺序；
同一层级的无依赖节点并行执行；
上游失败自动阻断下游；
支持超时和重试策略。

三、Python 实现轻量 DAG 引擎

""" dag_engine.py - 轻量 DAG 任务调度引擎 """ import asyncio import logging from datetime import datetime from typing import Callable, Dict, List, Optional, Set logger = logging.getLogger(__name__) class TaskNode: """DAG 中的一个任务节点""" def __init__( self, name: str, func: Callable, timeout: int = 3600, retries: int = 1, ): self.name = name self.func = func self.timeout = timeout self.retries = retries self.dependencies: Set[str] = set() self.status = "pending" # pending|running|success|failed|skipped self.start_time: Optional[datetime] = None self.end_time: Optional[datetime] = None def depends_on(self, *task_names: str): """声明依赖任务""" self.dependencies.update(task_names) class DAGEngine: """DAG 任务调度引擎""" def __init__(self): self.tasks: Dict[str, TaskNode] = {} def add_task(self, task: TaskNode): """注册任务""" if task.name in self.tasks: raise ValueError(f"任务 {task.name} 已存在") self.tasks[task.name] = task def validate(self) -> bool: """校验 DAG 合法性：无循环依赖，所有依赖存在""" task_names = set(self.tasks.keys()) for task in self.tasks.values(): for dep in task.dependencies: if dep not in task_names: raise ValueError(f"任务 {task.name} 依赖了不存在的任务 {dep}") # 检测循环依赖（DFS） def has_cycle(name: str, visited: Set[str], path: Set[str]) -> bool: visited.add(name) path.add(name) for dep in self.tasks[name].dependencies: if dep in path: return True if dep not in visited: if has_cycle(dep, visited, path): return True path.discard(name) return False for name in task_names: if has_cycle(name, set(), set()): raise ValueError(f"检测到循环依赖，包含任务 {name}") return True async def run(self) -> Dict[str, str]: """执行所有任务，按拓扑序并行""" self.validate() completed: Set[str] = set() results: Dict[str, str] = {} while len(completed) < len(self.tasks): ready = [] for name, task in self.tasks.items(): if name in completed: continue if task.dependencies.issubset(completed): ready.append(task) if not ready: # 没有就绪任务，但仍有未完成任务，说明有阻塞 blocked = [ name for name in self.tasks if name not in completed ] for name in blocked: self.tasks[name].status = "skipped" results[name] = "skipped" completed.add(name) logger.warning(f"任务 {name} 因上游失败被跳过") break # 并行执行就绪任务 coros = [self._execute_task(task) for task in ready] task_results = await asyncio.gather(*coros, return_exceptions=True) for task, result in zip(ready, task_results): if isinstance(result, Exception): task.status = "failed" results[task.name] = str(result) logger.error(f"任务 {task.name} 执行异常: {result}") else: task.status = result results[task.name] = result completed.add(task.name) return results async def _execute_task(self, task: TaskNode) -> str: """执行单个任务，支持超时和重试""" task.status = "running" task.start_time = datetime.now() logger.info(f"[{task.name}] 开始执行") for attempt in range(task.retries + 1): try: # 支持异步和同步函数 if asyncio.iscoroutinefunction(task.func): result = await asyncio.wait_for( task.func(), timeout=task.timeout ) else: loop = asyncio.get_running_loop() result = await asyncio.wait_for( loop.run_in_executor(None, task.func), timeout=task.timeout, ) task.end_time = datetime.now() duration = (task.end_time - task.start_time).total_seconds() logger.info( f"[{task.name}] 完成, 耗时 {duration:.1f}s" ) return "success" except asyncio.TimeoutError: logger.warning( f"[{task.name}] 超时({task.timeout}s), 第 {attempt+1} 次重试" ) if attempt == task.retries: task.end_time = datetime.now() return "failed" await asyncio.sleep(min(2 ** attempt, 30)) except Exception as e: logger.error( f"[{task.name}] 异常: {e}, 第 {attempt+1} 次重试" ) if attempt == task.retries: task.end_time = datetime.now() raise # 由调用方处理 return "failed" # ---- 使用示例 ---- async def main(): engine = DAGEngine() # 定义任务 extract = TaskNode("extract", lambda: __import__("time").sleep(2)) clean_a = TaskNode("clean_a", lambda: print("清洗 A 数据")) clean_b = TaskNode("clean_b", lambda: print("清洗 B 数据")) aggregate = TaskNode("aggregate", lambda: print("聚合计算")) report = TaskNode("report", lambda: print("生成报表")) # 声明依赖关系 clean_a.depends_on("extract") clean_b.depends_on("extract") aggregate.depends_on("clean_a", "clean_b") report.depends_on("aggregate") # 注册并运行 for t in [extract, clean_a, clean_b, aggregate, report]: engine.add_task(t) results = await engine.run() print("执行结果:", results) if __name__ == "__main__": asyncio.run(main())