4、数据质量测试：Great Expectations库入门，定义Expectation Suite，验证数据完整性

数据流水线跑通了，ETL 没报错，数据也落库了——然后呢？你敢直接拿这份数据出报表、喂模型吗？

说实话，我见过太多「流水线绿灯、数据全红」的案例。有一次，上游一个字段类型从 int 悄悄变成了 string，下游聚合直接炸了，但流水线状态还是绿的。从那以后，我养成了一个习惯：流水线必须自带数据质量门禁。

这一章，我们就来聊聊数据质量测试的利器——Great Expectations。我会带你从零入门，定义你的第一个 Expectation Suite，并真正用它拦住脏数据。

核心观点：数据质量不是事后复盘，而是流水线中的一道强制关卡。Great Expectations 就是帮你设卡的工具。

4.1 为什么是 Great Expectations？

市面上数据质量工具不少，但 Great Expectations（简称 GX）有几个让我坚持用它的理由：

  • 声明式期望：你只需要说「我希望 column A 不为空」，GX 帮你生成验证逻辑。
  • 自动化文档：每次验证自动生成数据质量报告，团队都能看懂。
  • 与流水线深度集成：能嵌入 Airflow、Prefect、Spark 等主流调度框架。
  • 社区活跃：Expectation 种类超过 100 种，覆盖常见场景。

我个人最看重的是第一点。你想想看，传统做法是写一堆 if-else 和断言，维护成本极高。GX 把「期望」和「验证」解耦了，改规则不用改代码。

4.2 快速安装与初始化

先装包，这一步没什么特别的：

pip install great_expectations

装完之后，初始化一个数据上下文（Data Context）。这是 GX 的核心配置入口：

great_expectations init

这条命令会在当前目录生成一个 great_expectations/ 文件夹，里面包含：

  • expectations/ —— 存放你的 Expectation Suite
  • checkpoints/ —— 存放验证配置
  • plugins/ —— 自定义插件目录
  • uncommitted/ —— 本地敏感配置（已加入 .gitignore）

小提示：如果你在团队协作，建议把 great_expectations.yml 和 expectations/ 提交到 Git，uncommitted/ 不要提交。

4.3 连接数据源

GX 支持 Pandas、Spark、SQL 等多种数据源。这里我用 Pandas 举例，因为它最轻量，适合入门。

import great_expectations as gx

context = gx.get_context()

# 添加一个 Pandas 数据源
datasource = context.sources.add_pandas("my_pandas_datasource")

# 假设你有一个 DataFrame
import pandas as pd
df = pd.DataFrame({
    "order_id": [1001, 1002, 1003, None],
    "amount": [250.0, 150.0, 300.0, 200.0],
    "status": ["completed", "pending", "completed", "cancelled"]
})

# 将 DataFrame 注册为数据资产
data_asset = datasource.add_dataframe_asset("orders_asset")
batch_request = data_asset.build_batch_request(dataframe=df)

嗯，这里要注意：batch_request 是 GX 里的一个关键概念。它描述的是「你要验证哪一份数据」，可以是一个 DataFrame、一张表、或者一个 SQL 查询结果。

4.4 定义你的第一个 Expectation Suite

Expectation Suite 就是一组期望的集合。你可以用两种方式创建：

  • 交互式：用 Jupyter Notebook 逐步添加期望
  • 编程式：直接写 Python 代码定义

我个人习惯用编程式，因为可重复、可版本控制。下面是一个例子：

expectation_suite = context.add_expectation_suite(
    expectation_suite_name="orders_suite"
)

# 添加期望：order_id 不能为空
expectation_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)

# 添加期望：amount 必须在 0 到 10000 之间
expectation_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0, max_value=10000
    )
)

# 添加期望：status 只能是特定值
expectation_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status", value_set=["completed", "pending", "cancelled"]
    )
)

# 保存 Suite
context.save_expectation_suite(expectation_suite)

你看，代码读起来就像自然语言一样。这就是 GX 的设计哲学——让数据质量规则可读、可理解。

避坑指南：我曾经在定义期望时，把 min_value 和 max_value 写反了，结果所有数据都被标记为异常。建议每次添加期望后，先跑一次小批量验证，确认逻辑正确。

4.5 运行验证并查看结果

Suite 定义好了，接下来就是验证数据：

validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="orders_suite"
)

results = validator.validate()

# 输出验证结果
print(results["statistics"])
# 输出示例：
# {
#   "evaluated_expectations": 3,
#   "successful_expectations": 2,
#   "unsuccessful_expectations": 1,
#   "success_percent": 66.7
# }

为什么会有一个失败？因为我们的数据里 order_id 有一行是 None，而我们的期望要求它不能为空。GX 会精确告诉你哪一行、哪一列出了问题。

你还可以生成一份漂亮的 HTML 报告：

context.build_data_docs()
# 报告会生成在 great_expectations/uncommitted/data_docs/ 下

这份报告可以直接分享给业务方，他们不需要懂代码也能看懂数据质量状况。

4.6 核心概念一览

为了帮你快速建立知识框架，我把 GX 的核心概念整理成了表格：

概念 说明 类比
Data Context GX 的配置中心，管理所有资源 项目的「总控台」
Datasource 数据源连接（Pandas/Spark/SQL） 水龙头接口
Data Asset 具体的数据对象（表/文件/DataFrame） 水龙头里的水
Batch Request 描述你要验证哪一批数据 取水指令
Expectation 一条数据质量规则 水质标准
Expectation Suite 一组期望的集合 水质标准手册
Validator 执行验证的引擎 水质检测仪
Checkpoint 自动化验证配置（含动作） 自动检测流水线

4.7 知识体系总览

下面这张图帮你把本章的核心逻辑串起来：

Great Expectations 数据质量验证流程 数据源 Pandas / Spark / SQL Batch Request 指定验证数据范围 Validator 执行验证引擎 Expectation Suite 一组数据质量规则 验证结果 通过 / 失败 / 统计 关键流程说明 1. 从数据源获取数据 2. 通过 Batch Request 指定验证范围 3. Validator 加载 Expectation Suite 执行验证

4.8 进阶：Checkpoint 与流水线集成

手动验证只是第一步。真正生产环境里，你需要把验证嵌入到流水线中。GX 的 Checkpoint 就是干这个的。

checkpoint = context.add_checkpoint(
    name="orders_checkpoint",
    validations=[
        {
            "batch_request": batch_request,
            "expectation_suite_name": "orders_suite"
        }
    ],
    action_list=[
        {
            "name": "store_validation_result",
            "action": {"class_name": "StoreValidationResultAction"}
        },
        {
            "name": "update_data_docs",
            "action": {"class_name": "UpdateDataDocsAction"}
        }
    ]
)

# 运行 Checkpoint
checkpoint_result = checkpoint.run()

Checkpoint 跑完后，会自动更新数据文档，你可以在 CI/CD 中根据 checkpoint_result["success"] 决定是否阻断流水线。

实战建议：在 Airflow 中，我通常把 Checkpoint 作为一个 PythonOperator 节点。如果验证失败，直接 raise Exception，下游任务自动跳过。这样脏数据永远到不了生产表。

4.9 常见 Expectation 速查

刚开始用 GX，你可能会被上百种 Expectation 搞晕。别急，先记住这几个最常用的：

Expectation 名称 作用 典型场景
ExpectColumnValuesToNotBeNull 列值不能为空 主键、必填字段
ExpectColumnValuesToBeUnique 列值必须唯一 订单号、用户 ID
ExpectColumnValuesToBeBetween 列值在指定范围内 金额、年龄、评分
ExpectColumnValuesToBeInSet 列值必须在指定集合中 状态、类型枚举
ExpectColumnMeanToBeBetween 列均值在指定范围内 监控指标漂移
ExpectTableRowCountToBeBetween 表行数在指定范围内 数据完整性检查

你想想看，有了这些 Expectation，大部分数据质量问题都能在早期被发现。我曾经用 ExpectColumnValuesToBeUnique 抓出过一次上游重复推送数据的问题，那次要不是 GX 拦住，下游的报表就要出大乱子。

4.10 本章小结

这一章我们走通了 Great Expectations 的完整流程：从安装、连接数据源，到定义 Expectation Suite，再到运行验证和集成 Checkpoint。核心就三件事：

  • 定义期望：用声明式规则描述「好数据」长什么样
  • 自动验证：让 Validator 替你检查每一批数据
  • 阻断脏数据：通过 Checkpoint 把质量门禁嵌入流水线
← 上一章 📖 返回目录 下一章 →