Python数据质量神器：Great Expectations从入门到实战教程

作者：

在

一、Great Expectations 库概述

Great Expectations 是一款专注于数据验证、数据文档化与数据质量监控的 Python 开源库，核心用于保障数据 pipeline 中数据的准确性、完整性与一致性。其原理是通过定义「数据期望」规则，自动校验数据是否符合预期，同时生成可视化数据文档。该库采用 Apache-2.0 开源协议，优点是适配多数据源、规则易编写、可集成主流数据工具，缺点是初次配置稍繁琐，轻量数据验证场景略显冗余。

二、Great Expectations 安装与基础环境配置

2.1 库的安装

在使用 Great Expectations 之前，需要通过 pip 完成安装，打开命令行执行以下命令：

pip install great-expectations

安装完成后，可以在 Python 环境中导入库验证是否安装成功，无报错则说明安装正常。

import great_expectations as ge
from great_expectations.data_context import FileDataContext

2.2 初始化项目环境

Great Expectations 采用项目化管理，需要初始化工作目录，执行命令后会自动生成配置文件夹：

great_expectations init

初始化完成后，目录结构如下：

great_expectations/
├── great_expectations.yml  # 主配置文件
├── expectations/           # 数据验证规则存储目录
├── checkpoints/            # 验证任务配置目录
├── plugins/                # 插件目录
├── uncommitted/            # 本地配置与缓存文件
└── data_docs/               # 数据文档生成目录

该目录结构用于统一管理验证规则、数据源与报告，方便团队协作与版本控制。

三、Great Expectations 核心使用流程

3.1 加载数据并创建验证对象

Great Expectations 支持 Pandas DataFrame、Spark DataFrame、SQL 数据源等多种数据格式，这里以最常用的 Pandas 数据为例，先创建测试数据，再生成验证对象。

import pandas as pd
# 创建模拟业务数据（用户订单数据）
data = {
    "user_id": [1001, 1002, 1003, 1004, 1005, None],
    "order_id": [2024001, 2024002, 2024003, 2024004, 2024005, 2024006],
    "order_amount": [99.0, 199.0, 299.0, 399.0, 499.0, 599.0],
    "pay_status": ["已支付", "未支付", "已支付", "已支付", "未支付", "已支付"],
    "create_time": ["2024-01-01", "2024-01-02", "2024-01-03", 
                    "2024-01-04", "2024-01-05", "2024-01-06"]
}
df = pd.DataFrame(data)
# 创建 Great Expectations 验证对象
ge_df = ge.from_pandas(df)

代码说明：首先构建模拟订单数据，包含用户ID、订单ID、订单金额、支付状态、创建时间字段，通过 ge.from_pandas() 将普通 DataFrame 转换为支持数据验证的 GE 对象。

3.2 编写基础数据验证规则

Great Expectations 的核心是「期望（Expectation）」，即提前定义数据应该满足的规则，以下是常用的基础验证规则：

# 1. 验证列是否存在
ge_df.expect_column_to_exist("user_id")
# 2. 验证列值不允许为空（除指定列外）
ge_df.expect_column_values_to_not_be_null("order_id")
# 3. 验证列值唯一
ge_df.expect_column_values_to_be_unique("order_id")
# 4. 验证数值范围
ge_df.expect_column_values_to_be_between("order_amount", min_value=0, max_value=1000)
# 5. 验证列值属于指定集合
ge_df.expect_column_values_to_be_in_set("pay_status", ["已支付", "未支付"])
# 6. 验证日期格式
ge_df.expect_column_values_to_match_strftime_format("create_time", "%Y-%m-%d")

代码说明：每条 expect_* 方法对应一条验证规则，覆盖列存在性、非空、唯一性、数值范围、枚举值、日期格式等高频验证场景，无需编写复杂判断逻辑。

3.3 执行验证并查看结果

编写完规则后，调用 validate() 方法执行验证，返回包含验证结果的字典，可直观查看哪些规则通过、哪些失败。

# 执行数据验证
validation_result = ge_df.validate()
# 打印整体验证结果
print("数据验证是否通过：", validation_result.success)
# 打印详细验证统计
print("验证规则总数：", validation_result.statistics["evaluated_expectations"])
print("通过规则数：", validation_result.statistics["successful_expectations"])
print("失败规则数：", validation_result.statistics["unsuccessful_expectations"])
# 查看失败规则详情
for result in validation_result.results:
    if not result.success:
        print("\n失败规则：", result.expectation_config.expectation_type)
        print("失败列：", result.expectation_config.kwargs["column"])
        print("失败原因：", result.result)

代码说明：validate() 会批量执行所有定义的规则，success 字段表示整体是否通过，statistics 提供统计信息，失败规则会返回具体列与异常数据详情，方便快速定位问题。

3.4 生成可视化数据质量报告

Great Expectations 支持自动生成可视化数据文档（Data Docs），无需手动编写报告，可在浏览器中直观查看数据质量。

# 初始化数据上下文
context = FileDataContext.create(project_root_dir="./")
# 保存验证规则
expectation_suite = ge_df.get_expectation_suite()
expectation_suite.expectation_suite_name = "order_data_validation_suite"
context.save_expectation_suite(expectation_suite, overwrite=True)
# 构建验证任务
checkpoint = context.add_or_update_checkpoint(
    name="order_data_checkpoint",
    expectation_suite_name="order_data_validation_suite",
    batch_request=context.get_batch_request_class()(
        datasource_name="my_pandas_datasource",
        data_asset_name="order_data",
    ),
)
# 运行任务并生成报告
checkpoint_result = context.run_checkpoint(checkpoint_name="order_data_checkpoint")
# 打开数据文档
context.open_data_docs()

代码说明：通过保存验证规则、创建检查点、执行验证三步，自动生成 HTML 格式的可视化报告，打开浏览器即可查看所有规则的执行情况、数据分布、异常数据明细。

四、进阶使用：结合业务场景的复杂数据验证

4.1 多条件组合验证

在实际业务中，往往需要多条件组合验证，Great Expectations 支持自定义过滤条件，实现复杂逻辑验证。

# 验证：支付状态为已支付时，订单金额必须大于0
ge_df.expect_column_values_to_be_between(
    column="order_amount",
    min_value=0.01,
    max_value=None,
    row_condition="pay_status == '已支付'"
)
# 验证：用户ID不为空时，必须为整数类型
ge_df.expect_column_values_to_be_of_type(
    column="user_id",
    type_="int64",
    row_condition="user_id IS NOT NULL"
)

代码说明：通过 row_condition 参数添加过滤条件，实现按行筛选验证，适用于业务关联字段的合规性检查。

4.2 自定义验证规则

对于特殊业务规则，内置方法无法满足时，可通过自定义函数实现专属验证逻辑。

# 自定义验证规则：订单ID必须以2024开头
def custom_order_id_check(value):
    return str(value).startswith("2024")
# 应用自定义规则
ge_df.expect_column_values_to_be_true(
    column="order_id",
    condition=custom_order_id_check,
    condition_value="value"
)

代码说明：自定义函数返回布尔值，通过 expect_column_values_to_be_true 调用，适配企业个性化数据规范。

4.3 集成 SQL 数据源验证

Great Expectations 不仅支持本地数据，还可直接连接 MySQL、PostgreSQL 等数据库，验证线上数据。

from sqlalchemy import create_engine
# 连接数据库
engine = create_engine("mysql+pymysql://用户名:密码@主机:端口/数据库名")
# 从SQL查询创建验证对象
ge_sql_df = ge.from_sql(
    sql="SELECT * FROM order_table WHERE create_time >= '2024-01-01'",
    con=engine
)
# 执行验证
ge_sql_df.expect_column_values_to_not_be_null("order_id")
sql_validation_result = ge_sql_df.validate()
print("数据库数据验证结果：", sql_validation_result.success)

代码说明：通过 SQLAlchemy 连接数据库，直接查询数据并验证，适用于数据仓库、业务数据库的实时质量监控。

五、实际项目案例：电商订单数据全流程质量监控

5.1 案例背景

某电商平台每日产生数十万订单数据，需要保障：

核心字段（订单ID、用户ID、金额）无空值；
订单金额、支付状态符合业务逻辑；
日期格式规范，数据无重复；
自动生成每日数据质量报告。

5.2 完整代码实现

import pandas as pd
import great_expectations as ge
from great_expectations.data_context import FileDataContext

# 1. 加载生产环境订单数据
# 实际场景可替换为数据库读取或文件读取
df = pd.read_csv("ecommerce_orders.csv")
ge_df = ge.from_pandas(df)

# 2. 定义全量业务验证规则
# 基础完整性验证
ge_df.expect_column_to_exist("user_id")
ge_df.expect_column_to_exist("order_id")
ge_df.expect_column_to_exist("order_amount")
ge_df.expect_column_values_to_not_be_null("order_id")
ge_df.expect_column_values_to_be_unique("order_id")

# 业务合规性验证
ge_df.expect_column_values_to_be_between("order_amount", min_value=0.01, max_value=99999)
ge_df.expect_column_values_to_be_in_set("pay_status", ["已支付", "未支付", "退款中"])
ge_df.expect_column_values_to_match_strftime_format("create_time", "%Y-%m-%d %H:%M:%S")

# 关联逻辑验证
ge_df.expect_column_values_to_be_between(
    column="order_amount",
    min_value=0.01,
    row_condition="pay_status == '已支付'"
)

# 3. 执行验证
result = ge_df.validate()

# 4. 输出验证结果
if result.success:
    print("订单数据质量合格，可进入后续分析流程")
else:
    print("订单数据存在异常，请修复后再处理")

# 5. 生成并保存数据质量报告
context = FileDataContext.create(project_root_dir="./")
suite = ge_df.get_expectation_suite()
suite.expectation_suite_name = "ecommerce_order_validation"
context.save_expectation_suite(suite, overwrite=True)

checkpoint = context.add_or_update_checkpoint(
    name="daily_order_check",
    expectation_suite_name="ecommerce_order_validation",
)
context.run_checkpoint(checkpoint_name="daily_order_check")
context.open_data_docs()

代码说明：该案例完整模拟电商订单数据的质量监控流程，从数据加载、规则定义、验证执行到报告生成，可直接集成到数据 pipeline 中，实现自动化数据校验。

六、相关资源

Pypi地址：https://pypi.org/project/great-expectations/
Github地址：https://github.com/great-expectations/great_expectations
官方文档地址：https://docs.greatexpectations.io/docs/

关注我，每天分享一个实用的Python自动化工具。

实用工具