📚 课程目录第4章数据质量监控

4、数据质量监控：Great Expectations、Deequ、数据血缘追踪

数据质量这事，说起来简单做起来难。我见过太多团队，模型上线前才发现训练数据里有 30% 的空值，或者特征分布跟生产环境完全对不上。说白了，数据质量监控就是给数据管道装个「安检门」——数据进来之前，先过一遍检查。

这一章我们聊三个工具：Great Expectations、Deequ，以及数据血缘追踪。它们解决的是不同层面的问题，但目标一致：让数据可信。

4.1 为什么需要数据质量监控？

先看一个真实场景。我曾经接手过一个推荐系统项目，离线训练 AUC 0.82，上线后直接掉到 0.65。查了三天，发现是上游特征表某天开始字段类型从 int 变成了 string，模型推理时全部解析失败。

这种问题，如果没有自动化监控，靠人肉巡检根本来不及。数据质量监控的核心价值就三点：

早发现：数据异常在源头就被捕获，而不是等到模型效果崩了才后知后觉
可追溯：知道哪批数据、哪个环节出了问题，能快速定位责任人
可复现：异常数据被标记、隔离，不影响下游训练和推理

核心原则：数据质量监控不是「一次性检查」，而是持续集成的一部分。每次数据管道运行，都应该自动触发质量校验。

4.2 Great Expectations：声明式数据质量框架

Great Expectations（简称 GX）是我个人最常用的数据质量工具。它的设计理念很优雅——用 Expectation 来描述数据应该长什么样，然后自动验证。

4.2.1 核心概念

概念	说明	类比
Expectation	对数据列的断言，比如「age 列不能有负数」	类似单元测试中的 assert
Expectation Suite	一组 Expectation 的集合，对应一张表或一个数据集	测试用例集
Data Docs	自动生成的可视化报告，展示校验结果	测试报告
Checkpoint	执行校验的配置单元，可以绑定数据源和 Suite	CI 中的测试任务

4.2.2 实战：定义并运行 Expectation

嗯，这里要注意：GX 支持多种数据源，包括 Pandas、Spark、SQL 数据库。下面是一个典型的校验流程。

# 安装：pip install great_expectations

import great_expectations as gx

# 1. 创建 Data Context
context = gx.get_context()

# 2. 连接数据源（以 Pandas DataFrame 为例）
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset("my_asset")

# 3. 定义 Expectation Suite
suite = context.add_expectation_suite("user_profile_suite")

# 4. 添加校验规则
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="age", min_value=0, max_value=120
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(
        column="user_id"
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnDistinctValuesToBeInSet(
        column="gender", value_set=["M", "F", "Other"]
    )
)

# 5. 创建 Checkpoint 并运行
checkpoint = context.add_checkpoint(
    name="user_profile_check",
    datasource_name="my_datasource",
    data_asset_name="my_asset",
    expectation_suite_name="user_profile_suite",
)

# 6. 执行校验
results = checkpoint.run(batch_parameters={"dataframe": df})
print(results["success"])  # True 表示全部通过

我的习惯：每个 Expectation Suite 都配上 Data Docs 自动生成。这样非技术人员也能通过网页看到数据质量状态，不用每次都来找我问「数据有没有问题」。

4.2.3 避坑指南

我曾经犯过一个错误：把 Expectation 写得过于严格。比如要求某个字段「必须 100% 非空」，结果因为上游一次合法的小批量数据变更，整个管道都阻塞了。

不要用阈值：用 mostly=0.95 允许 5% 的异常，比直接卡死更实用
区分硬约束和软约束：主键唯一性这种必须硬校验，分布偏移可以设告警阈值
定期更新 Suite：数据模式会变，Expectation 也要跟着迭代

4.3 Deequ：Spark 生态下的数据质量库

如果你的数据管道跑在 Spark 上，Deequ 是更好的选择。它是 AWS 开源的库，专门针对大数据场景设计。我一般在处理 TB 级数据时才会用它——GX 在单机场景更方便，但面对大规模分布式数据，Deequ 的性能优势很明显。

4.3.1 核心特性

特性	说明	适用场景
Constraint	类似 GX 的 Expectation，定义数据约束	字段级校验
Anomaly Detection	基于历史数据自动检测异常	分布漂移、趋势突变
Profiling	自动生成数据集的统计画像	探索性分析、基线建立
Metrics Repository	存储历史校验结果，支持趋势分析	长期质量监控

4.3.2 实战：用 Deequ 校验 DataFrame

// Maven 依赖：com.amazon.deequ:deequ:2.0.3-spark-3.3

import com.amazon.deequ.{VerificationSuite, VerificationResult}
import com.amazon.deequ.checks.{Check, CheckLevel}
import com.amazon.deequ.constraints.ConstraintStatus

// 1. 定义校验规则
val verificationResult = VerificationSuite()
  .onData(df)
  .addCheck(
    Check(CheckLevel.Error, "用户表基础校验")
      .isComplete("user_id")          // 非空
      .isUnique("user_id")            // 唯一
      .isContainedIn("gender", Array("M", "F", "Other"))
      .hasMin("age", _ >= 0)          // 最小值约束
      .hasMax("age", _ <= 120)        // 最大值约束
  )
  .addCheck(
    Check(CheckLevel.Warning, "分布告警")
      .hasSize(_ >= 1000000)          // 数据量不少于 100 万行
      .hasApproxQuantile("age", 0.5, _ >= 18)  // 中位数年龄 >= 18
  )
  .run()

// 2. 检查结果
if (verificationResult.status == VerificationResult.Success) {
  println("所有校验通过")
} else {
  verificationResult.checkResults.foreach { case (check, constraintResults) =>
    constraintResults.filter(_.status != ConstraintStatus.Success)
      .foreach { cr =>
        println(s"失败约束: ${cr.constraint} - ${cr.message.getOrElse("")}")
      }
  }
}

注意：Deequ 的 Anomaly Detection 需要 Metrics Repository 支持。如果你没有配置存储后端（比如 S3 或 HDFS），每次运行都是独立校验，无法做趋势分析。我建议至少保留 30 天的历史指标。

4.4 数据血缘追踪：让数据流动可见

数据质量监控只能告诉你「数据有问题」，但数据血缘追踪能告诉你「问题从哪来、影响到哪去」。说白了，血缘就是数据的「族谱」——记录每张表、每个字段是怎么来的，又被谁用了。

4.4.1 为什么需要血缘追踪？

我记得有一次线上事故：某个特征表因为上游 ETL 脚本变更，导致字段含义变了。下游十几个模型全部受影响，但排查花了两天——因为没人知道哪些模型依赖了这个字段。

如果有血缘追踪，你只需要查一下这个字段的「下游依赖」，所有受影响的任务一目了然。

4.4.2 常见血缘追踪工具

工具	类型	特点
Apache Atlas	企业级元数据管理	支持 Hive、HBase、Kafka 等，集成度高
DataHub	开源数据目录	UI 友好，支持实时血缘
Amundsen	数据发现平台	Lyft 开源，搜索能力强
Marquez	轻量级血缘引擎	WeWork 开源，适合与 Airflow 集成

4.4.3 实战：用 Marquez 追踪血缘

我个人比较喜欢 Marquez，因为它轻量、跟 Airflow 配合得很好。下面是一个简单的集成示例。

# 安装：pip install marquez-python

from marquez_client import MarquezClient

client = MarquezClient(
    url="http://localhost:5000",
    ="your-"
)

# 1. 定义数据源
client.create_source(
    source_name="hive_warehouse",
    source_type="HIVE",
    connection_url="jdbc:hive2://warehouse:10000"
)

# 2. 定义数据集
client.create_dataset(
    dataset_name="user_profile",
    dataset_type="TABLE",
    physical_name="ods.user_profile",
    source_name="hive_warehouse",
    fields=[
        {"name": "user_id", "type": "string", "description": "用户ID"},
        {"name": "age", "type": "int", "description": "年龄"},
        {"name": "gender", "type": "string", "description": "性别"}
    ]
)

# 3. 记录 ETL 任务的血缘关系
client.create_job(
    job_name="etl.user_profile_daily",
    job_type="BATCH",
    inputs=[{"namespace": "default", "name": "ods.user_profile"}],
    outputs=[{"namespace": "default", "name": "dwd.user_feature"}]
)

# 4. 运行任务时记录血缘
client.create_run(
    job_name="etl.user_profile_daily",
    run_id="run_20240101_001",
    nominal_start_time="2024-01-01T00:00:00Z",
    nominal_end_time="2024-01-01T01:00:00Z"
)

我的建议：血缘追踪不要追求「全量覆盖」。先给核心数据链路（比如训练数据、线上特征）加上血缘，再逐步扩展到其他表。一口吃不成胖子，血缘治理也是迭代的过程。

4.5 知识体系总览

下面这张图把本章的核心逻辑串起来了。你可以看到：数据质量监控和血缘追踪是互补关系——监控负责「发现问题」，血缘负责「定位问题」。

4.6 本章小结

数据质量监控不是「锦上添花」，而是 MLOps 的底线。没有质量保障的数据，再好的模型也是空中楼阁。

Great Expectations：适合单机或小规模数据，声明式 API 上手快，Data Docs 可视化做得好
Deequ：Spark 生态首选，支持大规模分布式校验和异常检测
数据血缘追踪：解决「问题从哪来、影响到哪去」的定位问题，推荐 Marquez 或 DataHub

最后说一句：工具只是手段，流程才是核心。把数据质量监控嵌入到 CI/CD 管道中，让每次数据变更都自动触发校验——这才是 MLOps 的工程化思维。