Great Expectations终极指南：用数据契约告别“脏数据“噩梦-创锋一号

Great Expectations终极指南：用数据契约告别"脏数据"噩梦

【免费下载链接】great_expectationsAlways know what to expect from your data.项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations

想象一下，你刚部署了一个机器学习模型，结果因为数据质量问题导致预测完全偏离现实。或者你的数据分析报表因为缺失值太多而无法生成。数据质量问题就像幽灵一样潜伏在每个数据项目中，随时可能爆发。今天，我将向你展示如何用Great Expectations这个强大的数据验证框架，为你的数据项目建立坚不可摧的质量防线。

Great Expectations是一个开源的数据质量验证框架，它让你能够像编写测试代码一样定义数据规则，自动验证数据质量，并生成专业的数据文档。简单来说，它就是数据的"质量检查员"，确保你的数据始终符合预期。

数据质量的"隐形杀手"：为什么传统方法总是失败？

在深入技术细节之前，让我们先看看传统数据验证方法的常见痛点：

❌ 手动检查：效率低下且容易出错

# 传统方式：手动编写验证脚本 def validate_data(df): # 检查缺失值 if df['price'].isnull().any(): print("警告：价格列存在缺失值！") # 检查数据类型 if not pd.api.types.is_numeric_dtype(df['quantity']): print("错误：数量列应为数值类型！") # 检查数值范围 if (df['age'] < 0).any() or (df['age'] > 150).any(): print("异常：年龄值超出合理范围！") # ... 更多手动检查

这种方法的问题显而易见：代码冗长、难以维护、缺乏统一标准，而且每次数据源变化都需要重新编写验证逻辑。

❌ 临时解决方案：缺乏系统性和可重复性

大多数团队采用临时脚本或手动检查，导致：

验证规则分散在不同脚本中
没有统一的验证标准
结果难以追踪和审计
无法自动化集成到CI/CD流程

Great Expectations：数据验证的"瑞士军刀"

核心概念：用"期望"定义数据规则

Great Expectations的核心思想很简单：声明式地定义你对数据的期望。就像你期望员工按时上班一样，你可以期望数据符合特定规则。

# Great Expectations方式：声明式验证 import great_expectations as gx # 1. 定义数据期望（就像编写测试用例） validator.expect_column_values_to_not_be_null(column="user_id") validator.expect_column_values_to_be_between( column="age", min_value=0, max_value=150 ) validator.expect_column_values_to_match_regex( column="email", regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" )

架构全景：理解Great Expectations的工作流程

Great Expectations数据验证架构：从Python API到Data Docs的完整流程

这张图展示了Great Expectations的完整架构。想象一下，数据上下文（Data Context）就像数据验证项目的"大脑"，它管理着所有的配置、期望套件和验证结果。期望套件（Expectation Suite）是你定义的数据规则集合，而数据文档（Data Docs）则是自动生成的验证报告。

实战演练：从零构建电商数据质量监控系统

场景设定：电商订单数据验证

假设我们有一个电商平台，每天处理数万笔订单。我们需要确保：

订单金额没有负数
用户邮箱格式正确
订单状态在有效范围内
发货时间晚于下单时间

步骤1：快速安装与项目初始化

# 创建项目目录并安装Great Expectations mkdir ecommerce-data-quality cd ecommerce-data-quality python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows pip install great-expectations pandas

步骤2：创建数据上下文

import great_expectations as gx from great_expectations.data_context import FileDataContext # 创建数据上下文 - 这是所有验证操作的"指挥中心" context = FileDataContext.create(project_root_dir="./gx_project") print("✅ 项目结构已创建！") print("📁 查看生成的文件：") print(" - great_expectations.yml # 主配置文件") print(" - expectations/ # 期望套件存储") print(" - checkpoints/ # 检查点配置") print(" - data_docs/ # 自动生成的文档")

步骤3：连接真实数据源

让我们连接一个模拟的电商订单CSV文件：

import pandas as pd from datetime import datetime # 创建模拟订单数据 orders_data = { 'order_id': [1001, 1002, 1003, 1004, 1005], 'user_email': ['alice@example.com', 'bob@example.com', 'invalid-email', 'charlie@example.com', 'diana@example.com'], 'order_amount': [99.99, 150.50, -10.00, 75.25, 200.00], 'order_status': ['pending', 'shipped', 'pending', 'cancelled', 'delivered'], 'order_date': ['2024-01-01', '2024-01-01', '2024-01-02', '2024-01-02', '2024-01-03'], 'ship_date': ['2024-01-02', '2024-01-03', '2024-01-04', None, '2024-01-05'] } df = pd.DataFrame(orders_data) df.to_csv('./data/orders.csv', index=False) # 配置数据源 datasource_config = { "name": "ecommerce_orders", "class_name": "Datasource", "execution_engine": { "class_name": "PandasExecutionEngine" }, "data_connectors": { "default_connector": { "class_name": "InferredAssetFilesystemDataConnector", "base_directory": "./data", "default_regex": { "pattern": "(.*)\\.csv", "group_names": ["data_asset_name"] } } } } context.add_datasource(**datasource_config)

步骤4：定义智能数据期望

现在，让我们创建一些真正有用的数据验证规则：

# 创建期望套件 suite_name = "ecommerce_orders_suite" context.create_expectation_suite( expectation_suite_name=suite_name, overwrite_existing=True ) # 获取验证器 batch_request = context.get_batch_request( datasource_name="ecommerce_orders", data_connector_name="default_connector", data_asset_name="orders" ) validator = context.get_validator( batch_request=batch_request, expectation_suite_name=suite_name ) print("🔍 开始定义数据期望规则...") # 1. 基础验证：确保关键字段存在且格式正确 validator.expect_table_columns_to_match_ordered_list( column_list=["order_id", "user_email", "order_amount", "order_status", "order_date", "ship_date"] ) # 2. 业务规则验证：订单金额必须为正数 validator.expect_column_values_to_be_between( column="order_amount", min_value=0, max_value=None, strict_min=True ) # 3. 数据质量验证：邮箱格式必须正确 validator.expect_column_values_to_match_regex( column="user_email", regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", mostly=0.8 # 允许20%的数据不符合规则 ) # 4. 枚举值验证：订单状态必须在预定义列表中 validator.expect_column_values_to_be_in_set( column="order_status", value_set=["pending", "shipped", "delivered", "cancelled"] ) # 5. 时间逻辑验证：发货日期必须在订单日期之后 validator.expect_column_pair_values_A_to_be_greater_than_B( column_A="ship_date", column_B="order_date", or_equal=True, ignore_row_if="either_value_is_missing" ) # 6. 唯一性验证：订单ID必须唯一 validator.expect_column_values_to_be_unique(column="order_id") # 保存期望套件 validator.save_expectation_suite(discard_failed_expectations=False) print("✅ 数据期望规则已保存！")

步骤5：创建自动化检查点

检查点就像是数据验证的"定时任务"，可以自动执行验证并触发后续操作：

# 创建检查点配置 checkpoint_config = { "name": "daily_order_validation", "config_version": 1.0, "class_name": "SimpleCheckpoint", "expectation_suite_name": suite_name, "batch_request": batch_request, "action_list": [ { "name": "store_validation_result", "action": {"class_name": "StoreValidationResultAction"} }, { "name": "update_data_docs", "action": {"class_name": "UpdateDataDocsAction"} }, { "name": "send_slack_notification", "action": { "class_name": "SlackNotificationAction", "slack_webhook": "${SLACK_WEBHOOK}", # 从环境变量读取 "notify_on": "all" # 所有结果都通知 } } ], "evaluation_parameters": { "today": datetime.now().strftime("%Y-%m-%d") }, "runtime_configuration": { "result_format": { "result_format": "COMPLETE", "unexpected_index_column_names": ["order_id"] } } } # 添加检查点 context.add_checkpoint(**checkpoint_config) # 运行验证 results = context.run_checkpoint(checkpoint_name="daily_order_validation") if results["success"]: print("🎉 数据验证通过！所有订单数据符合预期。") else: print("⚠️ 数据验证发现问题，请查看详细报告。") print(f"失败期望数量: {results['run_results'][list(results['run_results'].keys())[0]]['validation_result']['statistics']['unsuccessful_expectations']}")

步骤6：查看可视化验证报告

Great Expectations最强大的功能之一就是自动生成的数据文档。让我们生成并查看验证报告：

# 生成数据文档 context.build_data_docs() print("📊 数据文档已生成！") print("📍 打开以下文件查看详细报告：") print(" file://" + os.path.abspath("./gx_project/uncommitted/data_docs/local_site/index.html"))

Great Expectations数据验证结果可视化：展示验证规则的执行结果和统计数据

高级功能：让数据验证更智能

1. 使用数据助手自动生成期望

Great Expectations的数据助手功能可以分析你的数据并自动生成验证规则：

from great_expectations.rule_based_profiler.data_assistant import OnboardingDataAssistant # 使用数据助手自动分析数据 data_assistant_result = context.assistants.onboarding.run( batch_request=batch_request, exclude_column_names=["order_id"] # 排除不需要分析的列 ) # 获取自动生成的期望套件 expectation_suite = data_assistant_result.get_expectation_suite( expectation_suite_name="auto_generated_suite" ) # 保存自动生成的规则 context.save_expectation_suite(expectation_suite)

2. 条件期望：更精细的验证逻辑

有时候，不同的数据子集需要不同的验证规则。条件期望让你能够实现这种精细控制：

# 条件期望：不同订单状态有不同的验证规则 validator.expect_column_values_to_be_between( column="order_amount", min_value=0, max_value=1000, condition_parser="great_expectations__experimental__", row_condition='order_status=="pending"', condition_name="pending_orders_amount_check" ) validator.expect_column_values_to_be_between( column="order_amount", min_value=0, max_value=5000, condition_parser="great_expectations__experimental__", row_condition='order_status=="delivered"', condition_name="delivered_orders_amount_check" )

3. 自定义期望：满足特殊业务需求

如果内置期望不能满足你的需求，可以创建自定义期望：

from great_expectations.expectations.expectation import Expectation from great_expectations.expectations.registry import register_expectation class ExpectOrderAmountToBeMultipleOf(Expectation): """自定义期望：验证订单金额是特定倍数的整数""" def validate_configuration(self, configuration): pass @classmethod def _validate(cls, configuration, runtime_configuration=None, execution_engine=None): # 实现验证逻辑 column = configuration.kwargs.get("column") multiple_of = configuration.kwargs.get("multiple_of") # 获取数据并验证 batch_data = execution_engine.get_batch_data() column_values = batch_data[column] # 检查是否为指定倍数的整数 success = all(value % multiple_of == 0 for value in column_values) return { "success": success, "result": {"observed_value": f"所有值都是{multiple_of}的倍数"} } # 注册自定义期望 register_expectation(ExpectOrderAmountToBeMultipleOf) # 使用自定义期望 validator.expect_order_amount_to_be_multiple_of( column="order_amount", multiple_of=5 )

集成到现有工作流

与Airflow集成：自动化数据质量检查

from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta import great_expectations as gx def validate_ecommerce_data(): """Airflow任务：每日验证电商数据""" context = gx.get_context() # 运行检查点 results = context.run_checkpoint( checkpoint_name="daily_order_validation" ) if not results["success"]: # 发送告警 send_alert_email(results) raise ValueError("数据验证失败，请检查数据质量问题") return "数据验证通过" default_args = { 'owner': 'data_team', 'depends_on_past': False, 'start_date': datetime(2024, 1, 1), 'email_on_failure': True, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'ecommerce_data_validation', default_args=default_args, description='每日电商数据质量验证', schedule_interval='0 2 * * *', # 每天凌晨2点运行 catchup=False ) validate_task = PythonOperator( task_id='validate_ecommerce_data', python_callable=validate_ecommerce_data, dag=dag, )

与CI/CD管道集成：代码合并前的数据验证

# .github/workflows/data-validation.yml name: Data Quality Validation on: pull_request: paths: - 'data/**' - 'expectations/**' jobs: validate-data: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.9' - name: Install dependencies run: | pip install great-expectations pandas - name: Run data validation run: | python -c " import great_expectations as gx context = gx.get_context() # 验证所有期望套件 for suite_name in context.list_expectation_suite_names(): print(f'验证期望套件: {suite_name}') results = context.run_checkpoint( checkpoint_name=f'{suite_name}_checkpoint' ) if not results['success']: print(f'❌ {suite_name} 验证失败') exit(1) print('✅ 所有数据验证通过') "

常见问题与解决方案

问题1：安装依赖冲突

# 解决方案：使用虚拟环境和精确版本 python -m venv gx_env source gx_env/bin/activate pip install "great-expectations==0.18.0" pandas==1.5.3

问题2：大型数据集验证缓慢

# 解决方案：优化验证配置 checkpoint_config = { "runtime_configuration": { "result_format": { "result_format": "SUMMARY", # 使用摘要格式而不是完整结果 "partial_unexpected_count": 10 # 只显示前10个异常值 }, "batch_spec": { "sampling_method": "random", "sampling_kwargs": {"n": 1000} # 随机采样1000行 } } }

问题3：期望套件管理混乱

# 解决方案：使用命名规范和版本控制 # 期望套件命名规范：{数据源}_{表名}_{版本}.json # 例如：ecommerce_orders_v1.json # 定期清理旧版本 import os import json from datetime import datetime, timedelta def cleanup_old_expectation_suites(context, days_to_keep=30): """清理30天前的期望套件""" cutoff_date = datetime.now() - timedelta(days=days_to_keep) for suite_name in context.list_expectation_suite_names(): suite = context.get_expectation_suite(suite_name) created_date = suite.meta.get("created_date") if created_date and datetime.fromisoformat(created_date) < cutoff_date: context.delete_expectation_suite(suite_name) print(f"已删除过期期望套件: {suite_name}")

下一步行动建议

1. 立即开始的小步骤

选择一个关键数据表开始验证
定义3-5个最重要的数据规则
设置每日自动验证
分享验证报告给团队

2. 进阶学习路径

探索更多期望类型：查看官方文档中的完整期望列表
学习规则式分析器：自动生成数据质量规则
集成到数据流水线：与Airflow、Prefect等工具结合
创建自定义期望：满足特定业务需求

3. 最佳实践清单

✅ 从简单规则开始，逐步增加复杂度
✅ 为每个期望添加有意义的描述
✅ 定期审查和更新期望规则
✅ 将验证结果纳入数据治理流程
✅ 建立数据质量SLA（服务水平协议）

总结：数据质量的新范式

Great Expectations不仅仅是一个工具，它代表了一种数据质量即代码的新范式。通过将数据验证规则代码化、版本化、自动化，你可以：

提前发现问题：在数据问题影响业务之前发现并解决
建立信任：通过透明的验证报告建立数据信任
提高效率：自动化验证流程，释放数据工程师的时间
促进协作：统一的数据质量标准促进团队协作

记住，数据质量不是一次性任务，而是一个持续的过程。从今天开始，用Great Expectations为你的数据项目建立坚实的质量基础。

Great Expectations检查点工作流程：从数据源到验证结果的完整自动化流程

这张流程图展示了Great Expectations的完整验证流程。从数据源开始，经过期望套件验证，最终生成可视化报告和触发后续操作。这不仅是技术实现，更是数据质量管理的最佳实践。

开始你的数据质量之旅吧！选择一个数据集，定义几个简单的期望，然后看看Great Expectations如何帮助你发现那些隐藏的数据问题。你的数据团队会感谢你的。

【免费下载链接】great_expectationsAlways know what to expect from your data.项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析