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(columnuser_id) validator.expect_column_values_to_be_between( columnage, min_value0, max_value150 ) validator.expect_column_values_to_match_regex( columnemail, regexr^[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: [aliceexample.com, bobexample.com, invalid-email, charlieexample.com, dianaexample.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, indexFalse) # 配置数据源 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_namesuite_name, overwrite_existingTrue ) # 获取验证器 batch_request context.get_batch_request( datasource_nameecommerce_orders, data_connector_namedefault_connector, data_asset_nameorders ) validator context.get_validator( batch_requestbatch_request, expectation_suite_namesuite_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( columnorder_amount, min_value0, max_valueNone, strict_minTrue ) # 3. 数据质量验证邮箱格式必须正确 validator.expect_column_values_to_match_regex( columnuser_email, regexr^[a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,}$, mostly0.8 # 允许20%的数据不符合规则 ) # 4. 枚举值验证订单状态必须在预定义列表中 validator.expect_column_values_to_be_in_set( columnorder_status, value_set[pending, shipped, delivered, cancelled] ) # 5. 时间逻辑验证发货日期必须在订单日期之后 validator.expect_column_pair_values_A_to_be_greater_than_B( column_Aship_date, column_Border_date, or_equalTrue, ignore_row_ifeither_value_is_missing ) # 6. 唯一性验证订单ID必须唯一 validator.expect_column_values_to_be_unique(columnorder_id) # 保存期望套件 validator.save_expectation_suite(discard_failed_expectationsFalse) 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_namedaily_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_requestbatch_request, exclude_column_names[order_id] # 排除不需要分析的列 ) # 获取自动生成的期望套件 expectation_suite data_assistant_result.get_expectation_suite( expectation_suite_nameauto_generated_suite ) # 保存自动生成的规则 context.save_expectation_suite(expectation_suite)2. 条件期望更精细的验证逻辑有时候不同的数据子集需要不同的验证规则。条件期望让你能够实现这种精细控制# 条件期望不同订单状态有不同的验证规则 validator.expect_column_values_to_be_between( columnorder_amount, min_value0, max_value1000, condition_parsergreat_expectations__experimental__, row_conditionorder_statuspending, condition_namepending_orders_amount_check ) validator.expect_column_values_to_be_between( columnorder_amount, min_value0, max_value5000, condition_parsergreat_expectations__experimental__, row_conditionorder_statusdelivered, condition_namedelivered_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_configurationNone, execution_engineNone): # 实现验证逻辑 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( columnorder_amount, multiple_of5 )集成到现有工作流与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_namedaily_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(minutes5), } dag DAG( ecommerce_data_validation, default_argsdefault_args, description每日电商数据质量验证, schedule_interval0 2 * * *, # 每天凌晨2点运行 catchupFalse ) validate_task PythonOperator( task_idvalidate_ecommerce_data, python_callablevalidate_ecommerce_data, dagdag, )与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/checkoutv3 - name: Set up Python uses: actions/setup-pythonv4 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_namef{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-expectations0.18.0 pandas1.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_keep30): 清理30天前的期望套件 cutoff_date datetime.now() - timedelta(daysdays_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），仅供参考