Great Expectations实战：构建数据质量测试套件
·
Great Expectations实战：构建数据质量测试套件
项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations 本文深入探讨了Great Expectations框架的核心组件Expectation Suite（期望套件）的创建、管理与应用。详细介绍了如何通过多种方式创建Suite，包括基础创建、JSON配置导入，以及如何进行期望规则的CRUD操作、参数化配置和元数据管理。文章还涵盖了Suite的持久化存储、高级管理功能如渲染和序列化，以及批量导入导出的最佳实践，为构建可维护、可扩展的数据质量保障体系提供了系统化指导。
Expectation Suite的创建与管理
Great Expectations的核心组件之一就是Expectation Suite（期望套件），它是一个集合式的期望配置容器，用于组织和维护数据质量测试规则。Expectation Suite提供了强大的管理功能，让数据团队能够系统地定义、存储和执行数据质量检查。
Expectation Suite的核心概念
Expectation Suite是一个类似集合的数据结构，用于存储和管理多个Expectation（期望）。每个Expectation代表一个具体的数据质量检查规则，而Suite则将这些规则组织成一个逻辑单元，通常对应一个特定的数据源或业务场景。
创建Expectation Suite
创建Expectation Suite有多种方式，从简单的内存创建到完整的持久化存储：
基础创建方式
import great_expectations as gx
# 方式1：通过DataContext创建
context = gx.get_context()
suite = context.suites.add(name="my_data_suite")
# 方式2：直接实例化
from great_expectations.core.expectation_suite import ExpectationSuite
suite = ExpectationSuite(
name="customer_data_suite",
expectations=[], # 初始为空
meta={"description": "客户数据质量检查套件", "owner": "数据质量团队"},
notes="用于验证客户主数据的完整性和准确性"
)
从JSON配置创建
# 从JSON配置创建Suite
suite_config = {
"name": "transaction_suite",
"expectations": [
{
"expectation_type": "expect_column_to_exist",
"kwargs": {"column": "transaction_id"}
}
],
"meta": {
"data_source": "transaction_db",
"version": "1.0"
}
}
suite = ExpectationSuite(**suite_config)
Expectation的管理操作
Expectation Suite提供了完整的CRUD操作来管理期望规则：
添加Expectation
# 添加各种类型的Expectation
suite.add_expectation(
ExpectColumnToExist(column="user_id")
)
suite.add_expectation(
ExpectColumnValuesToBeBetween(
column="age",
min_value=18,
max_value=100
)
)
suite.add_expectation(
ExpectColumnValuesToNotBeNull(column="email")
)
批量操作示例
# 批量添加Expectation
expectations_to_add = [
ExpectColumnToExist(column="order_id"),
ExpectColumnValuesToBeUnique(column="order_id"),
ExpectColumnValuesToNotBeNull(column="customer_id"),
ExpectColumnValuesToBeInSet(
column="status",
value_set=["pending", "completed", "cancelled"]
)
]
for expectation in expectations_to_add:
suite.add_expectation(expectation)
删除和更新操作
# 删除Expectation
if suite.expectations:
suite.delete_expectation(suite.expectations[0])
# 更新Expectation（通过删除后重新添加）
old_expectation = suite.expectations[0]
suite.delete_expectation(old_expectation)
# 创建更新后的Expectation
updated_expectation = ExpectColumnValuesToBeBetween(
column="price",
min_value=0,
max_value=10000 # 更新最大值
)
suite.add_expectation(updated_expectation)
Suite参数化配置
Expectation Suite支持参数化配置，使得期望规则可以动态适应不同的数据环境：
# 定义Suite参数
suite.suite_parameters = {
"min_salary": 3000,
"max_salary": 20000,
"allowed_departments": ["IT", "Finance", "HR"]
}
# 使用参数的Expectation
suite.add_expectation(
ExpectColumnValuesToBeBetween(
column="salary",
min_value={"$PARAMETER": "min_salary"},
max_value={"$PARAMETER": "max_salary"}
)
)
suite.add_expectation(
ExpectColumnValuesToBeInSet(
column="department",
value_set={"$PARAMETER": "allowed_departments"}
)
)
元数据管理
Expectation Suite提供了丰富的元数据管理功能：
# 设置Suite元数据
suite.meta = {
"description": "员工数据质量检查套件",
"data_source": "hr_database.employees",
"created_by": "data_engineer@company.com",
"created_date": "2024-01-15",
"last_modified": "2024-01-20",
"version": "2.1",
"tags": ["hr", "employee", "critical"],
"data_freshness_requirements": {
"max_age_days": 1,
"update_frequency": "daily"
}
}
# 添加说明笔记
suite.notes = [
"此套件用于验证员工主数据的完整性",
"包含基本字段验证和业务规则检查",
"定期更新以适应业务变化"
]
持久化存储与管理
Expectation Suite支持多种存储后端，确保配置的持久化和版本管理：
保存到存储系统
# 保存Suite到配置的存储后端
suite.save()
# 检查Suite是否已保存
if suite._has_been_saved():
print("Suite已持久化存储")
# 验证Suite的新鲜度（是否与存储版本一致）
freshness_diagnostics = suite.is_fresh()
if not freshness_diagnostics.success:
print("Suite需要更新:", freshness_diagnostics.errors)
存储后端配置示例
# 查看当前存储配置
store = suite._store
print(f"存储类型: {type(store).__name__}")
print(f"存储后端: {store.store_backend}")
# 获取Suite的存储键
storage_key = store.get_key(name=suite.name, id=suite.id)
print(f"存储键: {storage_key}")
高级管理功能
Expectation的渲染和序列化
# 渲染Expectation内容（用于显示和文档生成）
suite.render()
# 转换为字典格式（用于序列化）
suite_dict = suite.to_dict()
print(json.dumps(suite_dict, indent=2))
# 转换为JSON字符串
suite_json = suite.to_json()
批量导入导出
# 从文件导入Suite配置
def import_suite_from_file(file_path):
with open(file_path, 'r') as f:
config = json.load(f)
return ExpectationSuite(**config)
# 导出Suite到文件
def export_suite_to_file(suite, file_path):
suite_data = suite.to_dict()
with open(file_path, 'w') as f:
json.dump(suite_data, f, indent=2)
# 使用示例
export_suite_to_file(suite, "my_suite.json")
imported_suite = import_suite_from_file("my_suite.json")
最佳实践建议
根据项目经验，以下是一些Expectation Suite管理的最佳实践：
- 命名规范：使用清晰的命名约定，如
{数据源}_{业务域}_suite - 版本控制：在meta中维护版本信息，便于追踪变更
- 模块化设计：按业务域拆分多个Suite，避免单个Suite过于庞大
- 参数化配置：充分利用Suite参数实现环境适配
- 文档完善：使用notes和meta字段充分记录Suite的用途和约束
# 最佳实践示例
best_practice_suite = ExpectationSuite(
name="ecommerce_orders_suite",
meta={
"version": "1.0.0",
"domain": "E-commerce",
"description": "电商订单数据质量验证套件",
"owner": "data-quality-team",
"expected_columns": ["order_id", "customer_id", "amount", "status"]
},
notes=[
"验证订单数据的完整性和业务规则一致性",
"包含基础字段验证和金额计算逻辑检查"
]
)
通过系统化的Expectation Suite管理，数据团队能够建立可维护、可扩展的数据质量保障体系，确保数据质量规则的一致性和可靠性。
常用Expectation类型与应用场景
Great Expectations提供了丰富的Expectation类型，每种类型针对不同的数据质量验证需求。了解这些Expectation的分类和应用场景对于构建有效的数据质量测试套件至关重要。
Expectation分类体系
Great Expectations的Expectation体系采用层次化设计，主要分为以下几个核心类别：
核心Expectation类型详解
1. 表级别Expectations (BatchExpectation)
表级别Expectations对整个数据表进行验证，关注表的整体属性和统计特征。
常用表级别Expectations：
| Expectation名称 | 功能描述 | 应用场景 | 示例代码 |
|---|---|---|---|
expect_table_row_count_to_be_between |
验证表行数范围 | 数据量监控，ETL完整性检查 | expect_table_row_count_to_be_between(min_value=1000, max_value=5000) |
expect_table_column_count_to_equal |
验证表列数 | 数据结构一致性检查 | expect_table_column_count_to_equal(value=10) |
expect_table_columns_to_match_ordered_list |
验证列顺序和名称 | 数据模式验证 | expect_table_columns_to_match_ordered_list(['id', 'name', 'age']) |
应用场景示例：
- 数据管道完整性验证
- 数据模式一致性检查
- 数据量波动监控
2. 列映射Expectations (ColumnMapExpectation)
列映射Expectations对单列的每个值进行逐行验证，是最常用的Expectation类型。
常用列映射Expectations：
详细应用场景：
数据完整性验证：
# 检查关键字段非空
validator.expect_column_values_to_not_be_null(
column="user_id",
mostly=0.99 # 允许1%的空值
)
# 检查可选字段的空值比例
validator.expect_column_values_to_be_null(
column="middle_name",
mostly=0.8 # 期望80%为空值
)
数据有效性验证：
# 枚举值验证
validator.expect_column_values_to_be_in_set(
column="status",
value_set=["active", "inactive", "pending"]
)
# 数值范围验证
validator.expect_column_values_to_be_between(
column="age",
min_value=0,
max_value=120
)
# 正则表达式验证
validator.expect_column_values_to_match_regex(
column="email",
regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
)
数据一致性验证：
# 唯一性约束
validator.expect_column_values_to_be_unique(column="user_id")
# 时序数据单调性
validator.expect_column_values_to_be_increasing(column="timestamp")
# 数据格式一致性
validator.expect_column_values_to_match_strftime_format(
column="birth_date",
strftime_format="%Y-%m-%d"
)
3. 列聚合Expectations (ColumnAggregateExpectation)
列聚合Expectations对整列数据进行聚合计算和统计验证。
常用列聚合Expectations：
| 统计指标 | Expectation名称 | 应用场景 |
|---|---|---|
| 平均值 | expect_column_mean_to_be_between |
数据分布中心趋势验证 |
| 中位数 | expect_column_median_to_be_between |
抗异常值的数据中心验证 |
| 标准差 | expect_column_stdev_to_be_between |
数据离散程度验证 |
| 唯一值数量 | expect_column_unique_value_count_to_be_between |
数据多样性验证 |
| 分位数 | expect_column_quantile_values_to_be_between |
数据分布形态验证 |
应用示例：
# 收入数据的统计验证
validator.expect_column_mean_to_be_between(
column="income",
min_value=50000,
max_value=80000
)
validator.expect_column_stdev_to_be_between(
column="income",
min_value=10000,
max_value=20000
)
# 分类数据的多样性验证
validator.expect_column_unique_value_count_to_be_between(
column="product_category",
min_value=5,
max_value=20
)
4. 多列关联Expectations
多列关联Expectations验证多个列之间的关系和业务逻辑。
常用多列Expectations：
# 多列组合唯一性（复合主键）
validator.expect_compound_columns_to_be_unique(
column_list=["user_id", "session_id"]
)
# 列值对等关系
validator.expect_column_pair_values_to_be_equal(
column_A="total_price",
column_B="unit_price * quantity"
)
# 列值大小关系
validator.expect_column_pair_values_a_to_be_greater_than_b(
column_A="revenue",
column_B="cost"
)
# 多列求和验证
validator.expect_multicolumn_sum_to_equal(
column_list=["jan_sales", "feb_sales", "mar_sales"],
sum_total=1000000
)
按数据质量维度分类
从数据质量维度来看，Expectations可以进一步分类：
| 数据质量维度 | 代表Expectations | 业务价值 |
|---|---|---|
| 完整性 | expect_column_values_to_not_be_null |
确保关键数据不缺失 |
| 准确性 | expect_column_values_to_be_between |
保证数据值域正确 |
| 一致性 | expect_column_values_to_match_regex |
维持数据格式统一 |
| 唯一性 | expect_column_values_to_be_unique |
避免数据重复 |
| 时效性 | expect_column_values_to_be_increasing |
验证时间序列连续性 |
| 有效性 | expect_column_values_to_be_in_set |
确保数据符合业务规则 |
实际应用场景组合
在实际项目中，通常需要组合使用多种Expectation类型：
用户数据质量检查：
# 表级别检查
validator.expect_table_row_count_to_be_between(min_value=1000, max_value=10000)
# 列完整性检查
validator.expect_column_values_to_not_be_null(column="user_id")
validator.expect_column_values_to_not_be_null(column="email", mostly=0.95)
# 数据有效性检查
validator.expect_column_values_to_be_in_set(
column="status",
value_set=["active", "inactive", "suspended"]
)
validator.expect_column_values_to_match_regex(
column="phone",
regex=r"^\+?[1-9]\d{1,14}$"
)
# 业务逻辑检查
validator.expect_column_values_to_be_between(
column="age",
min_value=18,
max_value=100
)
validator.expect_column_pair_values_a_to_be_greater_than_b(
column_A="registration_date",
column_B="birth_date"
)
电商订单数据验证：
# 订单表结构验证
validator.expect_table_columns_to_match_set(
column_set=["order_id", "customer_id", "order_date", "total_amount", "status"]
)
# 金额数据验证
validator.expect_column_values_to_be_between(
column="total_amount",
min_value=0,
max_value=10000
)
# 状态流转验证
validator.expect_column_values_to_be_increasing(
column="update_timestamp",
strictly=True
)
# 业务规则验证
validator.expect_compound_columns_to_be_unique(
column_list=["order_id", "customer_id"]
)
高级应用场景
条件验证
使用mostly参数实现条件验证，允许一定比例的数据不符合期望：
# 允许5%的数据不符合邮箱格式
validator.expect_column_values_to_match_regex(
column="email",
regex=email_regex,
mostly=0.95
)
# 允许10%的空值
validator.expect_column_values_to_not_be_null(
column="optional_field",
mostly=0.9
)
动态阈值验证
使用Suite Parameters实现动态阈值：
validator.expect_column_mean_to_be_between(
column="daily_sales",
min_value={"$PARAMETER": "min_daily_sales"},
max_value={"$PARAMETER": "max_daily_sales"}
)
多数据源一致性验证
# 验证两个表的行数一致
validator.expect_table_row_count_to_equal_other_table(
other_table_name="customers_backup"
)
# 验证关键指标的一致性
validator.expect_column_mean_to_be_between(
column="revenue",
min_value={"$PARAMETER": "expected_min_revenue"},
max_value={"$PARAMETER": "expected_max_revenue"}
)
最佳实践建议
- 分层验证：先进行表级别验证，再进行列级别验证
- 渐进严格：从宽松的验证开始，逐步增加严格度
- 业务导向：根据业务重要性确定验证优先级
- 性能考虑：对大数据集使用抽样验证
- 文档化：为每个Expectation添加清晰的注释和业务说明
通过合理组合不同类型的Expectations，可以构建出全面而高效的数据质量监控体系，确保数据的可靠性、准确性和一致性，为数据驱动的决策提供坚实基础。
数据验证结果分析与解读
在Great Expectations中，数据验证是整个数据质量保障流程的核心环节。当执行验证操作后，系统会生成详细的验证结果，这些结果包含了丰富的信息，帮助我们全面了解数据质量状况。
验证结果结构解析
Great Expectations的验证结果采用分层结构设计，主要包含以下几个核心组件：
1. 期望验证结果（ExpectationValidationResult）
每个单独的期望验证都会生成一个ExpectationValidationResult对象，包含以下关键信息：
{
"success": True/False, # 验证是否成功
"expectation_config": { # 期望配置信息
"expectation_type": "expect_column_to_exist",
"kwargs": {"column": "user_id"}
},
"result": { # 验证结果详情
"observed_value": 10000, # 观测到的实际值
"element_count": 10000, # 元素总数
"missing_count": 0, # 缺失值数量
"missing_percent": 0.0 # 缺失值百分比
},
"meta": {}, # 元数据信息
"exception_info": { # 异常信息（如果验证失败）
"raised_exception": False,
"exception_message": "",
"exception_traceback": ""
}
}
2. 期望套件验证结果（ExpectationSuiteValidationResult）
当执行完整的期望套件验证时，会生成ExpectationSuiteValidationResult，它聚合了所有单个期望的验证结果：
结果分析关键指标
成功率分析
验证结果的核心指标是成功率，它反映了数据质量的整体状况：
# 计算整体成功率
total_expectations = len(validation_result.results)
successful_expectations = sum(1 for result in validation_result.results if result.success)
overall_success_rate = successful_expectations / total_expectations * 100
详细统计信息
Great Expectations提供了丰富的统计信息来帮助分析数据质量：
| 统计指标 | 说明 | 分析价值 |
|---|---|---|
evaluated_expectations |
已评估的期望数量 | 了解验证范围 |
successful_expectations |
成功的期望数量 | 数据质量整体水平 |
unsuccessful_expectations |
失败的期望数量 | 需要关注的问题数量 |
success_percent |
成功率百分比 | 量化数据质量 |
success |
整体是否成功 | 快速判断状态 |
失败结果深度分析
当验证失败时，需要深入分析失败原因：
1. 缺失值分析
# 分析缺失值情况
if not result.success and "missing_percent" in result.result:
missing_percent = result.result["missing_percent"]
if missing_percent > 5.0:
print(f"警告：列 {column} 缺失值比例高达 {missing_percent}%")
2. 范围越界分析
# 分析数值范围异常
if result.expectation_config["expectation_type"] == "expect_column_values_to_be_between":
min_val = result.expectation_config["kwargs"]["min_value"]
max_val = result.expectation_config["kwargs"]["max_value"]
observed_min = result.result.get("observed_value", {}).get("min")
observed_max = result.result.get("observed_value", {}).get("max")
if observed_min < min_val or observed_max > max_val:
print(f"数值范围异常：期望 [{min_val}, {max_val}]，实际 [{observed_min}, {observed_max}]")
3. 唯一性违反分析
# 分析唯一性约束
if result.expectation_config["expectation_type"] == "expect_column_values_to_be_unique":
unexpected_count = result.result.get("unexpected_count", 0)
if unexpected_count > 0:
print(f"唯一性违反：发现 {unexpected_count} 个重复值")
可视化分析工具
Great Expectations提供了多种可视化方式来分析验证结果：
1. 数据文档（Data Docs）
数据文档自动生成详细的HTML报告，包含：
- 期望套件概览
- 每个期望的详细结果
- 可视化图表（直方图、箱线图等）
- 失败期望的详细分析
2. 验证结果统计面板
3. 时间序列分析
对于定期执行的验证，可以分析验证结果的时间趋势：
# 时间序列成功率分析
success_rates_over_time = []
for validation_run in historical_validation_results:
success_rate = validation_run.statistics["success_percent"]
success_rates_over_time.append({
"timestamp": validation_run.meta["run_id"]["run_time"],
"success_rate": success_rate
})
高级分析技巧
1. 根本原因分析（RCA）
当验证失败时，进行根本原因分析：
def perform_root_cause_analysis(validation_result):
failed_expectations = [r for r in validation_result.results if not r.success]
root_causes = []
for failure in failed_expectations:
expectation_type = failure.expectation_config["expectation_type"]
if expectation_type.startswith("expect_column"):
column = failure.expectation_config["kwargs"].get("column")
root_causes.append({
"column": column,
"expectation_type": expectation_type,
"issue": analyze_failure_reason(failure)
})
return root_causes
2. 影响评估
评估验证失败对下游系统的影响：
def assess_impact(failed_expectations, impact_matrix):
total_impact_score = 0
for failure in failed_expectations:
expectation_type = failure.expectation_config["expectation_type"]
impact = impact_matrix.get(expectation_type, 1) # 默认影响系数为1
total_impact_score += impact
return {
"total_impact": total_impact_score,
"severity": "高危" if total_impact_score > 10 else "中危" if total_impact_score > 5 else "低危"
}
3. 自动化告警规则
基于验证结果配置自动化告警：
alert_rules = {
"critical": {
"condition": lambda result: result.statistics["success_percent"] < 80,
"message": "数据质量严重下降，成功率低于80%"
},
"warning": {
"condition": lambda result: result.statistics["success_percent"] < 95,
"message": "数据质量警告，成功率低于95%"
}
}
def check_alerts(validation_result):
triggered_alerts = []
for level, rule in alert_rules.items():
if rule["condition"](validation_result):
triggered_alerts.append({
"level": level,
"message": rule["message"],
"success_rate": validation_result.statistics["success_percent"]
})
return triggered_alerts
最佳实践建议
- 定期审查验证结果：建立定期审查机制，重点关注失败和警告的期望
- 设置合理的阈值：根据业务重要性设置不同的成功率阈值
- 建立问题跟踪流程：将验证失败转化为具体的数据质量问题工单
- 监控趋势变化：关注数据质量指标的时间趋势，及时发现潜在问题
- 与数据血缘结合：将验证结果与数据血缘信息关联，了解影响范围
通过系统化的验证结果分析和解读，团队可以建立有效的数据质量监控体系，确保数据的可靠性、准确性和一致性，为数据驱动的决策提供坚实保障。
自定义Expectation开发指南
Great Expectations提供了强大的自定义Expectation功能，允许开发者根据特定业务需求创建专属的数据质量验证规则。本文将深入探讨自定义Expectation的开发流程、核心组件和最佳实践。
Expectation架构概述
Great Expectations的Expectation系统采用分层架构设计，通过基类和多种类型的Expectation模板提供灵活的扩展能力。
核心组件解析
1. Expectation基类
所有自定义Expectation都必须继承自基类Expectation，并实现以下核心方法：
from great_expectations.expectations.expectation import Expectation
from great_expectations.expectations.core import ExpectationConfiguration
from great_expectations.core.expectation_validation_result import ExpectationValidationResult
class CustomExpectation(Expectation):
# 定义领域键和成功键
domain_keys = ("column",)
success_keys = ("min_value", "max_value")
def _validate(self, metrics, runtime_configuration=None, execution_engine=None):
"""核心验证逻辑"""
# 实现具体的验证逻辑
pass
def get_validation_dependencies(self, execution_engine=None, runtime_configuration=None):
"""定义验证依赖的度量指标"""
# 配置所需的度量指标
pass
2. 配置验证
每个Expectation都需要实现配置验证方法，确保输入参数的合法性：
def validate_configuration(self, configuration: Optional[ExpectationConfiguration] = None):
"""验证配置参数的有效性"""
if configuration is None:
configuration = self.configuration
# 检查必需参数
if "column" not in configuration.kwargs:
raise ValueError("column参数是必需的")
# 验证参数类型和范围
min_val = configuration.kwargs.get("min_value")
if min_val is not None and not isinstance(min_val, (int, float)):
raise ValueError("min_value必须是数字类型")
Expectation类型详解
Great Expectations提供了多种预定义的Expectation类型，开发者可以根据需求选择合适的基类：
| Expectation类型 | 适用场景 | 核心方法 |
|---|---|---|
| ColumnMapExpectation | 列级逐行验证 | _pandas, _sqlalchemy, _spark |
| ColumnAggregateExpectation | 列级聚合验证 | 聚合函数实现 |
| BatchExpectation | 批次级验证 | 全表操作 |
| MulticolumnMapExpectation | 多列关联验证 | 多列处理逻辑 |
ColumnMapExpectation示例
from great_expectations.expectations.expectation import ColumnMapExpectation
class ExpectColumnValuesToBeCustom(ColumnMapExpectation):
# 元数据定义
library_metadata = {
"tags": ["custom", "validation"],
"contributors": ["YourName"]
}
# 映射到不同执行引擎的方法
def _pandas(cls, series, **kwargs):
"""Pandas执行引擎实现"""
min_value = kwargs.get("min_value")
max_value = kwargs.get("max_value")
return (series >= min_value) & (series <= max_value)
def _sqlalchemy(cls, column, _dialect, **kwargs):
"""SQLAlchemy执行引擎实现"""
min_value = kwargs.get("min_value")
max_value = kwargs.get("max_value")
from sqlalchemy import and_
return and_(column >= min_value, column <= max_value)
def _spark(cls, column, **kwargs):
"""Spark执行引擎实现"""
min_value = kwargs.get("min_value")
max_value = kwargs.get("max_value")
from pyspark.sql.functions import col
return (col(column) >= min_value) & (col(column) <= max_value)
渲染器开发
为了在Data Docs中提供友好的展示，需要实现渲染器方法：
from great_expectations.render import renderer
from great_expectations.render.renderer_configuration import RendererConfiguration
class ExpectColumnValuesToBeCustom(ColumnMapExpectation):
# ... 其他代码 ...
@renderer(renderer_type="atomic.prescriptive.summary")
def _prescriptive_summary(cls, configuration=None, result=None, runtime_configuration=None):
"""预设性摘要渲染器"""
renderer_configuration = RendererConfiguration(
configuration=configuration,
result=result,
runtime_configuration=runtime_configuration
)
min_val = configuration.kwargs.get("min_value")
max_val = configuration.kwargs.get("max_value")
column = configuration.kwargs.get("column")
template = f"列 {column} 的值应在 [{min_val}, {max_val}] 范围内"
return [{
"content_block_type": "string_template",
"string_template": {
"template": template,
"params": {}
}
}]
度量指标集成
自定义Expectation需要定义所需的度量指标：
def get_validation_dependencies(self, execution_engine=None, runtime_configuration=None):
"""定义验证依赖的度量指标"""
dependencies = super().get_validation_dependencies(
execution_engine, runtime_configuration
)
# 添加自定义度量指标
dependencies.set_metric_configuration(
"column_values.custom_metric",
MetricConfiguration(
metric_name="column_values.custom_metric",
metric_domain_kwargs=self.get_domain_kwargs(),
metric_value_kwargs=self.get_success_kwargs()
)
)
return dependencies
测试策略
为确保自定义Expectation的质量，需要编写全面的测试用例：
import pytest
import pandas as pd
from great_expectations.core import ExpectationConfiguration
class TestCustomExpectation:
def test_expectation_with_valid_data(self):
"""测试有效数据的情况"""
df = pd.DataFrame({"test_column": [1, 2, 3, 4, 5]})
expectation = ExpectColumnValuesToBeCustom(
column="test_column",
min_value=1,
max_value=5
)
result = expectation.validate(df)
assert result.success
def test_expectation_with_invalid_data(self):
"""测试无效数据的情况"""
df = pd.DataFrame({"test_column": [0, 2, 3, 4, 6]})
expectation = ExpectColumnValuesToBeCustom(
column="test_column",
min_value=1,
max_value=5
)
result = expectation.validate(df)
assert not result.success
assert len(result.result["unexpected_list"]) == 2
部署与注册
完成开发后，需要将自定义Expectation注册到Great Expectations中：
# 在__init__.py中注册
from .custom_expectations import ExpectColumnValuesToBeCustom
__all__ = [
"ExpectColumnValuesToBeCustom"
]
# 或者在运行时动态注册
from great_expectations.expectations.registry import register_expectation
register_expectation(ExpectColumnValuesToBeCustom)
性能优化建议
开发自定义Expectation时需要考虑性能因素：
- 批量操作：尽量使用向量化操作而非逐行处理
- 懒加载：延迟计算直到真正需要时
- 缓存机制：对重复计算的结果进行缓存
- 执行引擎优化：为不同执行引擎提供专门的实现
def _pandas(cls, series, **kwargs):
"""优化后的Pandas实现"""
# 使用向量化操作提高性能
min_value = kwargs.get("min_value")
max_value = kwargs.get("max_value")
# 批量处理，避免逐行操作
return series.between(min_value, max_value)
错误处理与日志
健全的错误处理机制是高质量Expectation的关键：
def _validate(self, metrics, runtime_configuration=None, execution_engine=None):
try:
# 主要的验证逻辑
custom_metric = metrics["column_values.custom_metric"]
# 处理边界情况
if custom_metric is None:
return self._build_evr(
success=False,
message="无法计算自定义度量指标"
)
success = self._evaluate_metric(custom_metric)
return self._build_evr(success=success)
except Exception as e:
logger.error(f"验证过程中发生错误: {str(e)}")
return self._build_evr(
success=False,
message=f"验证失败: {str(e)}"
)
通过遵循这些开发指南，您可以创建出功能强大、性能优异且易于维护的自定义Expectation，为数据质量验证提供精准的业务规则支持。
总结
Great Expectations通过其强大的Expectation Suite管理系统和丰富的内置期望类型，为数据团队提供了构建全面数据质量测试套件的坚实基础。本文详细阐述了从Suite的创建、管理到自定义Expectation开发的完整流程，包括架构概述、核心组件解析、不同类型Expectation的实现，以及渲染器开发、度量指标集成和测试策略。通过遵循文中的最佳实践和性能优化建议，开发者可以创建出精准、高效且易于维护的数据质量验证规则，确保数据的可靠性、准确性和一致性，为数据驱动的决策提供坚实保障。
更多推荐
26
0- 0
已为社区贡献2条内容
所有评论(0)