3. Great Expectations 入门
在金融数据场景中，数据质量问题的代价往往不是“报表不好看”，而是估值偏差、风控误判、监管报送错误。Great Expectations（简称 GX）之所以适合金融团队，是因为它把“数据应该长什么样”从口头约定变成了可执行、可版本化、可自动化的断言。本节从核心概念出发，完成安装配置，并跑通第一个 Expectation 与快速验证流程。
一、核心原理：Expectation 是数据契约的最小单元
GX 的核心思想是：用声明式断言描述数据应满足的规则，并让规则可被批量执行、记录和告警。理解以下四个概念，就掌握了 GX 的骨架。
- Expectation（期望）：一条具体断言，例如“trade_id 不能为空”“price 必须大于 0”。它是数据契约的最小单元。
- Expectation Suite（期望套件）：一组 Expectation 的集合，通常对应一张表或一个数据域，如“交易流水质量套件”。
- Batch（批次）：被验证的数据切片，可以是一张表、一个分区、一段 SQL 查询结果或一个 DataFrame。
- Checkpoint（检查点）：把 Batch 与 Suite 绑定并执行验证的调度单元，负责产出 Validation Result 与 Data Docs。
四者关系可以理解为：Suite 定义“规则”，Batch 提供“样本”，Checkpoint 负责“执行”，Validation Result 记录“结论”。在金融实践中，Suite 应纳入 Git 版本管理，Checkpoint 则接入 Airflow 等调度器，实现每日跑批后自动校验。
二、安装与初始化配置
GX 支持 Python 3.8 及以上版本，推荐在虚拟环境中安装，避免与生产依赖冲突。金融团队通常使用 pandas 处理中小规模数据，使用 SQLAlchemy 连接数仓。
# 创建并激活虚拟环境
python -m venv gx-env
source gx-env/bin/activate # Windows 使用 gx-env\Scripts\activate
# 安装 Great Expectations 及常用依赖
pip install great_expectations pandas sqlalchemy
# 验证安装
great_expectations --version
安装完成后，在项目目录中初始化 Data Context。Data Context 是 GX 的配置中心，管理数据源、Suite、Checkpoint 和 Data Docs 的存储位置。
great_expectations init
初始化会生成 great_expectations/ 目录，关键结构如下：
| 目录 / 文件 | 作用 | 金融场景建议 |
|---|---|---|
| great_expectations.yml | Data Context 主配置 | 纳入 Git，敏感连接串用环境变量 |
| expectations/ | 存放 Expectation Suite | 按数据域拆分，如 trade、risk |
| checkpoints/ | 存放 Checkpoint 配置 | 与调度任务一一对应 |
| data_docs/ | 生成的 HTML 验证报告 | 发布到内网供业务方查看 |
| plugins/ | 自定义 Expectation 代码 | 实现行业特有规则 |
great_expectations.yml 中使用 ${DB_PASSWORD} 形式引用环境变量，避免凭据入库。三、第一个 Expectation：从 DataFrame 开始
下面用一段模拟交易数据，创建第一个 Expectation 并完成验证。这种方式无需连接数据库，适合快速理解 GX 的工作方式。
import great_expectations as gx
import pandas as pd
# 1. 获取 Data Context
context = gx.get_context()
# 2. 构造模拟交易数据
df = pd.DataFrame({
"trade_id": ["T001", "T002", "T003", None],
"price": [10.5, 20.0, -3.2, 15.8],
"quantity": [100, 200, 150, 80],
})
# 3. 创建数据源与 Batch
datasource = context.sources.add_pandas("trade_datasource")
asset = datasource.add_dataframe_asset(name="trade_df")
batch_request = asset.build_batch_request(dataframe=df)
# 4. 创建 Expectation Suite
suite = context.add_expectation_suite("trade_quality_suite")
# 5. 添加 Expectation
validator = context.get_validator(
batch_request=batch_request,
expectation_suite_name="trade_quality_suite",
)
validator.expect_column_values_to_not_be_null("trade_id")
validator.expect_column_values_to_be_between("price", min_value=0)
validator.expect_column_values_to_be_between("quantity", min_value=1)
validator.save_expectation_suite()
上述代码定义了三条规则：trade_id 非空、price 大于等于 0、quantity 大于等于 1。运行后，GX 会返回每条 Expectation 的通过情况与失败样本。
四、快速验证流程与结果解读
定义好 Suite 后，通过 Checkpoint 执行验证。Checkpoint 是生产环境的标准入口，它把 Suite、Batch 和 Action 组合在一起。
checkpoint = context.add_or_update_checkpoint(
name="trade_checkpoint",
validations=[{
"batch_request": batch_request,
"expectation_suite_name": "trade_quality_suite",
}],
)
result = checkpoint.run()
print(result.success)
执行后典型输出如下：
Calculating Metrics: 100%|██████████| 6/6 [00:00<00:00, 320.15it/s]
{
"success": false,
"statistics": {
"evaluated_expectations": 3,
"successful_expectations": 2,
"unsuccessful_expectations": 1,
"success_percent": 66.67
}
}
结果中 success 为 false，说明存在失败规则。结合 Data Docs 可以定位到具体是哪一列、哪一行不满足。常见失败原因包括：trade_id 存在空值、price 出现负数。此时应回到上游数据管道排查，而不是直接修改断言。
五、常见误区与复盘
- 误区一：一次性写几十条 Expectation。建议按数据域分批建设，先覆盖主键、金额、时间三类关键字段，再逐步扩展。
- 误区二：只在本地跑，不接入调度。GX 的价值在于持续验证，Checkpoint 应接入 Airflow 或 Cron，失败时触发告警。
- 误区三：忽略 Data Docs。Data Docs 是业务方理解数据质量的主要入口，应定期发布并保留历史记录。
- 误区四：Suite 不做版本管理。规则变更应像代码一样走评审流程，避免“悄悄放宽标准”。
复盘本节：核心概念是 Suite、Batch、Checkpoint、Validation Result 四件套；安装配置的关键是 Data Context 与目录结构；第一个 Expectation 从 DataFrame 入手最直观；快速验证流程则通过 Checkpoint 完成并输出可读结果。掌握这些后，就可以进入更复杂的数据源接入与自定义 Expectation 阶段。