使用 Great Expectations 在 Kedro 工作流中实现数据质量验证：Hook 与 Pipeline 双路径实战-平芜编程栈

使用 Great Expectations 在 Kedro 工作流中实现数据质量验证：Hook 与 Pipeline 双路径实战

【免费下载链接】kedroKedro is a toolbox for production-ready data science. It uses software engineering best practices to help you create data engineering and data science pipelines that are reproducible, maintainable, and modular.项目地址: https://gitcode.com/GitHub_Trending/ke/kedro

Great Expectations（GE）是一个开源的、面向生产环境的数据质量框架，支持对数据进行验证、文档化与画像分析。本指南以 Kedro 官方spaceflights-pandas项目为背景，讲解如何将 GE 以Kedro Hook（自动在数据加载/保存时校验，零侵入）和Pipeline 节点（将校验显式化为 DAG 中的质量门禁）两种方式集成进 Kedro 工作流，并覆盖 Expectation Suite 的组织方式、文件型 Data Context 的持久化方案，以及 Kedro 内置 Dataset Validation 与 GE 的取舍关系。读完本文，你将能够在自己的 Kedro 项目中落地可复用、可观测、可追溯的数据质量校验体系。

核心概念：Expectation（期望）

在 Great Expectations 中，数据校验规则被称为expectation（期望）——一条关于数据结构的、可证伪、可验证的断言。典型示例包括：

"这一列永远不应为 null"
"这一列的值应介于 0 与 100 之间"
"这一列应只包含这些指定的类别"

当你运行验证时，Great Expectations 会检查数据是否满足这些期望，并返回详细结果，明确指出哪些通过、哪些失败。全部可用期望类型的完整列表见 Great Expectations 官方的 expectations 参考文档。

小提示：如果你的需求仅仅是 catalog 数据集上的 schema 级检查，Kedro 内置的 Dataset Validation 只需一行 YAML 即可覆盖常见场景，且支持自定义 validator 类——你完全可以在它背后包装 Great Expectations 的检查逻辑。这意味着：轻量 schema 校验走内置机制，复杂业务规则校验走 GE。

前置条件与环境准备

开始之前，你需要准备：

一个可运行的 Kedro 项目。本文示例基于spaceflights-pandasstarter；如果你不熟悉 Spaceflights 项目，建议先阅读 Spaceflights 教程。
在项目中安装 Great Expectations。

按以下步骤搭建环境：

kedro new --starter=spaceflights-pandas --name spaceflights-great-expectations

进入项目目录后，在requirements.txt中追加 GE 依赖：

great-expectations>=1.8.0

安装项目依赖：

uv pip install -r requirements.txt

great-expectations>=1.8.0这一版本门槛与本指南使用的代码 API 强相关——下文将说明为什么 1.0+ 版本彻底改变了使用方式。

理解 Great Expectations（1.0+ 的 API 形态）

Great Expectations 1.0+ 引入了重大 API 变更：一切都在 Python 代码中完成，而不是通过 CLI 命令和 YAML 配置文件。理解下面六个核心组件，就理解了 GE 的工作方式。

六个关键组件

Context（上下文）：你的验证操作工作区，一切操作的入口。
```
context = gx.get_context()
```
不加参数时得到的是 ephemeral（临时）Data Context；传入context_root_dir则得到持久化的文件型 Context（详见后文"文件型 Data Context"一节）。
Data Source（数据源）：连接你的数据。在本文场景中，数据源对应内存中的 pandas DataFrame。
```
source = context.data_sources.add_or_update_pandas("my_source")
```
Data Asset（数据资产）：数据源内的一个具体数据集。
```
asset = source.add_dataframe_asset("companies")
```

Batch（批次）：待校验数据的一个具体实例（快照）。

batch_request = asset.build_batch_request(options={"dataframe": df}) batch = asset.get_batch(batch_request)

Expectation Suite（期望套件）：一组要运行的期望的集合。

suite = gx.ExpectationSuite(name="my_validation") suite.expectations = [...]

Validation Result（验证结果）：对某个 batch 运行期望后得到的结果。
```
result = batch.validate(suite) if not result.success: # 处理验证失败
```

验证工作流

DataFrame → Batch → Apply Expectations → Validation Result

当你验证数据时，Great Expectations 会：

对数据拍摄快照（即 "batch"）；
逐条对快照运行每个期望；
返回详细结果，展示哪些通过、哪些失败。

关于 Context 的选型建议：交互式开发阶段，可以使用 ephemeral GX Data Context（运行之间不持久化文件），非常适合探索和迭代式创建期望；而生产运行阶段，应持久化期望套件并使用基于文件的 Data Context，使套件、验证结果与历史可复现、可共享。

集成方式总览：Hook 还是 Pipeline 节点

本指南介绍两种在 Kedro 中使用 GE 做数据校验的方式，各有适用场景：

作为 Kedro Hook（Approach 1）：数据加载/保存时自动校验。这是低摩擦的入门集成方式，无需改动既有 pipeline 代码。
作为 Pipeline 运行的一部分（Approach 2）：在 pipeline 中显式加入校验节点，当你希望校验在 DAG 中可见、或希望建立显式的数据血缘时使用。

定义期望（Expectations）

为了保持项目整洁，建议把数据期望定义在独立的 Python 模块中。这种分离有几点关键优势：

可复用性：同一套期望可在 hooks、独立校验节点、临时测试甚至 CI 中复用；
可维护性：所有校验规则集中在一处，更易于更新与评审；
清晰性：pipeline 代码专注于业务逻辑，校验逻辑专注于数据质量；
一致性：任何校验数据集的地方都导入同一套 suite。

在本例中，我们在src/spaceflights_great_expectations/expectations.py中创建一个字典，把数据集名称映射到应用于它们的期望列表，并提供一个便捷函数get_suite()，根据字典规则构建 GE 的ExpectationSuite对象：

import great_expectations as gx EXPECTATION_SUITES = { "companies": [ gx.expectations.ExpectColumnToExist(column="company_rating"), ], "reviews": [ gx.expectations.ExpectColumnToExist(column="review_scores_rating"), ], "model_input_table": [ gx.expectations.ExpectColumnToExist(column="price"), gx.expectations.ExpectColumnValuesToNotBeNull(column="price"), ], } def get_suite(name: str) -> gx.ExpectationSuite: suite = gx.ExpectationSuite(name=f"{name}_validation") suite.expectations = EXPECTATION_SUITES[name] return suite

这些期望与 starter 中的真实数据是对应的：在仓库自带的示例数据中，companies.csv 包含company_rating列，reviews.csv 包含review_scores_rating列，而model_input_table是 data science pipeline 的建模输入，price是其目标列——对这些列做存在性与非空校验，正是数据进入建模环节前的合理质量门禁。

Approach 1：作为 Kedro Hook 自动校验

Hooks 允许你在不修改既有 pipeline 代码的前提下，让数据在 pipeline 中流转时自动被校验。Kedro Hook 是会在 pipeline 执行到特定时间点自动运行的函数，本文用到其中两个：

before_node_run：节点执行前运行（适合校验输入）；
after_node_run：节点执行后运行（适合校验输出）。

把校验逻辑放进 hook，就形成了一张"安全网"——既能捕获坏数据，又不会让 pipeline 定义变得臃肿。

在项目的src/spaceflights_great_expectations/目录下创建或编辑hooks.py：

from typing import Any from kedro.framework.hooks import hook_impl from kedro.pipeline.node import Node import great_expectations as gx import pandas as pd import logging from .expectations import get_suite, EXPECTATION_SUITES logger = logging.getLogger(__name__) class DataValidationHooks: """Validate datasets using Great Expectations.""" def __init__(self): self.context = gx.get_context() @hook_impl def before_node_run(self, node: Node, inputs: dict[str, Any]) -> None: for name, data in inputs.items(): if name in EXPECTATION_SUITES and isinstance(data, pd.DataFrame): self._validate(data, name) @hook_impl def after_node_run(self, node: Node, outputs: dict[str, Any]) -> None: for name, data in outputs.items(): if name in EXPECTATION_SUITES and isinstance(data, pd.DataFrame): self._validate(data, name) def _validate(self, df: pd.DataFrame, name: str) -> None: logger.info(f"Validating {name}...") source = self.context.data_sources.add_or_update_pandas(name) asset = source.add_dataframe_asset(name) batch_request = asset.build_batch_request(options={"dataframe": df}) batch = asset.get_batch(batch_request) suite = get_suite(name) result = batch.validate(suite) if not result.success: errors = [r.expectation_type for r in result.results if not r.success] raise ValueError( f"Validation failed for {name}:\n" + "\n".join(f" - {e}" for e in errors) ) logger.info(f"✓ {name} passed validation")

这个 Hook 实现的工作机制可概括为四点：

配置驱动：EXPECTATION_SUITES字典将数据集名映射到期望列表，是校验规则的唯一事实来源；
自动触发：每个节点运行前后，hook 都会检查其输入/输出是否需要校验；
选择性校验：只校验显式配置过的数据集，未配置的数据集零开销；
快速失败（fail-fast）：校验失败立即抛出ValueError，pipeline 在运行下游节点前停止，并给出清晰的错误信息（列出所有失败期望的类型）。

注册 Hook

在src/spaceflights_great_expectations/settings.py中注册自定义 hook：

from spaceflights_great_expectations.hooks import DataValidationHooks HOOKS = (DataValidationHooks(),)

从源码结构看，HOOKS是 Kedro 项目设置项之一，默认值为空元组，settings 加载逻辑会在项目配置阶段读取它，并通过 pluggy 插件管理器注册这些实现。@hook_impl装饰器来自 kedro/framework/hooks/markers.py，它和hook_spec一样，都是基于 Kedro 的"kedro"命名空间声明的 pluggy 标记。

运行并观察日志

保持 pipeline 不变，正常运行：

kedro run

你会看到数据校验日志与常规 Kedro 日志交织在一起：

INFO Validating reviews... hooks.py:47 Calculating Metrics: 100%|██████████████████████████| 2/2 [00:00<00:00, 3436.55it/s] INFO ✓ reviews passed validation hooks.py:67 INFO Running node: create_model_input_table_node: create_model_input_table() -> node.py:420 INFO Validating model_input_table... hooks.py:47 Calculating Metrics: 100%|██████████████████████████| 8/8 [00:00<00:00, 4960.74it/s] INFO ✓ model_input_table passed validation hooks.py:67 INFO Saving data to model_input_table (ParquetDataset)... data_catalog.py:1008 INFO Completed node: create_model_input_table_node runner.py:245 INFO Completed 6 out of 9 tasks runner.py:246 INFO Loading data from model_input_table (ParquetDataset)... data_catalog.py:1048 INFO Loading data from params:model_options (MemoryDataset)... data_catalog.py:1048 INFO Validating model_input_table... hooks.py:47 Calculating Metrics: 100%|██████████████████████████| 8/8 [00:00<00:00, 4488.89it/s] INFO ✓ model_input_table passed validation

注意model_input_table被校验了两次：一次是它作为下游节点输入被加载时（before_node_run），一次是它被创建后（after_node_run），这正是本 hook 在节点执行前后分别校验输入与输出的直接体现。

Hook 机制源码级佐证

从 Kedro 源码可以进一步印证这一流程的底层实现：runner/task.py 中的_collect_inputs_from_hook会在节点真正运行前调用hook_manager.hook.before_node_run(...)，而_call_node_run在node.run(inputs)成功完成后调用hook_manager.hook.after_node_run(...)。值得注意的是，before_node_run的返回值会被合并进节点的输入字典，即它不仅可以"检查"数据，还可以"替换"输入；而after_node_run则纯粹是事后观察点。两个 hook 的完整参数签名（含catalog、is_async、run_id等可选用参数）定义在 kedro/framework/hooks/specs.py 的NodeSpecs类中。

注意：Hooks 执行顺序与并行运行存在边界。Kedro 的 Hooks 指南明确指出：使用ParallelRunner时，catalog、context、pipeline级 hook 会在主进程中执行，但dataset与node级 hook 不会在并行 worker 进程中运行。如果你的项目依赖before_node_run/after_node_run做数据校验，应使用SequentialRunner（默认）或ThreadRunner。此外，hook 实现参数不能带默认值（pluggy 的 opt-in 参数机制会传入默认值而非实际值），这也是本文示例签名干净、无默认参数的原因。

Approach 2：作为 Pipeline 节点显式校验

另一种做法是把 GE 校验实现为 Kedro pipeline 中的显式节点。节点方式有这些优势：

可见性：校验节点会出现在 Kedro-Viz 中，质量门禁在哪里一目了然；
可控性：可以利用 tags 分组或按名称运行 pipeline 等特性轻松地运行或跳过校验；
灵活性：可以把校验放在任意阶段——预处理前、转换后、建模前等；
数据血缘：被校验的数据集显式出现在 data catalog 中。

下面以创建一个数据校验节点并把它加入data_processingpipeline 为例。

首先，在src/spaceflights_great_expectations/pipelines/data_processing/nodes.py中添加validate_datasets节点：

import pandas as pd import great_expectations as gx from spaceflights_great_expectations.expectations import ( EXPECTATION_SUITES, get_suite, ) def validate_datasets( companies: pd.DataFrame, reviews: pd.DataFrame, shuttles: pd.DataFrame ) -> None: context = gx.get_context() datasets = { "companies": companies, "reviews": reviews, "shuttles": shuttles, } for name, df in datasets.items(): if name not in EXPECTATION_SUITES: continue source = context.data_sources.add_or_update_pandas(name) asset = source.add_dataframe_asset(name) batch_request = asset.build_batch_request(options={"dataframe": df}) batch = asset.get_batch(batch_request) suite = get_suite(name) result = batch.validate(suite) if not result.success: raise gx.exceptions.ValidationError(f"Validation failed for: {name}")

接着，更新src/spaceflights_great_expectations/pipelines/data_processing/pipeline.py，把新节点编排进 pipeline：

from .nodes import create_model_input_table, preprocess_companies, preprocess_shuttles, validate_datasets def create_pipeline(**kwargs) -> Pipeline: return Pipeline( [ Node( func=validate_datasets, inputs=["companies", "reviews", "shuttles"], outputs=None, name="validade_datasets_node", ), Node( func=preprocess_companies, inputs="companies", outputs="preprocessed_companies", name="preprocess_companies_node", ), Node( func=preprocess_shuttles, inputs="shuttles", outputs="preprocessed_shuttles", name="preprocess_shuttles_node", ), Node( func=create_model_input_table, inputs=["preprocessed_shuttles", "preprocessed_companies", "reviews"], outputs="model_input_table", name="create_model_input_table_node", ), ] )

现在 pipeline 在起始位置拥有了一个显式的校验门禁：

[Load data] → validate_datasets → preprocess_companies → ...

如果校验失败，预处理节点永远不会运行，从而节省计算时间，并阻止坏数据向后续环节传播。这与默认的SequentialRunner按依赖序执行节点、失败即中止的行为一致；即使使用并行 runner，由于校验节点与下游节点存在数据依赖（companies/reviews/shuttles是下游节点的输入），调度器也会保证先校验后处理。

变体：拆分独立的校验节点

除了"一个节点校验所有数据"，也可以为每个数据集创建独立校验节点：

def validate_companies(companies: pd.DataFrame) -> pd.DataFrame: """Validate companies data and pass it through.""" # ... validation logic ... return companies # Pass through if valid def validate_reviews(reviews: pd.DataFrame) -> pd.DataFrame: """Validate reviews data and pass it through.""" # ... validation logic ... return reviews

然后在 pipeline 中编排：

Pipeline([ node( func=validate_companies, inputs="companies", outputs="validated_companies", name="validate_companies_node", ), node( func=preprocess_companies, inputs="validated_companies", # Use validated data outputs="preprocessed_companies", name="preprocess_companies_node", ), # ... ])

这种方式：

创建显式的数据血缘（companies→validated_companies）；
允许不同数据集并行校验；
更容易针对特定数据集跳过校验。

选型建议：可以从 hook 方式起步（改动最小），当需要 Kedro-Viz 中的显式可见性、或需要严格控制执行顺序时，再增加 pipeline 节点。同时要决策好失败策略：是第一个失败就中止运行（fail-fast），还是收集多个问题后再统一报错——hook 示例采用前者，而收集模式可参考 Kedro 内置 Dataset Validation 中 Panderalazy选项"收集全部失败再报告"的思路。

替代方案：使用文件型 Data Context

如果不想在 Kedro hooks 或节点中硬编码期望，可以把 Great Expectations 数据上下文作为文件维护在外部。这种方式将数据校验配置与代码分离，便于跨环境、跨项目复用同一批期望。

首先，在项目根目录创建本地 GE 工作区：

mkdir great_expectations

然后在 Python 中初始化 context：

import great_expectations as gx context = gx.get_context(context_root_dir="great_expectations")

这会在指定路径创建 GE 上下文目录结构：

great_expectations/ ├── checkpoints/ ├── expectations/ ├── plugins/ ├── uncommitted/ ├── validation_definitions └── great_expectations.yml

该目录就是你的文件型 data context，存放所有配置、期望套件与验证结果。

与其在代码中内联定义期望，不如把期望存进expectations/目录下的 JSON 或 YAML 文件。例如，为 companies 数据集创建一个期望套件文件：

great_expectations/expectations/companies_suite.json

每个套件定义某个数据集的校验规则，如列存在性、null 检查或值范围。这些文件可以手工创建、由画像分析（profiling）代码生成，或从 GX Python API 导出：

import great_expectations as gx context = gx.get_context(context_root_dir="great_expectations") suite = gx.ExpectationSuite(name="companies_suite") suite.add_expectation( gx.expectations.ExpectColumnToExist(column="company_rating") ) context.suites.add(suite)

你会看到期望被写入companies_suite.json文件：

{ "expectations": [ { "id": "b6c459dc-6272-4509-a986-212cc65af82e", "kwargs": { "column": "company_rating" }, "meta": {}, "severity": "critical", "type": "expect_column_to_exist" } ], "id": "b43064bb-e486-401b-b9da-0224961de88b", "meta": { "great_expectations_version": "1.8.0" }, "name": "companies_suite", "notes": null }

注意 JSON 中"great_expectations_version": "1.8.0"与前置条件中great-expectations>=1.8.0的版本要求相互印证，说明该文件结构正是 1.8.x 时代的序列化格式。

在 Kedro hook 或 pipeline 节点中，不再用gx.get_context()创建内存上下文，而是加载指向项目目录的文件型上下文：

from pathlib import Path import great_expectations as gx def validate_companies(companies: pd.DataFrame,) -> None: context = gx.get_context(context_root_dir=Path.cwd() / "great_expectations") suite = context.suites.get("companies_suite") source = context.data_sources.add_or_update_pandas("companies_source") asset = source.add_dataframe_asset("companies") batch_request = asset.build_batch_request(options={"dataframe": companies}) batch = asset.get_batch(batch_request) result = batch.validate(suite) if not result.success: raise gx.exceptions.ValidationError(f"Validation failed for: 'companies'")

使用文件型 Data Context 的优势：

持久化期望套件与验证历史；
跨环境、跨团队成员共享套件；
可（可选）与 GX UI / GX Cloud 集成，获得更丰富的验证历史与协作评审体验。

两种集成路径的对比与选型决策

维度	Hook 方式（Approach 1）	Pipeline 节点方式（Approach 2）
侵入性	低，pipeline 代码零改动	高，需新增节点并改写 pipeline 编排
可见性	仅在日志中可见	在 Kedro-Viz 中显式呈现质量门禁
控制粒度	按数据集名自动匹配，全局生效	可精确控制执行顺序、可打 tag、可按 pipeline 运行/跳过
数据血缘	隐式	显式（如`companies → validated_companies`）
适用阶段	起步、快速建立安全网	需要可观测性与严格编排的生产场景

此外，还有一个常被忽略的"第三条路径"：Kedro 内置的 Dataset Validation。它在catalog.yml中为数据集声明validator，由DataCatalog在 load/save 时强制执行，支持 Pandera 等后端、severity: warn观察模式、enabled: false关闭开关，以及KEDRO_DATASET_VALIDATION环境变量应急开关。由于它接受自定义 validator 类，你可以把 GE 的校验逻辑包装成 Kedro validator 接入其中——这样既能复用 GE 的期望生态，又能获得 Kedro 声明式配置与三级开关（per-dataset / per-project / per-run）的便利。