汤不热吧在构建可靠的AI模型时，数据的质量是决定模型性能和稳定性的基石，正所谓“Garbage In, Garbage Out”。如果训练数据或推理数据中包含缺失值、异常值或分布偏移，即使是最先进的模型也会失效。Great Expectations (GX) 是一个强大的开源工具包，它允许我们定义、验证和文档化数据的“期望” (Expectations)，从而在数据进入模型流水线之前捕获质量问题。
本文将深入探讨如何在实际的AI基础设施中集成Great Expectations，实现自动化数据质量检查。
1. 基础环境搭建
首先，我们需要安装Great Expectations及其依赖库。
pip install great-expectations pandas
2. 初始化 Data Context
Great Expectations 的核心概念是 Data Context，它存储了所有配置、期望套件（Expectation Suites）和校验结果。我们将使用命令行工具进行初始化：
great_expectations init
执行此命令后，GX 会在当前目录下创建 great_expectations/ 文件夹，其中包含配置文件 (great_expectations.yml)、期望套件和数据文档 (Data Docs) 的存储位置。
3. 连接数据源并创建期望套件
假设我们有一个名为 feature_data.csv 的特征数据集，需要在模型训练前进行质量校验。
示例数据（feature_data.csv）：
user_id,feature_a,feature_b,target
1,10.5,True,0
2,,False,1
3,20.1,True,0
4,5.0,True,1
5,15.8,False,0
6,1000.0,True,0
我们首先通过 CLI 连接数据源：
great_expectations datasource new
选择 1) Files on a filesystem (for example, CSV, JSON, etc.)，然后选择 2) Pandas 作为执行引擎，并指定数据路径。假设数据源名称为 my_local_data。
接下来，我们创建期望套件。GX 最方便的功能是使用 Data Assistant 自动生成初始期望。
great_expectations suite new
选择刚刚配置的数据源和资产。GX 会启动一个 Jupyter Notebook。在 Notebook 中，我们可以使用交互式工具定义如下关键期望：
- user_id 列必须是唯一的 (expect_column_values_to_be_unique)。
- feature_a 列的非空值比例应大于 90% (expect_column_values_to_not_be_null)。
- feature_a 列的值必须在合理范围内 (例如，0 到 100) (expect_column_values_to_be_between)。
- target 列的值必须是 0 或 1 (expect_column_distinct_values_to_be_in_set)。
保存 Notebook 后，GX 会生成一个 JSON 文件（例如 expectations/baseline_feature_suite.json），其中包含了所有的质量规则。
4. 自动化校验 (Checkpoint)
在生产环境中，我们不能手动运行 Notebook。我们需要设置一个 Checkpoint 来自动化执行校验，并在数据质量不达标时中断流水线。
以下 Python 脚本展示了如何在 CI/CD 或 MLOps 管道中运行 Checkpoint：
import great_expectations as gx
# 1. 获取 Data Context
context = gx.get_context()
# 2. 定义 Batch Request
# 假设数据源名为 'my_local_data'，数据资产名为 'feature_data.csv'
batch_request = gx.data_context.types.BatchRequest(
datasource_name="my_local_data",
data_connector_name="default_inferred_data_connector_name", # 使用默认的 Data Connector
data_asset_name="feature_data.csv",
)
# 3. 配置 Checkpoint
checkpoint_config = {
"name": "feature_data_checkpoint",
"config_version": 1,
"class_name": "SimpleCheckpoint",
"batch_request": batch_request,
"validations": [
{
"batch_request": batch_request,
"expectation_suite_name": "baseline_feature_suite", # 期望套件的名称
}
],
}
# 4. 运行 Checkpoint
print("开始执行数据质量检查...")
results = context.run_checkpoint(**checkpoint_config)
# 5. 处理结果
if results["success"]:
print("\n✅ 数据质量检查成功！数据集符合所有期望。可以继续模型训练。")
else:
print(f"\n❌ 数据质量检查失败，发现 {results['statistics']['unsuccessful_validation_count']} 个不符合项。")
print("请访问 Data Docs 查看详细报告。")
# 在实际流水线中，此处应触发警报并停止下游步骤
# context.open_data_docs()
# 6. 构建并打开 Data Docs (可选，用于人工审核)
context.build_data_docs()
print("Data Docs 已生成，请在浏览器中打开以查看详细报告。")
5. 结果分析 (Data Docs)
当校验失败时，GX 会生成静态 HTML 报告，即 Data Docs。这些文档提供了清晰的可视化界面，展示了哪些期望通过了，哪些失败了，以及数据分布的统计信息。例如，在我们的示例数据中，feature_a 的值 1000.0 可能会因为超出 expect_column_values_to_be_between(0, 100) 的期望而导致 Checkpoint 失败。
通过将 Checkpoint 结果集成到 MLOps 仪表板或 Slack/邮件通知中，我们可以确保任何数据质量问题在影响模型训练或推理服务之前被及时捕获和解决。
总结
Great Expectations 是数据质量管理在AI基础设施中的一把利器。通过将数据质量规则显式化为可执行的期望，并在关键数据摄取点设置自动化 Checkpoint，我们能够大大提升AI系统的健壮性和可信度，从而确保模型在面对真实世界数据时能够保持预期的性能。