数据治理、质量与血缘：从 dbt 测试到 OpenLineage

0. 元信息

• 主题路径：docs/topics/data-engineering-basics/subtopics/data-quality-and-lineage/README.md
• 父主题：data-engineering-basics
• 适合对象：会写 SQL/Python、跑过数据流水线、希望从”数据能用”过渡到”数据可信、可追溯”的工程师
• 建议周期：2～2.5 周，每周 10～14 小时
• 前置知识：SQL、Python、Airflow；建议先完成父主题第 4 阶段
• 最终目标：能为给定数仓设计一套 dbt + Great Expectations + OpenLineage 的数据治理方案，并用 SQL/UI 查询血缘、配置质量门禁、定义数据契约

1. 学习路线

数据契约 → dbt 转换与测试 → Great Expectations 期望 → Soda 分布异常 → OpenLineage 血缘 → Marquez / DataHub 目录 → 可观测

2. 阶段周数分配

每天 1.5～2 小时。2 周方案专注前 5 阶段；2.5 周方案多 2 天做血缘与治理落地。

3. 核心知识 / 产出 / 标准表

4. 第一周（每天 1.5~2 小时）

Day 1 约定：本计划使用 Docker Compose 起 PostgreSQL + dbt + Great Expectations + Marquez。dbt 跑 dbt build / dbt test；GE 跑 great_expectations checkpoint run；血缘事件用 openlineage-client 发到 Marquez HTTP endpoint。所有契约存为 YAML / Protobuf，版本化在 Git。

Day 7 执行次序

步骤 A —— 最小可用治理链路（60~90 分钟）

1. dbt build 跑通（run + test）；
2. GE checkpoint 跑通（5 passed）；
3. 写 Python 脚本发 OpenLineage 事件到 Marquez；
4. Marquez UI 看到血缘；
5. 故意改源 schema 制造契约漂移，看告警。

步骤 B —— 4 类边界用例（30~45 分钟）

Day 7 当天必完成步骤 A；步骤 B 至少完成 B1、B2。

5. 阶段通用验收

1. 不看答案独立重写 dbt 3 层 + 5 类测试 + GE Checkpoint + OpenLineage 事件流；
2. 用自己的话解释”为什么需要数据契约""为什么测试要阻塞下游""为什么血缘要列级而非仅表级”；
3. 画一张图：契约 + 转换 + 测试 + 血缘全链路图（三件套之一）；
4. 测试契约漂移、测试失败、分布异常、血缘事件丢失、SLA miss 5 类边界；
5. 准备至少 3 组自定义数据集并贴出实际测试结果与血缘截图；
6. 记录 dbt build 时长、测试通过率、GE Checkpoint 失败率、OpenLineage 事件成功率 4 个核心指标；
7. 能修改已有契约（加列 / 改 SLA）并验证血缘自动更新。

交付存放：第 3 项的图、第 5 项的输出、第 6 项的指标，统一存到 week1/notes/ 或 week1/<dataset>/README.md。

6. 最终验收

• 独立设计并交付一套数据治理方案（dbt + GE/Soda + OpenLineage + Marquez），覆盖契约 / 测试 / 血缘 / SLA 4 大件；

• 至少完成 18 个实战项（分布建议）：

  子阶段	题目数量	难度	平台建议
  数据契约	2	Easy / Medium	Confluent SR / Protobuf
  dbt 转换与测试	4	Medium	dbt 官方 Tutorial
  Great Expectations	4	Medium	GE 官方 Examples
  Soda	2	Medium	Soda 官方 Examples
  OpenLineage + 目录	4	Medium / Hard	OpenLineage Spec + Marquez
  治理落地	2	Hard	Monte Carlo / Datafold 博客

  约束：至少 9 项达到 Medium，至少 3 项达到 Hard；每项必须留可运行 dbt project + 测试报告 + 血缘截图。

• 完成 1 个综合项目：端到端治理（dbt + GE + OpenLineage + Marquez + Slack 告警 + 1 次故障复盘）；

• 能用 15 分钟讲清数据契约四要素、5 类 dbt 测试、GE vs Soda 选型、OpenLineage 事件结构、Marquez 查血缘。

7. 综合项目

首选：端到端数据治理（必做：数据契约 + dbt + GE + OpenLineage + Marquez + Slack 告警）。
备选：分布异常监测（Soda + 历史基线 + 自动告警）。
备选：数据目录 + 治理门户（DataHub + dbt 集成 + 标签 + Glossary + 影响分析）。

端到端治理必做要求：

• 输入：订单 / 用户 / 商品 3 张源表（PostgreSQL）+ 至少 1 份 Protobuf 契约；
• 输出：必输出（1）dbt 3 层 models（staging / intermediate / marts）+ 5 类测试；（2）GE Checkpoint + Data Docs 站点；（3）OpenLineage 事件流到 Marquez；（4）Marquez UI 截图（Job / Dataset 血缘）；（5）Slack 告警（dbt 测试失败 + GE Checkpoint 失败 + SLA miss）；（6）1 次故障复盘（模拟契约漂移）；
• 算法 / 工程：用 dbt build 触发 OpenLineage 自动事件；用 dbt source freshness 配 SLA；用 ge_checkpoint 失败阻塞下游；
• 进阶可选：用 Soda 做分布异常检测；接 Monte Carlo / Datafold 做可观测；用 Avatica / Trino 做数据查询入口。

任何综合项目都必须包含：

1. 需求说明与数据契约（4 要素齐：schema + owner + SLA + version）；
2. 全链路架构图与命名规范；
3. 核心代码（dbt models + GE Checkpoint + OpenLineage 事件 + Slack 告警 callback）；
4. 边界测试（契约漂移、测试失败、分布异常、血缘事件丢失、SLA miss）；
5. 可观测（GE Data Docs + Marquez UI + Slack 告警日志 + Grafana metrics）；
6. README（设计取舍、测试选型理由、下一步）；
7. 复盘记录（模拟 1 次 P0 故障，写 retrospective.md）；
8. notes/ 规范：

本主题贡献

数据治理的核心矛盾是”数据团队生产量大 vs 业务方不敢信”。本子主题专门讲清 Great Expectations + OpenLineage + Marquez 如何把这条信任墙拆掉——以 Great Expectations 为”期望即代码”的事实标准，OpenLineage 为血缘事件开放规范，Marquez 为血缘目录事实实现，SLA 为治理硬指标；不重复父主题的 dbt 基础与 Iceberg 存储。

3 职责

1. 用 Great Expectations Checkpoint 把 10 条 Expectations（ExpectColumnValuesToNotBeNull / ExpectColumnValuesToBeUnique / ExpectColumnValuesToBeBetween / ExpectColumnMeanToBeBetween / ExpectColumnStdevToBeBetween）做成可阻塞的质量门禁，失败阻塞下游 dbt run / Airflow Task。
2. 用 OpenLineage 事件（Run Start / Complete / Fail，含 run.facets.errorMessage / nominalTime / dataset.facets.schema）把每个 Task 的输入输出数据集上报到 Marquez，血缘由”人维护”变成”自动产生”。
3. 用 SLA 三件套（freshness / completeness / validity）为关键数据集定硬指标——dbt source freshness + GE ExpectColumnValuesToNotBeNull + Marquez nominalTime 监控，SLA miss 接 Slack / PagerDuty 告警。

4 交付物

1. 一份 Great Expectations 项目（10 条 Expectations + Checkpoint + Data Docs 站点），含 5 种分布异常检测（均值 / 标准差 / 分位数 / 唯一值比例 / NULL 比例），Checkpoint 失败 exit code 非 0。
2. 一份 OpenLineage 事件脚本（Python openlineage-client SDK + dbt-ol 集成 + Airflow OpenLineage provider），每个 Task 自动发 Run Start / Complete，含 Job / Dataset / Run Facet。
3. 一份 Marquez 部署（docker compose up marquez）+ 血缘查询示例（marquez-cli lineage get <dataset>），含 Job → Dataset 血缘截图 + 列级血缘验证。
4. 一份 SLA 监控配置（dbt source freshness + GE Checkpoint + Marquez nominalTime 监控），含 Slack / PagerDuty 告警 callback，SLA 文档写明 4 要素：freshness / completeness / validity / owner。

3 指标

1. Great Expectations Checkpoint 通过率 ≥ 99%（失败 Checkpoint 阻塞下游，强制修复后重跑）。
2. OpenLineage 事件成功率 ≥ 99.5%（Task 完成时事件必达 Marquez，缺失即告警）。
3. SLA miss 告警响应 < 5 min（freshness 超时 → Slack 通知 → on-call 介入）。

8. 推荐开源资料

许可证提示：dbt Core / GE / Soda / OpenLineage / Marquez / DataHub 都是 Apache-2.0；Unity Catalog 是商业版。复制 Apache 项目示例前请保留 LICENSE 与 NOTICE。默认做法是读规范后自己写测试与契约，而不是复制官方 Example。

默认使用顺序：先读 Reis《Fundamentals of Data Engineering》第 8~10 章建立治理心智 → 装 dbt + GE → 写数据契约（YAML + Protobuf）→ 写 dbt 3 层 models → 配 5 类 dbt 测试 → 加 GE Checkpoint + Data Docs → 起 Marquez + OpenLineage → 用 Slack 告警 → 跑端到端治理演练 → 写 notes/retrospective.md。

9. 学习资料汇聚（v0.3 自包含）

本节由本计划生成。链接指向原始材料或作者公开内容。规范会演进，记录时务必写明版本与日期。

9.1 背景与动机

数据团队的核心痛点不是”没数据”，而是”数据不可信”。脏数据进仓 → 错报表 → 错决策 → 业务损失。传统做法靠人工抽样检查，扩展性差。dbt 2016 年开源把”转换即代码 + 测试即代码”带入主流；Great Expectations 2018 年、Soda 2020 年先后开源，把”期望/检查即代码”做成事实标准。OpenLineage 2021 年由 Linux Foundation 发起，把血缘（lineage）从厂商特性做成开放标准，与 Marquez / DataHub / Unity Catalog 互操作。今天数据团队的”治理成熟度”由血缘完整度、契约覆盖率、测试通过率三项指标衡量。

9.2 概念地图

flowchart LR
  Src[源系统] --> Contract[数据契约]
  Contract --> Ingest[摄取]
  Ingest --> Lake[Lakehouse]
  Lake --> dbt[dbt 转换]
  dbt --> Test[dbt 测试]
  Lake --> GE[GE 期望]
  dbt --> Soda[Soda 检查]
  Test --> Gate[质量门禁]
  GE --> Gate
  Soda --> Gate
  Gate --> Serve[服务 BI/API/ML]
  dbt --> OL[OpenLineage 事件]
  Ingest --> OL
  Lake --> OL
  OL --> MQ[Marquez]
  OL --> DH[DataHub]
  Serve --> Obs[可观测]

9.3 基础知识讲解

9.3.1 经典论文 / 规范

9.3.2 经典书籍

9.3.3 优秀博客与文档

9.3.4 核心人物

9.3.5 开发方法

9.4 经典问题与经典案例（≥5 道）

9.5 学习难点

概念难点

思维难点

工程难点

9.6 技术标准与接口

Entity

Scope

dbt 是转换 + 测试；GE / Soda 是质量检查；OpenLineage 是血缘标准；Marquez / DataHub / Atlas 是目录实现。它们互补，不替代。

Structure

• dbt：sources → staging → intermediate → marts，测试写在 schema.yml。
• GE：ExpectationSuite → Checkpoint → Data Docs。
• Soda：SodaCL 检查定义 → Soda Core 扫描 → 告警。
• OpenLineage：run / job / dataset / facet，事件推送到 HTTP 或 Kafka。
• 数据契约：schema + owner + SLA + version，可入 Schema Registry。

Ecosystem

• 转换：dbt、SQLMesh、Coalesce。
• 质量：GE、Soda、Monte Carlo、Datafold、Bigeye。
• 血缘：OpenLineage、Marquez、DataHub、Apache Atlas、Unity Catalog。
• 契约：Schema Registry (Confluent / Apicurio)、Protobuf、Avro。

Depth Tiers

本子主题目标：L3。

Source

• dbt 官方文档：引用快照 2026-07-30。
• Great Expectations 官方文档：引用快照 2026-07-30。
• OpenLineage 官方文档：引用快照 2026-07-30。

9.7 关键代码

9.7.1 dbt model + schema.yml（5 类测试）

# 9.7.1 dbt project/models/staging/schema.yml
version: 2

models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: user_id
        tests:
          - not_null
          - relationships:
              to: ref('stg_users')
              field: user_id
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'paid', 'shipped', 'cancelled']
      - name: order_date
        tests:
          - not_null
      - name: amount
        tests:
          - not_null
    config:
      contract:
        enforced: true

-- 9.7.1.2 dbt incremental model
{{ config(materialized='incremental', unique_key='order_id') }}

SELECT
    order_id,
    user_id,
    amount,
    order_date,
    status
FROM {{ source('raw', 'orders') }}

{% if is_incremental() %}
WHERE order_date > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}

9.7.2 Great Expectations Checkpoint

# 9.7.2 Great Expectations 配置 + Checkpoint
import great_expectations as gx

context = gx.get_context(mode="file", project_root_dir="./gx")

batch = context.get_batch(
    batch_definition=context.data.add_pandas_filesystem(
        name="orders", base_directory="./data"
    ),
    batch_parameters={"path": "./data/orders.parquet"},
)

suite = context.suites.add(
    gx.ExpectationSuite(name="orders_suite")
)
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="amount", min_value=0, max_value=1_000_000
))
suite.add_expectation(gx.expectations.ExpectColumnMeanToBeBetween(
    column="amount", min_value=10, max_value=10_000
))

checkpoint = context.checkpoints.add(
    gx.Checkpoint(name="orders_checkpoint", suites=[suite], actions=[
        gx.checkpoint.UpdateDataDocsAction(name="docs"),
    ])
)
checkpoint.run(batch=batch)

9.7.3 OpenLineage + Airflow

# 9.7.3 Airflow + OpenLineage Provider 配置
from airflow import DAG
from airflow.providers.openlineage.sensors.openlineage import OpenLineageSensor
from airflow.operators.bash import BashOperator

dag = DAG("orders_etl", openlineage_integration=True)

extract = BashOperator(
    task_id="extract",
    bash_command="echo extract",
    dag=dag,
    inlets=[{"type": "database", "name": "raw.orders"}],
    outlets=[{"type": "table", "name": "dw.ods_orders"}],
)

10. 常见误区

• 把数据质量当监控看，不挂门禁；
• dbt 测试只写 unique / not_null；
• 契约只在文档里写，不入 Schema Registry；
• GE 期望只写语法不写分布；
• OpenLineage 事件只发 Task 起点不终点；
• 血缘只画表级不看列级；
• SLA 写了不告警；
• 数据目录没人维护；
• dbt 项目没有 staging/intermediate/marts 分层；
• 增量模型缺 unique_key 导致重复；
• 数据契约没 owner，事故找不到人；
• 质量事故不复盘。

11. 所有知识点分类

1. 编程语言
2. 数据结构与算法
3. 计算机基础
4. 工程技术
5. Web 与后端
6. 前端与客户端
7. 数据与人工智能
8. 项目与职业能力
9. 安全与可靠性

本计划归属：数据与人工智能 主 + 工程技术 辅。