当前位置：首页 > news >正文

Great Expectations数据验证终极指南：从零构建可靠数据管道

news 2026/5/24 12:46:13

Great Expectations数据验证终极指南：从零构建可靠数据管道

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

还在为数据质量问题头疼吗？每天面对脏数据、缺失值和格式错误的困扰？Great Expectations（数据期望）正是解决这些痛点的终极工具！这个开源数据验证框架让你用声明式语法定义数据规则，自动生成测试报告，彻底告别手动编写检查脚本的繁琐工作。作为数据工程师和分析师，掌握Great Expectations能让你在5分钟内建立专业的数据质量保障体系。

为什么你需要数据验证框架？

数据质量问题往往在最后一刻才被发现，导致模型训练失败、报表错误和业务决策失误。传统的数据检查方法存在三个致命缺陷：

手工脚本难以维护：每个项目都要重写验证逻辑
缺乏统一标准：团队间验证规则不一致
结果不可追溯：无法审计历史数据质量

Great Expectations通过"期望"（Expectations）这一核心概念，让你像编写单元测试一样定义数据质量规则。当你的数据满足这些"期望"时，你就能真正信任它的质量。

Great Expectations数据验证流程图展示完整的数据质量保障体系

核心概念：理解Great Expectations的架构

在深入实战之前，让我们先了解几个关键概念，这能帮助你更好地使用这个强大的框架。

数据上下文（Data Context）

数据上下文是Great Expectations的"大脑"，管理所有配置、期望套件和验证结果。它分为三种类型：

临时上下文：适合快速实验和一次性验证
文件上下文：适合本地开发和团队协作
云上下文：适合企业级部署和集中管理

数据上下文作为Great Expectations的核心组件，统一管理所有配置和元数据

期望套件（Expectation Suite）

期望套件是一组数据质量规则的集合，就像为你的数据制定的"质量标准手册"。每个期望都定义了数据应该满足的条件，例如：

列值必须在特定范围内
列不能包含空值
列值必须符合特定格式

检查点（Checkpoint）

检查点是可重复执行的验证工作流，它定义了：

要验证哪些数据
使用哪个期望套件
验证后执行什么操作（如发送通知、生成报告）

快速开始：5分钟搭建数据验证流程

步骤1：环境安装与配置

首先创建虚拟环境并安装Great Expectations：

# 创建虚拟环境 python -m venv gx_env source gx_env/bin/activate # Linux/Mac # 或 gx_env\Scripts\activate # Windows # 安装基础包 pip install great-expectations # 如果需要特定数据源支持 pip install "great-expectations[pandas,sqlalchemy]"

步骤2：初始化数据上下文

在你的项目目录中初始化Great Expectations：

import great_expectations as gx from great_expectations.data_context import FileDataContext # 创建本地文件型数据上下文 context = FileDataContext.create(project_root_dir="./my_data_project")

这会在当前目录创建以下结构：

my_data_project/ ├── great_expectations.yml # 主配置文件 ├── expectations/ # 期望套件存储目录 ├── validations/ # 验证结果存储目录 ├── data_docs/ # 自动生成的数据文档 └── checkpoints/ # 检查点配置目录

步骤3：连接你的第一个数据源

让我们以CSV文件为例，连接到你的数据：

# 配置CSV数据源 datasource_config = { "name": "sales_data", "class_name": "Datasource", "execution_engine": { "class_name": "PandasExecutionEngine" }, "data_connectors": { "csv_connector": { "class_name": "InferredAssetFilesystemDataConnector", "base_directory": "./data", # 数据文件目录 "default_regex": { "pattern": "(.*)\\.csv", "group_names": ["data_asset_name"] } } } } context.add_datasource(**datasource_config)

实战演练：为电商数据定义质量规则

假设你有一份电商订单数据，包含以下字段：order_id、customer_id、amount、order_date、status。让我们为它创建质量规则。

创建期望套件

# 获取数据批次请求 batch_request = context.get_batch_request( datasource_name="sales_data", data_connector_name="csv_connector", data_asset_name="orders" # orders.csv文件 ) # 创建期望套件 expectation_suite_name = "orders_quality_suite" context.create_expectation_suite( expectation_suite_name=expectation_suite_name, overwrite_existing=True ) # 获取验证器 validator = context.get_validator( batch_request=batch_request, expectation_suite_name=expectation_suite_name )

定义数据质量期望

现在开始定义具体的质量规则：

# 1. 检查必需列是否存在 validator.expect_table_columns_to_match_ordered_list( column_list=["order_id", "customer_id", "amount", "order_date", "status"] ) # 2. 订单ID必须是唯一的 validator.expect_column_values_to_be_unique(column="order_id") # 3. 金额必须为正数 validator.expect_column_values_to_be_between( column="amount", min_value=0.01, max_value=1000000, mostly=0.99 # 允许1%的异常值 ) # 4. 订单状态必须是有效值 validator.expect_column_values_to_be_in_set( column="status", value_set=["pending", "processing", "shipped", "delivered", "cancelled"] ) # 5. 订单日期必须在合理范围内 validator.expect_column_values_to_be_between( column="order_date", min_value="2020-01-01", max_value="2025-12-31" ) # 保存期望套件 validator.save_expectation_suite(discard_failed_expectations=False)

使用交互式助手快速创建规则

Great Expectations提供了强大的交互式功能，帮助你快速定义期望：

使用自动补全功能快速选择期望类型，提升规则定义效率

在Notebook环境中交互式创建和测试期望，实时查看验证结果

执行验证并查看结果

配置检查点

检查点将验证流程自动化：

checkpoint_config = { "name": "daily_orders_checkpoint", "config_version": 1.0, "class_name": "SimpleCheckpoint", "expectation_suite_name": expectation_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_email_on_failure", "action": { "class_name": "EmailAction", "smtp_address": "smtp.yourcompany.com", "smtp_port": 587, "sender_login": "alerts@yourcompany.com", "sender_password": "your_password", "receiver_emails": ["data-team@yourcompany.com"], "subject": "数据验证失败：订单数据" } } ] } context.add_checkpoint(**checkpoint_config)

运行验证

# 执行验证 result = context.run_checkpoint(checkpoint_name="daily_orders_checkpoint") # 检查结果 if result["success"]: print("✅ 数据验证通过！所有期望都满足。") else: print("❌ 数据验证失败，请查看详细报告。") print(f"失败期望数量：{result['run_results']['validation_result']['statistics']['unsuccessful_expectations']}")

查看可视化报告

Great Expectations会自动生成美观的数据文档：

# 生成HTML报告 context.build_data_docs() # 在浏览器中打开报告 context.open_data_docs()

Great Expectations生成的数据文档界面，清晰展示每个字段的验证结果

高级技巧：让数据验证更智能

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

手动定义所有期望很耗时？试试数据助手（Data Assistant）：

# 创建数据助手 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.build_data_docs()

数据助手会自动分析你的数据分布，并推荐合适的期望规则，大大节省了配置时间。

监控数据质量趋势

Great Expectations不仅能检查单次数据质量，还能监控长期趋势：

数据助手生成的质量趋势图，帮助识别数据分布变化和异常模式

理解批处理生命周期

要真正掌握Great Expectations，你需要理解它的批处理机制：

Great Expectations的批处理生命周期，展示数据从请求到验证的完整流程

集成到现有工作流

与Airflow集成

将Great Expectations检查点集成到Airflow DAG中：

from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime import great_expectations as gx def run_gx_checkpoint(): context = gx.get_context() context.run_checkpoint(checkpoint_name="daily_orders_checkpoint") default_args = { "owner": "data_team", "start_date": datetime(2023, 1, 1), "retries": 1 } with DAG( "data_quality_monitoring", default_args=default_args, schedule_interval="@daily", catchup=False ) as dag: validate_task = PythonOperator( task_id="validate_data_quality", python_callable=run_gx_checkpoint )

与CI/CD流水线集成

在代码合并前自动运行数据验证：

# .github/workflows/data-quality.yml name: Data Quality Check on: pull_request: branches: [main] jobs: ># 创建干净的虚拟环境 python -m venv fresh_gx_env source fresh_gx_env/bin/activate # 安装指定版本的依赖 pip install "great-expectations==0.18.0" pip install pandas==1.5.3

问题2：大数据集性能问题

症状：验证过程非常缓慢，内存占用高

解决方案：

# 1. 使用采样 batch_request = { "datasource_name": "sales_data", "data_connector_name": "csv_connector", "data_asset_name": "orders", "limit": 10000 # 限制处理行数 } # 2. 使用Spark执行引擎处理大数据 datasource_config = { "name": "big_data_source", "class_name": "Datasource", "execution_engine": { "class_name": "SparkDFExecutionEngine" }, # ... 其他配置 }

问题3：自定义期望开发

当内置期望不满足需求时，可以创建自定义期望：

from great_expectations.expectations.expectation import ColumnExpectation class ExpectColumnValuesToBePositiveAndEven(ColumnExpectation): """自定义期望：列值必须为正偶数""" def validate_configuration(self, configuration): super().validate_configuration(configuration) return True def _validate(self, configuration, **kwargs): column_values = configuration["column_values"] success = all(value > 0 and value % 2 == 0 for value in column_values) return { "success": success, "result": {"observed_value": f"{sum(1 for v in column_values if v > 0 and v % 2 == 0)}/{len(column_values)}"} }