使用 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/kedroGreat ExpectationsGE是一个开源的、面向生产环境的数据质量框架支持对数据进行验证、文档化与画像分析。本指南以 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 --starterspaceflights-pandas --name spaceflights-great-expectations进入项目目录后在requirements.txt中追加 GE 依赖great-expectations1.8.0安装项目依赖uv pip install -r requirements.txtgreat-expectations1.8.0这一版本门槛与本指南使用的代码 API 强相关——下文将说明为什么 1.0 版本彻底改变了使用方式。理解 Great Expectations1.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(namemy_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 HookApproach 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(columncompany_rating), ], reviews: [ gx.expectations.ExpectColumnToExist(columnreview_scores_rating), ], model_input_table: [ gx.expectations.ExpectColumnToExist(columnprice), gx.expectations.ExpectColumnValuesToNotBeNull(columnprice), ], } def get_suite(name: str) - gx.ExpectationSuite: suite gx.ExpectationSuite(namef{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.pyfrom 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(fValidating {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( fValidation failed for {name}:\n \n.join(f - {e} for e in errors) ) logger.info(f✓ {name} passed validation)这个 Hook 实现的工作机制可概括为四点配置驱动EXPECTATION_SUITES字典将数据集名映射到期望列表是校验规则的唯一事实来源自动触发每个节点运行前后hook 都会检查其输入/输出是否需要校验选择性校验只校验显式配置过的数据集未配置的数据集零开销快速失败fail-fast校验失败立即抛出ValueErrorpipeline 在运行下游节点前停止并给出清晰的错误信息列出所有失败期望的类型。注册 Hook在src/spaceflights_great_expectations/settings.py中注册自定义 hookfrom 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:0000: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:0000: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:0000: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(fValidation failed for: {name})接着更新src/spaceflights_great_expectations/pipelines/data_processing/pipeline.py把新节点编排进 pipelinefrom .nodes import create_model_input_table, preprocess_companies, preprocess_shuttles, validate_datasets def create_pipeline(**kwargs) - Pipeline: return Pipeline( [ Node( funcvalidate_datasets, inputs[companies, reviews, shuttles], outputsNone, namevalidade_datasets_node, ), Node( funcpreprocess_companies, inputscompanies, outputspreprocessed_companies, namepreprocess_companies_node, ), Node( funcpreprocess_shuttles, inputsshuttles, outputspreprocessed_shuttles, namepreprocess_shuttles_node, ), Node( funccreate_model_input_table, inputs[preprocessed_shuttles, preprocessed_companies, reviews], outputsmodel_input_table, namecreate_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( funcvalidate_companies, inputscompanies, outputsvalidated_companies, namevalidate_companies_node, ), node( funcpreprocess_companies, inputsvalidated_companies, # Use validated data outputspreprocessed_companies, namepreprocess_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 中初始化 contextimport great_expectations as gx context gx.get_context(context_root_dirgreat_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_dirgreat_expectations) suite gx.ExpectationSuite(namecompanies_suite) suite.add_expectation( gx.expectations.ExpectColumnToExist(columncompany_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-expectations1.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_dirPath.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(fValidation failed for: companies)使用文件型 Data Context 的优势持久化期望套件与验证历史跨环境、跨团队成员共享套件可可选与 GX UI / GX Cloud 集成获得更丰富的验证历史与协作评审体验。两种集成路径的对比与选型决策维度Hook 方式Approach 1Pipeline 节点方式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的便利。总结在 Kedro 中接入 Great Expectations 的核心决策可以概括为三点规则先行把期望定义在独立模块或文件型 Data Context中确保同一套规则可在 hooks、节点、测试与 CI 中复用按需选路径起步用 hook 建立自动安全网需要 DAG 可见性与严格编排时切换/补充为 pipeline 节点重视上下文持久化生产环境使用文件型 Data Context让套件、验证结果与历史可复现、可共享。延伸阅读Kedro Data CatalogKedro HooksKedro Dataset Validation内置数据校验Kedro PipelinesSpaceflights 完整教程了解更多 Great Expectations其官方学习文档与 Expectations 参考文档见 greatexpectations.io【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 使用 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/kedroGreat ExpectationsGE是一个开源的、面向生产环境的数据质量框架支持对数据进行验证、文档化与画像分析。本指南以 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 --starterspaceflights-pandas --name spaceflights-great-expectations进入项目目录后在requirements.txt中追加 GE 依赖great-expectations1.8.0安装项目依赖uv pip install -r requirements.txtgreat-expectations1.8.0这一版本门槛与本指南使用的代码 API 强相关——下文将说明为什么 1.0 版本彻底改变了使用方式。理解 Great Expectations1.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(namemy_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 HookApproach 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(columncompany_rating), ], reviews: [ gx.expectations.ExpectColumnToExist(columnreview_scores_rating), ], model_input_table: [ gx.expectations.ExpectColumnToExist(columnprice), gx.expectations.ExpectColumnValuesToNotBeNull(columnprice), ], } def get_suite(name: str) - gx.ExpectationSuite: suite gx.ExpectationSuite(namef{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.pyfrom 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(fValidating {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( fValidation failed for {name}:\n \n.join(f - {e} for e in errors) ) logger.info(f✓ {name} passed validation)这个 Hook 实现的工作机制可概括为四点配置驱动EXPECTATION_SUITES字典将数据集名映射到期望列表是校验规则的唯一事实来源自动触发每个节点运行前后hook 都会检查其输入/输出是否需要校验选择性校验只校验显式配置过的数据集未配置的数据集零开销快速失败fail-fast校验失败立即抛出ValueErrorpipeline 在运行下游节点前停止并给出清晰的错误信息列出所有失败期望的类型。注册 Hook在src/spaceflights_great_expectations/settings.py中注册自定义 hookfrom 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:0000: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:0000: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:0000: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(fValidation failed for: {name})接着更新src/spaceflights_great_expectations/pipelines/data_processing/pipeline.py把新节点编排进 pipelinefrom .nodes import create_model_input_table, preprocess_companies, preprocess_shuttles, validate_datasets def create_pipeline(**kwargs) - Pipeline: return Pipeline( [ Node( funcvalidate_datasets, inputs[companies, reviews, shuttles], outputsNone, namevalidade_datasets_node, ), Node( funcpreprocess_companies, inputscompanies, outputspreprocessed_companies, namepreprocess_companies_node, ), Node( funcpreprocess_shuttles, inputsshuttles, outputspreprocessed_shuttles, namepreprocess_shuttles_node, ), Node( funccreate_model_input_table, inputs[preprocessed_shuttles, preprocessed_companies, reviews], outputsmodel_input_table, namecreate_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( funcvalidate_companies, inputscompanies, outputsvalidated_companies, namevalidate_companies_node, ), node( funcpreprocess_companies, inputsvalidated_companies, # Use validated data outputspreprocessed_companies, namepreprocess_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 中初始化 contextimport great_expectations as gx context gx.get_context(context_root_dirgreat_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_dirgreat_expectations) suite gx.ExpectationSuite(namecompanies_suite) suite.add_expectation( gx.expectations.ExpectColumnToExist(columncompany_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-expectations1.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_dirPath.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(fValidation failed for: companies)使用文件型 Data Context 的优势持久化期望套件与验证历史跨环境、跨团队成员共享套件可可选与 GX UI / GX Cloud 集成获得更丰富的验证历史与协作评审体验。两种集成路径的对比与选型决策维度Hook 方式Approach 1Pipeline 节点方式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的便利。总结在 Kedro 中接入 Great Expectations 的核心决策可以概括为三点规则先行把期望定义在独立模块或文件型 Data Context中确保同一套规则可在 hooks、节点、测试与 CI 中复用按需选路径起步用 hook 建立自动安全网需要 DAG 可见性与严格编排时切换/补充为 pipeline 节点重视上下文持久化生产环境使用文件型 Data Context让套件、验证结果与历史可复现、可共享。延伸阅读Kedro Data CatalogKedro HooksKedro Dataset Validation内置数据校验Kedro PipelinesSpaceflights 完整教程了解更多 Great Expectations其官方学习文档与 Expectations 参考文档见 greatexpectations.io【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考