Kedro Hooks 实战从内存监控到数据验证的五个完整示例【免费下载链接】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导读本文基于 Kedro 官方文档 docs/extend/hooks/examples.md系统讲解在 Kedro 项目中通过Hooks钩子机制注入横切行为的五种实战模式内存消耗跟踪、Great Expectations 数据验证、statsd/Grafana 管道可观测性、MLflow 模型指标跟踪以及用before_node_run动态覆盖节点输入。读完本文你将掌握hook_impl的完整使用套路——从定义、注册到在kedro run生命周期中触发的全过程并理解每个示例背后的源码级调用原理能够直接复制示例改造进自己的 Kedro 项目。前置知识Hooks 的核心概念在展开示例之前先明确 Kedro 中 Hooks 的基本模型。一个 Hook 由Hook 规范specification和Hook 实现implementation两部分组成Hook 规范由 Kedro 在 kedro/framework/hooks/specs.py 中预先定义描述了某个执行节点如before_node_run、after_dataset_loaded可以被注入额外行为的位置与参数签名。Hook 实现由你在项目中编写用hook_impl装饰器标记函数名与规范同名可以只声明规范参数的一个子集这是 pluggy 提供的opt-in 参数行为。Kedro 定义了一整套 Hook 规范完整清单与参数说明见 Hooks 介绍文档 及 kedro.framework.hooks API 参考。其中本文示例高频使用到的包括Hook可用参数before_pipeline_runrun_params,pipeline,catalogafter_pipeline_runrun_params,run_result,pipeline,catalogbefore_node_runnode,catalog,inputs,is_async,run_idafter_node_runnode,catalog,inputs,outputs,is_async,run_idbefore_dataset_loadeddataset_name,nodeafter_dataset_loadeddataset_name,data,node完整参数表见 docs/extend/hooks/introduction.md。实现 Hook 只需两步官方说明在src/package_name/hooks.py中定义 Hook 实现在src/package_name/settings.py的HOOKS元组中注册其实例# src/package_name/settings.py from package_name.hooks import MyHooks HOOKS (MyHooks(),)底层机制上Kedro 使用 pluggy 插件系统驱动 Hook 调度kedro/framework/hooks/manager.py中的_create_hook_manager()创建PluginManager并把NodeSpecs、PipelineSpecs、DataCatalogSpecs、DatasetSpecs、KedroContextSpecs五组规范全部注册进去。Hook 实现按LIFO后进先出顺序执行注册时必须是实例而非类否则会抛出TypeError见 manager.py 中的检查逻辑。注意Hook 实现参数不能带默认值。由于 pluggy 传参机制带默认值的参数会收到默认值而非 Kedro 实际传入的值详见 introduction.md。示例一内存消耗跟踪memory_profiler第一个示例展示如何在数据集加载前后测量内存消耗帮助定位流水线中的内存瓶颈。安装依赖pip install memory_profiler定义before_dataset_loaded与after_dataset_loadedHooks# src/package_name/hooks.py import logging from kedro.framework.hooks import hook_impl from memory_profiler import memory_usage def _normalise_mem_usage(mem_usage): # memory_profiler 0.56.0 returns list instead of float return mem_usage[0] if isinstance(mem_usage, (list, tuple)) else mem_usage class MemoryProfilingHooks: def __init__(self): self._mem_usage {} hook_impl def before_dataset_loaded(self, dataset_name: str) - None: before_mem_usage memory_usage( -1, interval0.1, max_usageTrue, retvalTrue, include_childrenTrue, ) before_mem_usage _normalise_mem_usage(before_mem_usage) self._mem_usage[dataset_name] before_mem_usage hook_impl def after_dataset_loaded(self, dataset_name: str) - None: after_mem_usage memory_usage( -1, interval0.1, max_usageTrue, retvalTrue, include_childrenTrue, ) # memory_profiler 0.56.0 returns list instead of float after_mem_usage _normalise_mem_usage(after_mem_usage) logging.getLogger(__name__).info( Loading %s consumed %2.2fMiB memory, dataset_name, after_mem_usage - self._mem_usage[dataset_name], )这里的memory_usage(-1, ...)表示测量当前进程PID -1的内存max_usageTrue返回采样区间的峰值内存include_childrenTrue会把子进程计入。_normalise_mem_usage兼容了 memory_profiler 旧版本返回列表、新版本返回 float 的行为差异。实现要点该示例使用了一个有状态的 Hook 类。实例的_mem_usage字典在before_dataset_loaded中记录基线、在after_dataset_loaded中做差值。这是 Kedro 中常见的模式——每个 Hook 在每个 Kedro session 中只有一个实例实例属性可以跨 Hook 调用共享状态详见 common_use_cases.md。注册并运行更新src/package_name/settings.pyHOOKS (MemoryProfilingHooks(),)然后重新运行管道$ kedro run源码佐证这两个 Hook 的触发点位于 kedro/runner/task.pyfor name in node.inputs: hook_manager.hook.before_dataset_loaded(dataset_namename, nodenode) inputs[name] catalog.load(name) hook_manager.hook.after_dataset_loaded( dataset_namename, datainputs[name], nodenode )可见before_dataset_loaded在catalog.load()之前触发after_dataset_loaded在数据加载完成之后触发——两个 Hook 正好夹住数据加载这个耗时/耗内存的操作。同步与异步模式下均会执行异步模式经由_synchronous_dataset_load包装以保证 Hook 同步执行见 task.py。与节点级 Hook 不同数据集级 Hookbefore_dataset_loaded/after_dataset_loaded等在ParallelRunner的 worker 进程中不会执行因为节点是在 worker 进程里运行的。如果项目依赖这些 Hook应改用SequentialRunner或ThreadRunner官方警告。示例二用 Great Expectations 做数据验证第二个示例在节点边界上对节点的输入和输出进行数据验证。这里使用 Great Expectations 作为验证引擎。注意如果只是想对 catalog 数据集在加载/保存时做验证Kedro 内置的 数据集验证dataset validation 无需任何 Hook 代码——在catalog.yml里声明validator即可。本示例适用于需要在节点边界节点执行前/后验证的场景二者定位不同。安装依赖pip install great-expectations定义before_node_run与after_node_runHooksGreat Expectations 有 V2 与 V3 两代 APIKedro 官方分别给出了实现。两代实现共用同一个 Hook 骨架before_node_run验证节点输入、after_node_run验证节点输出用映射表把数据集名关联到期望套件V2或 checkpointV3。V2 API期望套件 Validation Operator# src/package_name/hooks.py from typing import Any, Dict from kedro.framework.hooks import hook_impl from kedro.io import DataCatalog import great_expectations as ge class DataValidationHooks: # Map expectation to dataset DATASET_EXPECTATION_MAPPING { companies: raw_companies_dataset_expectation, preprocessed_companies: preprocessed_companies_dataset_expectation, } hook_impl def before_node_run( self, catalog: DataCatalog, inputs: Dict[str, Any], run_id: str ) - None: Validate inputs data to a node based on using great expectation if an expectation suite is defined in DATASET_EXPECTATION_MAPPING. self._run_validation(catalog, inputs, run_id) hook_impl def after_node_run( self, catalog: DataCatalog, outputs: Dict[str, Any], run_id: str ) - None: Validate outputs data from a node based on using great expectation if an expectation suite is defined in DATASET_EXPECTATION_MAPPING. self._run_validation(catalog, outputs, run_id) def _run_validation( self, catalog: DataCatalog, data: Dict[str, Any], run_id: str ): for dataset_name, dataset_value in data.items(): if dataset_name not in self.DATASET_EXPECTATION_MAPPING: continue dataset catalog._get_dataset(dataset_name) dataset_path str(dataset._filepath) expectation_suite self.DATASET_EXPECTATION_MAPPING[dataset_name] expectation_context ge.data_context.DataContext() batch expectation_context.get_batch( {path: dataset_path, datasource: files_datasource}, expectation_suite, ) expectation_context.run_validation_operator( action_list_operator, assets_to_validate[batch], run_idrun_id, )注册方式与示例一相同详见 Hooks 文档的注册章节然后运行 Kedro。验证失败时Great Expectations 会生成如下格式的报告V3 APICheckpointV3 API 以Checkpoint为核心。首先创建新的 checkpointgreat_expectations checkpoint new raw_companies_dataset_checkpoint然后从 checkpoint 配置文件的batch_request中移除data_connector_query因为下面的 Hook 会通过runtime_parameters直接传入内存中的 batch 数据不再依赖索引定位yaml_config f name: {my_checkpoint_name} config_version: 1.0 class_name: SimpleCheckpoint run_name_template: %Y%m%d-%H%M%S-my-run-name-template validations: - batch_request: datasource_name: {my_datasource_name} data_connector_name: default_runtime_data_connector_name data_asset_name: my_runtime_asset_name data_connector_query: index: -1 expectation_suite_name: {my_expectation_suite_name} # src/package_name/hooks.py from typing import Any, Dict from kedro.framework.hooks import hook_impl from kedro.io import DataCatalog import great_expectations as ge class DataValidationHooks: # Map checkpoint to dataset DATASET_CHECKPOINT_MAPPING { companies: raw_companies_dataset_checkpoint, } hook_impl def before_node_run( self, catalog: DataCatalog, inputs: Dict[str, Any], run_id: str ) - None: Validate inputs data to a node based on using great expectation if an expectation suite is defined in DATASET_EXPECTATION_MAPPING. self._run_validation(catalog, inputs, run_id) hook_impl def after_node_run( self, catalog: DataCatalog, outputs: Dict[str, Any], run_id: str ) - None: Validate outputs data from a node based on using great expectation if an expectation suite is defined in DATASET_EXPECTATION_MAPPING. self._run_validation(catalog, outputs, run_id) def _run_validation( self, catalog: DataCatalog, data: Dict[str, Any], run_id: str ): for dataset_name, dataset_value in data.items(): if dataset_name not in self.DATASET_CHECKPOINT_MAPPING: continue data_context ge.data_context.DataContext() data_context.run_checkpoint( checkpoint_nameself.DATASET_CHECKPOINT_MAPPING[dataset_name], batch_request{ runtime_parameters: { batch_data: dataset_value, }, batch_identifiers: { runtime_batch_identifier_name: dataset_name }, }, run_namerun_id, )V3 版本的关键差异通过runtime_parameters.batch_data把节点实际计算出的内存数据对象即inputs/outputs字典中的值直接交给 checkpoint 验证而不再像 V2 那样从文件路径重新读取因此更贴合节点边界即时验证的场景。源码佐证before_node_run的inputs参数携带的是已加载的实际数据而非数据集实例after_node_run的outputs同理见 specs.py 中 NodeSpecs 的文档。这解释了为什么 Hook 里可以直接把dataset_value传给 Great Expectations。示例三用 statsd Grafana 实现管道可观测性第三个示例把statsd指标与Grafana可视化结合实现对数据集大小和节点执行时长的监控。安装依赖pip install statsd定义before_node_run/after_node_run/after_pipeline_runHooks# src/package_name/hooks.py import sys from typing import Any, Dict import statsd from kedro.framework.hooks import hook_impl from kedro.pipeline.node import Node class PipelineMonitoringHooks: def __init__(self): self._timers {} self._client statsd.StatsClient(prefixkedro) hook_impl def before_node_run(self, node: Node) - None: node_timer self._client.timer(node.name) node_timer.start() self._timers[node.short_name] node_timer hook_impl def after_node_run(self, node: Node, inputs: Dict[str, Any]) - None: self._timers[node.short_name].stop() for dataset_name, dataset_value in inputs.items(): self._client.gauge(dataset_name _size, sys.getsizeof(dataset_value)) hook_impl def after_pipeline_run(self): self._client.incr(run)这个示例把三种不同的指标类型串在一起Timerbefore_node_run启动statsdtimerafter_node_run停止它从而度量单个节点的执行耗时。注意_timers以node.short_name为键而 timer 的名称用的是node.name更完整二者都来自节点对象。Gaugeafter_node_run中遍历节点的inputs字典用sys.getsizeof估算每个输入数据集在内存中的字节数以数据集名_size的 gauge 指标上报。Counterafter_pipeline_run中通过incr(run)记录管道运行次数。注册并运行按 Hooks 文档的注册章节 注册该 Hook 实现后运行 Kedro即可在 Grafana 中看到类似下面的监控面板实现要点这里的inputs: Dict[str, Any]声明的是节点输入数据集的实际值可以直接做sys.getsizeof估算若节点输入是大型 DataFramesys.getsizeof只反映对象外壳的大小生产环境可换成更精确的测量手段如df.memory_usage(deepTrue)。示例四用 MLflow 跟踪模型指标第四个示例演示如何把 Kedro 管道运行与 MLflow 实验跟踪打通以 Kedro 的run_id作为 MLflow run 名称记录参数、模型与指标。安装依赖pip install mlflow定义before_pipeline_run/after_pipeline_run/after_node_runHooks# src/package_name/hooks.py from typing import Any, Dict import mlflow import mlflow.sklearn from kedro.framework.hooks import hook_impl from kedro.pipeline.node import Node class ModelTrackingHooks: Namespace for grouping all model-tracking hooks with MLflow together. hook_impl def before_pipeline_run(self, run_params: Dict[str, Any]) - None: Hook implementation to start an MLflow run with the run_id of the Kedro pipeline run. mlflow.start_run(run_namerun_params[run_id]) mlflow.log_params(run_params) hook_impl def after_node_run( self, node: Node, outputs: Dict[str, Any], inputs: Dict[str, Any] ) - None: Hook implementation to add model tracking after some node runs. In this example, we will: * Log the parameters after the data splitting node runs. * Log the model after the model training node runs. * Log the models metrics after the model evaluating node runs. if node._func_name split_data: mlflow.log_params( {split_data_ratio: inputs[params:example_test_data_ratio]} ) elif node._func_name train_model: model outputs[example_model] mlflow.sklearn.log_model(model, model) mlflow.log_params(inputs[parameters]) hook_impl def after_pipeline_run(self) - None: Hook implementation to end the MLflow run after the Kedro pipeline finishes. mlflow.end_run()实现中的几个关键设计before_pipeline_run用run_params[run_id]作为 MLflow run 名称并把整个run_params字典含环境、runner、tags、from_nodes/to_nodes 等记录为 MLflow 参数——run_params的完整 schema 定义在 specs.py 的 PipelineSpecs。after_node_run通过node._func_name判断当前节点类型分节点执行不同的跟踪动作数据切分节点记录切分比例参数、训练节点用mlflow.sklearn.log_model记录模型并记录模型超参数。node._func_name是节点内部保存的函数名属性从源码结构看属于 Kedro Node 的私有属性示例中直接访问。after_pipeline_run在管道结束后调用mlflow.end_run()关闭本次 MLflow run与before_pipeline_run的start_run配对。注册并运行后可在 MLflow UI 中看到如下跟踪页面同一模式也适用于kedro-mlflow等更成熟的集成方案本文示例展示了不引入额外插件、纯 Hook 实现跟踪的最小路径。示例五用before_node_run动态覆盖节点输入最后一个示例展示了 Hooks 的一个高级特性如果before_node_runHook 实现返回一个字典该字典会被用来更新对应节点的输入。机制说明假设管道中有一个名为my_node的节点它接收两个输入first_input和second_input。我们想用别的值替换传给my_node的first_input可以定义如下 Hook# src/package_name/hooks.py from typing import Any, Dict, Optional from kedro.framework.hooks import hook_impl from kedro.pipeline.node import Node from kedro.io import DataCatalog class NodeInputReplacementHook: hook_impl def before_node_run( self, node: Node, catalog: DataCatalog ) - dict[str, Any] | None: Replace first_input for my_node if node.name my_node: # return the string filepath to the first_input dataset # instead of the underlying data dataset_name first_input filepath catalog._get_dataset(dataset_name)._filepath return {first_input: filepath} # second_input is not affected return None要点解析Hook 只对my_node生效通过node.name判断其余节点返回None不做任何覆盖返回值字典的键是节点输入名值是新的输入值。示例中把first_input替换为其数据集的文件路径字符串catalog._get_dataset(dataset_name)._filepathsecond_input不受影响before_node_run中创建的输入覆盖只作用于特定节点DataCatalog中对应的数据集本身不变。源码佐证覆盖如何生效before_node_run的返回语义在 specs.py 的 NodeSpecs.before_node_run 规范 中有明确定义Returns: Either None or a dictionary mapping dataset name(s) to new value(s). If returned, this dictionary will be used to update the node inputs, which allows to overwrite the node inputs.实际合并逻辑位于 kedro/runner/task.pyadditional_inputs self._collect_inputs_from_hook( node, catalog, inputs, is_async, hook_manager, run_idrun_id ) inputs.update(additional_inputs)即Hook 返回的字典先由_collect_inputs_from_hook收集其中会调用hook_manager.hook.before_node_run(...)然后通过inputs.update(additional_inputs)合并进节点输入字典最终传给_call_node_run。由于是dict.update返回字典中的键会覆盖同名的原始输入而未被提到的输入保持不变。重要约束在before_node_run中返回覆盖值时返回的键必须存在于节点的inputs字典中。如果返回了不在inputs中的数据集名节点会以如下错误失败Node name expected X input(s) expected_inputs, but got the following Y input(s) instead: actual_inputs完成 Hook 实现后同样需要按 Hooks 文档的注册章节 注册再运行 Kedro。注册与执行顺序的补充说明所有示例都遵循同一个注册流程官方文档在src/package_name/hooks.py中定义 Hook 类在src/package_name/settings.py的HOOKS元组中注册实例运行kedro run。关于执行顺序还有几个值得注意的细节多个实现按 LIFO 执行例如HOOKS (hook_a, hook_b,)时hook_b会先于hook_a执行插件 Hook 再按字母序排在项目 Hook 之后。一般不建议依赖执行顺序确有需要时可用 pluggy 的tryfirst/trylast参数控制见 introduction.md。插件自动发现Kedro 默认会通过kedro.hooksentry point 自动注册已安装插件的 Hook自动发现的 Hook 会先于settings.py中指定的 Hook 运行。可用DISABLE_HOOKS_FOR_PLUGINS (plugin_name,)禁用某插件的自动注册 Hook。调试辅助当项目日志级别设为DEBUG时pluggy 的 tracing 特性会记录每个 Hook 的执行便于排查但会影响性能日志级别为INFO或更高时可关闭详见 introduction.md。ParallelRunner 限制再次强调使用ParallelRunner时节点运行在 worker 进程中dataset级和node级 Hook 不会在 worker 中执行SequentialRunner与ThreadRunner则没有此限制。小结本文的五个示例覆盖了 Kedro Hooks 最典型的应用场景场景核心 Hook第三方依赖解决什么问题内存消耗跟踪before_dataset_loaded/after_dataset_loadedmemory_profiler定位数据集加载的内存峰值数据验证before_node_run/after_node_rungreat-expectations在节点边界验证输入/输出数据质量管道可观测性before_node_run/after_node_run/after_pipeline_runstatsd Grafana度量节点耗时、数据集大小、运行次数模型指标跟踪before_pipeline_run/after_node_run/after_pipeline_runmlflow记录参数、模型与指标串联实验节点输入覆盖before_node_run返回字典无按节点动态替换输入数据它们的共同模式是用hook_impl定义实现 → 在settings.py注册 → 由 pluggy 驱动的 Hook 管理器在 Kedro 执行时间线kedro/framework/hooks/manager.py上自动调度。数据集级 Hook 的触发点位于 kedro/runner/task.py节点级 Hook 的输入覆盖合并逻辑位于 kedro/runner/task.py规范的完整参数与返回语义定义于 kedro/framework/hooks/specs.py。如果你需要更多常见场景如按标签/命名空间为节点添加行为、加载外部凭据、pdb 后置调试、读取 DataCatalog metadata 等可以继续阅读 Hooks 常见用例想深入了解 Hooks 的设计原理与全部规范参数请参考 Hooks 介绍。【免费下载链接】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），仅供参考