3、DAG定义与依赖：如何定义有向无环图、任务依赖关系、上下游传递

调度系统最核心的抽象，说白了就是一张图。这张图叫 DAG（Directed Acyclic Graph），有向无环图。你想想看，一个数据流水线里，任务 A 跑完才能跑 B，B 跑完才能跑 C，这不就是一条链吗？但现实比这复杂得多——可能有分支、有合并、有等待。DAG 就是用来描述这种复杂关系的标准工具。

我刚开始接触调度系统时，觉得 DAG 不就是画几条线嘛。后来被坑了几次才明白：定义 DAG 的细节，直接决定了你的流水线是稳如老狗还是天天报警。

3.1 什么是 DAG？为什么必须是无环？

DAG 有两个关键词：有向和无环。

  • 有向：每条边都有方向，表示依赖关系。A → B 表示 A 是 B 的前置任务。
  • 无环：不能出现循环依赖。比如 A 依赖 B，B 依赖 C，C 又依赖 A——这就死锁了。

为什么不能有环？我举个例子。假设你有一个数据清洗任务，它依赖上一个批次的清洗结果。如果定义成环，调度系统会永远等下去，因为上一个批次还没跑完，而它又在等当前批次。嗯，这就是典型的「鸡生蛋蛋生鸡」问题。

核心原则：DAG 中的每个任务只能依赖它前面的任务，不能反向依赖。调度系统在启动时会做拓扑排序，如果发现环，直接报错。

下面这张图展示了一个典型的 ETL 流水线 DAG：

典型 ETL 流水线 DAG 示例 数据抽取 数据清洗 数据转换 质量校验 失败告警 数据加载 说明： 蓝色节点为正常处理流程，绿色节点为质量校验，红色节点为异常处理。 箭头方向表示数据流向，下游任务必须等待上游任务完成后才能启动。

3.2 任务依赖关系的三种定义方式

在实际工程中，定义依赖关系有几种常见方式。我根据项目经验，把它们整理成了一张对比表：

定义方式 适用场景 优点 缺点
显式依赖 任务数量少，依赖关系固定 清晰直观，容易调试 维护成本高，改一处要改多处
隐式依赖 任务数量多，依赖关系有规律 配置量小，扩展性好 排查问题困难，容易遗漏
动态依赖 依赖关系在运行时才能确定 灵活，能应对复杂场景 实现复杂，调试难度大

显式依赖是最直接的方式。比如在 Airflow 里，你直接写 task_a >> task_b。我在项目中遇到过一个小团队，所有依赖都用这种方式写，结果 DAG 有 200 多个任务，每次改依赖关系都要翻半天代码。后来我建议他们改用命名规范来做隐式依赖。

隐式依赖靠任务命名或标签来推导。比如所有以 extract_ 开头的任务自动排在 transform_ 前面。这种方式配置量小，但有个坑：如果命名不规范，依赖关系会乱掉。我曾经见过一个案例，有人把 transform_user 写成了 transfrom_user，结果调度系统没识别出来，任务直接并行跑了，数据全乱套。

动态依赖最灵活，也最危险。比如某个任务需要根据上游产出的数据量决定下游要启动多少个并行任务。我建议只在确实需要运行时才能确定依赖的场景下使用，比如动态分区加载。

3.3 上下游传递：数据与状态的接力棒

定义好依赖关系只是第一步。上下游之间怎么传递数据？怎么传递状态？这才是真正的工程难点。

我个人习惯把传递方式分为三种：

  1. 文件传递：上游写文件，下游读文件。适合大数据量场景，比如 HDFS 上的 Parquet 文件。
  2. 消息传递：上游发消息，下游消费消息。适合实时或准实时场景，比如 Kafka。
  3. 数据库传递：上游写数据库，下游读数据库。适合小数据量或需要事务保证的场景。

我的经验：选择传递方式时，先问自己三个问题：数据量多大？实时性要求多高？失败后能不能重跑？这三个问题能帮你筛掉 80% 的错误选择。

状态传递容易被忽视。比如上游任务跑完了，但产出的数据有问题。下游如果只看「任务完成」状态就启动，那就会把脏数据一路传播下去。我建议在上下游之间加一个数据质量检查点，只有数据通过校验，下游才能开始跑。

3.4 避坑指南：我踩过的那些坑

坑一：循环依赖检测不到位

我曾经在做一个实时数仓项目时，因为 DAG 定义文件里有一个隐藏的循环依赖，导致调度系统在凌晨 3 点卡死了。排查了 2 个小时才发现，是 A 依赖 B，B 依赖 C，C 又依赖 A——但 C 的依赖是在配置文件的另一行写的，肉眼根本看不出来。

解决方案：在 CI/CD 流程中加入 DAG 合法性检查，用拓扑排序算法提前检测环。别等到运行时才发现。

坑二：上下游版本不一致

有一次上游任务升级了输出格式，从 JSON 改成了 Avro，但下游任务没同步更新。结果下游解析数据时全部报错，整个流水线停了 40 分钟。嗯，这就是典型的「接口契约」问题。

解决方案：上下游之间定义明确的接口规范，用 Schema Registry 管理数据格式。每次变更都要走审批流程。

坑三：依赖粒度太粗

我见过有人把整个数仓的 ETL 写成一个 DAG，所有任务都依赖一个「全量抽取」任务。结果每次全量抽取跑 3 小时，后面的任务全在等。其实很多任务只需要依赖增量数据。

解决方案：按业务域拆分 DAG，每个 DAG 控制在 20-50 个任务。依赖粒度要细，能依赖表级别就不要依赖库级别。

3.5 实战：用代码定义 DAG

说了这么多理论，我们来看一段实际代码。下面是用 Python 定义的一个简单 DAG：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'etl_pipeline',
    default_args=default_args,
    description='ETL流水线示例',
    schedule_interval='0 2 * * *',  # 每天凌晨2点执行
    catchup=False,
)

def extract():
    print("抽取数据...")

def transform():
    print("转换数据...")

def load():
    print("加载数据...")

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform,
    dag=dag,
)

load_task = PythonOperator(
    task_id='load',
    python_callable=load,
    dag=dag,
)

# 定义依赖关系
extract_task >> transform_task >> load_task

这段代码定义了一个最简单的三条链。注意最后一行 extract_task >> transform_task >> load_task，这就是显式依赖的写法。如果你有多个下游，可以用 >> 链式调用，也可以用 [] 列表。

关键点：depends_on_past 这个参数要小心。如果设为 True，当前任务会依赖上一次调度中同一个任务的执行结果。这在某些场景下有用，但也会导致「一次失败，永远失败」的连锁反应。我一般只在数据有严格顺序要求时才开启。

好了，DAG 的定义和依赖关系就讲到这里。记住三个核心：无环是底线，依赖要清晰，传递要可靠。下次你设计流水线时，先画一张 DAG 图，把依赖关系理清楚，再动手写代码。这样能省掉后面 80% 的调试时间。

← 上一章 📖 返回目录 下一章 →