跳至内容

Great Expectations 教程：使用 Python 验证数据

通过本端到端教程，学习如何在 Python 中使用 Great Expectations 验证您的数据！
更新 2026年7月22日  · 8分钟 读

用 AI 探索

在 ChatGPT 中打开在 Claude 中打开在 Perplexity 中打开

数据质量和一致性就像房屋的地基——没有坚实的基础，其上构建的一切都可能坍塌。这正是数据验证发挥重要作用的地方。数据验证帮助您确保数据准确、一致且可靠。

Great Expectations 是一款开源的数据验证工具，能够及早发现数据问题，并确保您的数据达到所需的质量标准。

在本指南中，我们将通过一个实用的端到端示例，带您使用 Great Expectations 进行数据验证，帮助您快速上手！

什么是 Great Expectations？

Great Expectations（GX）是一套开源框架，已在现代数据管道中用于管理与自动化数据验证而广受欢迎。

其基于 Python 的框架旨在帮助数据团队保证数据的质量与一致性。用户可以定义“期望”（expectations）——即描述有效数据应具有何种特征的规则或测试——以自动验证数据是否符合这些标准。

Great Expectations 的一些优势包括：

  • 自动化数据验证 —— Great Expectations 将数据验证流程自动化，减少人工操作并将错误风险降至最低，确保数据持续满足预定义标准。
  • 与数据管道集成 —— 它可轻松集成多种数据源与平台，包括 SQL 数据库s、云存储，以及ETL 工具，从而在管道的不同阶段执行数据验证。
  • 清晰、可执行的验证结果 —— 该工具提供透明的验证结果，便于快速发现并解决数据质量问题。
  • 数据文档化 —— Great Expectations 可生成详尽、易用的数据验证文档，帮助团队在质量标准上达成一致，并为后续使用提供参考。
  • 可扩展性与灵活性 —— 作为开源工具，Great Expectations 具有高度可定制性，并可随您的数据验证需求扩展，灵活适配多种用例而无需高昂成本。

现在，让我们看一个端到端示例！

设置 Great Expectations

在本教程中，您将学习如何使用 GX Core（Great Expectations 的开源版本）来验证一个 Pandas DataFrame。我们将演示如何设置上下文、注册 Pandas 数据源、定义期望以及验证数据批次。

注意： 我们建议您配合DataLab 笔记本一起操作，您也可以自行创建 Python 脚本。

先决条件

  • 已安装 Python 3.9 至 3.12。
  • 为避免冲突，强烈建议在虚拟环境中安装 Great Expectations（声明： 虚拟环境的设置不在本文讨论范围内）。
  • 一个示例数据集。

注意： 如果使用提供的DataLab 笔记本，这些先决条件已满足，您可直接跳过。

使用以下命令通过 pip 安装 GX：

pip install great_expectations

该命令会安装核心包及所有必要的依赖。

Great Expectations 需要数据上下文来管理配置。我们使用临时（ephemeral）数据上下文以避免持久化配置。

import great_expectations as gx

# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

创建您的首个数据验证套件

GX 已设置完成，接下来创建一个数据验证套件。

数据源用于将 Great Expectations 连接到您的数据，而数据资产则表示特定的数据子集（例如表、DataFrame 或文件）。

本例中，我们将准备连接到名为 inventory_parts_df 的 DataFrame。示例数据集在提供的 DataLab 中可用，运行 SQL 代码块后即可创建：

如果您不使用 DataLab，请自行创建带有示例数据的 DataFrame。

现在，创建您的数据源和数据资产：

# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

批次定义用于标识并组织待验证的数据。此处我们添加一个覆盖整个 DataFrame 的批次定义：

# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

批次是与批次定义关联的一组数据。要进行验证，您需要检索批次并将其与您的 DataFrame（本例为 inventory_parts_df）关联：

# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

期望是用于验证数据的规则。在本例中，我们将定义以下简单期望：

  1. 确保 inventory_id 值非空。
  2. 确保 part_num 值唯一。
# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)

您可以在Expectation Gallery中查看所有可用的期望。我们鼓励您再添加一些！

定义期望后，GX 会输出期望套件配置：

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

该套件包含以下信息：

  1. 套件名称与 ID： 用于跟踪和管理套件的唯一名称（inventory_parts_suite）与标识符。
  2. 期望： 每条规则包含：
    • 检查类型（例如确保某列没有空值或具有唯一条目）。
    • 参数，如被验证的列名。
    • 元数据与每个期望的唯一 ID，便于跟踪与自定义。
  3. 元数据： Great Expectations 的版本信息，以确保工具兼容性。
  4. 备注： 可用于添加关于套件的描述性注释（可选）。

这种结构化输出既是文档，也是可复用的配置，用于验证您的数据集，使期望定义清晰、可追溯，并随时可用于将来。

5. 验证数据

最后，将批次与已定义的期望进行验证，并评估结果。

# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)

运行验证后，Great Expectations 会提供一份详细报告，说明数据集是否满足已定义的期望：

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

该报告详述了您的数据质量，突出展示了通过与未通过之处。以下是结果的简要说明：

整体验证：验证结果部分成功：50% 的期望通过，50% 失败。失败的期望说明存在需要关注的数据质量问题。本例中有一列未满足定义的规则。

期望 1：inventory_id 不应有缺失值

  • 结果：通过
  • 说明：inventory_id 列中的每个值均存在，没有空值或缺失项，表明该列具有良好的数据完整性。

期望 2：part_num 应具有唯一值

  • 结果：未通过
  • 说明： part_num 列包含 97.98% 的重复值，仅有少数值是唯一的。
  • 要点：
    • 示例重复值包括“3069b”和“33291”。
    • 该工具还展示了这些重复值的出现频次与行位置，便于定位并修复问题。

当然，这只是一个示例数据集，我们特意包含了一个通过与一个未通过的期望，便于您了解两种验证结果。

就是这样！您已成功完成端到端的数据验证。

将 Great Expectations 集成到数据管道

在生产环境中，验证必须直接嵌入工作流，以在各个阶段持续监控数据质量。

本节将讨论如何将 Great Expectations 集成到您的数据管道中。

以下示例旨在提供思路，可能需要本文未包含的额外配置。请查看各工具文档以获取最新语法！

与 ETL 工具集成

将 Great Expectations 与常见的 ETL 工具（如 Apache Airflow 或 Prefect）集成相对简单。将验证步骤直接嵌入 ETL 流程，可在影响下游分析之前实时捕获并处理数据问题。

下面演示一个将 Great Expectations 集成到 Prefect 的简单示例，以在自动化 ETL 工作流中运行数据验证：

from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # Check validation results and raise an alert if validation fails
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute the flow
flow.run()

在该示例中，我们定义了一个包含 Great Expectations 验证任务的 Prefect 流。

validate_data() 任务会加载 Great Expectations 上下文、获取数据批次并应用期望套件。

若数据未满足验证标准，该任务将抛出告警，停止工作流，防止下游错误。

持续数据验证

您可以使用多种工具调度验证作业，例如基于 Unix 的系统中的 cron 作业，或 Apache Airflow 等托管服务。本例将展示如何使用 Airflow 调度验证运行，它非常适合编排数据管道。

以下展示如何设置一个 Airflow DAG（有向无环图）以每日运行 Great Expectations 验证：

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Runs once a day
)
# Define the function to run the validation
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Set up the task in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Set the task in the DAG
validation_task

在该示例中，我们定义了一个 DAG，按天调度运行一次（@daily）。

run_validation() 函数通过加载 Great Expectations 上下文并针对数据运行已定义的期望套件来执行验证。

使用 Great Expectations 进行数据验证的最佳实践

为实现可扩展与高效，遵循最佳实践始终是明智之举，数据验证亦然。

从小处着手，循序迭代

先从基础的数据质量检查开始，并逐步扩展。起初聚焦于基本期望更好，可避免过度复杂化流程，从而更顺利地集成并便于排障。随着您对数据集的理解加深，可添加更复杂的验证。

跨团队协作

数据质量不只是技术问题。与业务团队协作定义期望，确保实施的验证与底层业务逻辑和目标保持一致。此类跨职能协作能确保数据服务于既定目的，并满足各方需求。

尽可能自动化

尽可能将数据验证自动化并集成到数据管道中。自动化的验证检查可在无需人工干预的情况下持续监控数据质量，显著提升效率。

结语

做得很好！您已经学会了如何在 Great Expectations 中配置与验证数据。这些技巧将帮助您在工作流中保持高数据质量与透明度。

若想继续提升技能，请查看以下资源：

FAQs

Great Expectations 与其他数据验证工具相比如何？

Great Expectations 是开源、灵活的，并且能很好地与现代数据管道集成。它以丰富的期望库与强大的文档而脱颖而出。

使用 Great Expectations 需要懂 Python 吗？

具备基础的 Python 知识会有所帮助，但 Great Expectations 提供了用户友好的 CLI 和完善的文档，使非编程人员也能上手。

Great Expectations 支持哪些类型的数据源？

Great Expectations 支持广泛的数据源，包括：

  • 关系型数据库，如 PostgreSQL、MySQL 与 SQL Server。
  • 云存储方案，如 AWS S3、Google Cloud Storage 与 Azure Blob Storage。
  • 文件格式，如 CSV、Parquet 与 Excel。
  • 大数据框架，如 Apache Spark 与 Databricks。您可以通过为数据源配置相应参数，轻松将 Great Expectations 连接到这些来源。

我可以将 Great Expectations 用于流式数据吗？

Great Expectations 主要用于批处理数据验证。尽管其本身不原生支持流式数据管道，您可以通过定期验证数据快照或微批次，将其集成到 Apache Kafka 或 Spark Structured Streaming 等框架中。

能否对期望与验证结果进行版本控制？

可以。您可以将期望与配置以 YAML 或 JSON 文件形式存储在 Git 仓库中以进行版本控制。对于验证结果，您可以设置数据库或基于文件的存储来跟踪随时间的变化，并将其集成到 CI/CD 流水线中以实现持续监控。

Great Expectations 如何应对数据集的架构演进？

Great Expectations 通过灵活的期望框架来处理模式演进（schema evolution）。如果架构发生变化，您可以：

  • 使用 expect_table_columns_to_match_set 或类似期望以动态验证列名。
  • 修改或创建新的期望套件以适配新架构。
  • 借助架构推断工具，自动为新增列更新期望。
主题
Python
数据工程

通过以下课程进一步学习数据工程！

Tracks

数据工程师 在 Python 中

40小时
掌握高需求技能，高效摄取、清洗、管理数据，并调度和监控管道，让你在数据工程领域脱颖而出。
查看详情Right Arrow
开始课程
查看更多Right Arrow