Great Expectations：数据质量监控

可视化演示

展示数据管道中的期望规则校验流程，以及CheckPoint验证通过/阻断机制和数据质量报告生成。

数据质量期望规则核心概念

Great Expectations（简称GE）是一个开源的Python数据质量框架，其核心理念是将数据质量检查抽象为「期望」（Expectation）。期望是对数据应该满足的条件的声明式描述，例如「某列不应包含空值」「某列的值应在指定范围内」「两列之间应满足某种数学关系」等。GE通过将这些期望以JSON格式持久化存储，实现了数据质量规则与业务代码的解耦，使得数据工程师、数据科学家和业务分析师可以用统一的语言来定义和沟通数据质量要求。

在GE的架构中，「Expectation」是最基本的构建单元。每个Expectation定义了一个原子级的数据质量断言，例如expect_column_values_to_notBeNull、expect_column_values_to_be_between、expect_column_mean_to_be_between等。GE内置了数十种Expectation类型，覆盖了统计分布校验、正则表达式匹配、集合关系验证、跨列一致性检查等多种场景。用户还可以通过继承Expectation基类来编写自定义期望，以满足特定业务需求。

「Expectation Suite」（期望套件）是多个Expectation的逻辑集合，通常对应一张数据表或一个数据集的完整质量规格。一个Expectation Suite可以包含数十甚至数百条期望规则，这些规则共同构成了对该数据集的质量契约（Data Contract）。当数据发生变化时，GE会用对应的Expectation Suite来验证数据是否仍然满足所有已定义的期望，从而在数据管道的各个环节实现质量门禁。

GE还引入了「Data Doc」的概念，将校验结果以人类可读的HTML报告形式呈现，使非技术人员也能直观了解数据质量状况。这种「将数据质量可视化」的设计哲学，使得GE不仅仅是一个测试工具，更是一个数据质量的沟通和协作平台。通过Data Doc，团队可以在数据质量问题上达成共识，并追踪质量改进的进展。

期望套件组织与声明式规则定义

Great Expectations采用声明式的方式定义数据质量规则，这意味着用户只需要描述「数据应该是什么样」，而不需要编写「如何检查数据」的过程式代码。GE的Expectation Suite以JSON文件的形式存储在本地文件系统或云端存储中，每个Suite对应一个独立的JSON文件，其中包含了该Suite下所有期望规则的完整定义，包括期望类型、参数配置、元数据标注等。这种设计使得期望规则可以被版本控制、审查和共享，就像管理代码一样管理数据质量规则。

GE支持通过多种方式创建和编辑Expectation Suite。最直接的方式是通过Python API以编程方式构建：首先创建一个空的Suite，然后依次调用add_expectation方法添加各种期望规则。另一种方式是通过GE提供的「Data Docs」界面交互式地探索数据，在浏览数据分布的过程中逐步添加期望。GE还支持通过Profiler自动分析数据并生成初始的期望集合，用户可以在此基础上进行调整和优化。无论使用哪种方式，最终生成的期望定义都是标准化的JSON格式，确保了跨环境的可移植性。

在组织层面，GE推荐使用「Expectation Suite命名规范」来管理不同粒度的质量规则。例如，可以为同一张表创建多个Suite：一个基础Suite包含非空和类型检查等通用规则，一个业务Suite包含特定业务逻辑的校验规则，一个ML Suite包含针对机器学习特征数据的统计分布检查。通过这种方式，不同团队和场景可以复用同一套基础规则，同时灵活扩展特定领域的质量检查。GE的Store抽象层支持将期望套件存储在本地文件系统、S3、GCS、Azure Blob等多种后端，方便团队协作和CI/CD集成。

数据源连接与Profiling自动生成

Great Expectations通过「Datasource」抽象层连接各种数据源，支持的关系型数据库包括PostgreSQL、MySQL、SQLite、Snowflake、BigQuery等，同时也支持Pandas DataFrame、Spark DataFrame等内存和分布式计算框架。Datasource的配置信息存储在ge.yml（GE 0.13版本）或great_expectations.yml（GE 0.15+版本）中，包含了连接字符串、执行器类型、数据资产定义等信息。通过统一的Datasource接口，GE可以无缝切换底层数据源而不影响期望规则的定义和执行。

「Profiling」（数据画像）是GE的一项核心功能，它能够自动分析数据集的统计特征并生成初始的Expectation Suite。当对一个Datasource执行profiling时，GE会自动计算每列的基本统计量（如均值、标准差、最小值、最大值、唯一值数量、空值比例等），并根据这些统计结果推断出合理的期望规则。例如，如果某列在样本中没有空值，Profiler会生成expect_column_values_to_not_null期望；如果某列的值在某个范围内，Profiler会生成expect_column_values_to_be_between期望并设置合理的阈值。

需要注意的是，Profiler生成的期望是基于样本数据的统计推断，并不一定完全适用于生产环境。因此，GE建议将Profiler的输出作为起点，由数据工程师根据业务知识和生产数据的实际情况进行调整。例如，Profiler可能根据训练数据推断出某数值列的范围是[0, 100]，但生产数据可能会出现略超此范围的值，这时需要根据业务容忍度适当放宽阈值。GE的Batch Request机制允许用户指定不同的数据批次进行校验，从而区分开发环境和生产环境的质量标准。

GE的Checkpoint机制与Datasource紧密配合，支持定义「批处理请求」（Batch Request）来指定要校验的数据范围。Batch Request可以基于SQL查询、日期分区、文件路径等多种方式精确指定数据批次，确保每次校验都针对正确的数据子集。这种精细的数据定位能力使得GE可以灵活应用于增量数据校验、回溯测试、数据回滚验证等多种场景。

检查点校验执行与管道阻断机制

「Checkpoint」是Great Expectations中用于执行数据质量校验的核心组件。一个Checkpoint将一个Expectation Suite与一个或多个Batch Request绑定在一起，定义了完整的质量检查流程。当Checkpoint运行时，GE会按照Batch Request获取数据批次，然后依次执行Expectation Suite中的所有期望规则，收集每条规则的通过/失败结果，最终生成一个「Validation Result」对象。Validation Result包含了每条期望的执行状态、实际统计值、意外数据的采样等详细信息。

GE的管道阻断机制通过「Action」来实现。Action是Checkpoint执行完成后自动触发的操作，常见的Action包括：将校验结果发送到通知系统（如Slack、邮件）、将结果存储到数据库、更新Data Docs、以及在校验失败时阻断下游管道。通过配置Action，团队可以实现自动化的质量门禁：当关键期望规则失败时，自动停止Airflow DAG中的后续任务，防止有质量问题的数据流入下游系统。这种「快速失败」（Fail Fast）的策略是构建可靠数据管道的重要实践。

在实际的数据工程中，GE通常与Apache Airflow、Prefect、Dagster等编排工具集成。以Airflow为例，可以将GE的Checkpoint执行封装为一个PythonOperator或自定义Sensor，当校验失败时抛出异常以阻断DAG的执行。GE还提供了REST API接口，使得非Python环境也能调用GE的校验功能。通过这种方式，GE可以嵌入到任何数据管道的架构中，作为独立的质量检查层发挥作用。

GE还支持「Evaluation Parameter」（评估参数）机制，允许在Expectation规则中引用动态值。例如，可以设置一个期望规则要求某列的最大值不超过前一次校验时该列最大值的110%，这种基于历史数据的动态阈值在异常检测场景中非常有用。Evaluation Parameter可以在Checkpoint之间传递，实现了跨批次校验的状态关联，使得数据质量监控更加智能化和自适应。

数据文档生成与质量报告闭环

Great Expectations的「Data Docs」模块能够将校验结果和期望定义自动转换为美观的HTML文档，形成数据质量的可视化报告。Data Docs包含三个核心页面：Expectation Suite页面展示所有已定义的质量规则及其元数据；Validation Result页面展示每次校验的详细结果，包括通过/失败的期望数量、意外数据的采样和统计摘要；Site Index页面提供所有文档的导航索引。这些HTML文档可以部署到内部网站、S3静态托管或任何Web服务器上，供团队成员随时查阅。

Data Docs的设计强调「可理解性」和「可操作性」。每条期望规则旁边都会显示人类可读的自然语言描述，例如「column「age」的值应介于0和120之间」，而不是晦涩的代码。校验结果页面会用绿色和红色直观标识通过和失败的规则，失败规则下方会展示具体的意外数据样本，帮助数据工程师快速定位问题根因。这种设计使得Data Docs不仅是技术工具，更是数据团队与业务团队之间沟通数据质量的桥梁。

为了实现质量报告的闭环管理，GE支持与多种监控和通知系统集成。通过配置Notification Action，当校验失败时可以自动发送告警到Slack频道、发送邮件通知、或者触发PagerDuty事件。结合Data Docs的Web部署，团队可以建立一个完整的数据质量仪表盘，实时展示各个数据管道的质量状态。GE的Store机制还支持将校验结果持久化到数据库中，便于历史趋势分析和质量SLA的追踪。

在AI项目的场景中，Data Docs可以用于记录模型训练数据的质量基线、特征漂移监控结果、以及模型推理输入数据的实时校验报告。通过定期生成和归档Data Docs，团队可以建立数据质量的审计轨迹（Audit Trail），满足合规要求并支持事后追溯。当模型性能下降时，可以回溯历史Data Docs来排查是否是数据质量问题导致的，从而形成「数据采集→质量校验→问题告警→根因分析→规则优化」的完整闭环。

AI 应用场景

Python 演示

import great_expectations as gximport pandas as pd # 创建模拟数据data = pd.DataFrame({    "user_id": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],    "age": [25, 32, 28, 45, 19, 55, 38, 22, 41, 30],    "score": [85.5, 92.0, 78.3, 95.1, 88.7, 76.2, 91.4, 83.6, 89.9, 94.2],    "category": ["A", "B", "A", "C", "B", "A", "C", "B", "A", "C"]}) # 初始化GE上下文context = gx.get_context() # 创建Datasourcedatasource = context.datasources.add_or_update_pandas_datasource(name="pandas_ds") # 添加数据资产asset = datasource.add_dataframe_asset(name="user_features") # 创建Batch Requestbatch_request = asset.build_batch_request(dataframe=data) # 创建Expectation Suitesuite = context.suites.add(gx.ExpectationSuite(name="user_features_suite")) # 添加期望规则suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id"))suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="age", min_value=0, max_value=120))suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="score", min_value=0, max_value=100))suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(column="category", value_set=["A", "B", "C"]))suite.add_expectation(gx.expectations.ExpectColumnMeanToBeBetween(column="score", min_value=70, max_value=100)) # 创建Checkpoint并运行checkpoint = context.checkpoints.add(gx.Checkpoint(    name="user_features_checkpoint",    suite_name="user_features_suite",    batch_request=batch_request))result = checkpoint.run() # 输出校验结果print(f"Success: {result.success}")print(f"Statistics: {result.statistics}")

1. import great_expectations as gx导入Great Expectations核心库
2. import pandas as pd导入Pandas用于创建模拟数据
3. data = pd.DataFrame({创建包含user_id、age、score、category四列的模拟用户特征数据
4. context = gx.get_context()初始化GE上下文，这是操作GE所有资源的入口点
5. datasource = context.datasources.add_or_update_pandas_datasource(name="pandas_ds")注册一个Pandas类型的Datasource作为数据连接
6. asset = datasource.add_dataframe_asset(name="user_features")在Datasource下创建名为user_features的数据资产
7. batch_request = asset.build_batch_request(dataframe=data)构建Batch Request，将DataFrame包装为GE可处理的批次
8. suite = context.suites.add(gx.ExpectationSuite(name="user_features_suite"))创建名为user_features_suite的期望套件
9. suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id"))添加期望：user_id列不允许为空值
10. suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="age", min_value=0, max_value=120))添加期望：age列的值应在0到120之间
11. suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(column="category", value_set=["A", "B", "C"]))添加期望：category列的值只能是A、B、C之一
12. checkpoint = context.checkpoints.add(gx.Checkpoint(创建Checkpoint将Suite与Batch Request绑定
13. result = checkpoint.run()执行Checkpoint，运行所有期望规则并收集校验结果
14. print(f"Success: {result.success}")输出校验是否全部通过，success为True表示所有期望均满足

常见问答

常见误区

参考来源

• docGreat Expectations Official DocumentationGreat Expectations官方文档，包含完整的API参考、概念指南、教程和最佳实践
• webpageGreat Expectations GitHub RepositoryGreat Expectations的开源代码仓库，包含框架源码、示例代码和社区贡献
• webpageData Quality Best Practices with Great ExpectationsGreat Expectations官方博客关于数据质量最佳实践的文章