Kedro Hooks 机制完全指南解读 kedro.framework.hooks 的 manager、markers 与 specs 全量 API【免费下载链接】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 官方 API 文档 kedro.framework.hooks 为核心骨架系统讲解 Kedro 的 Hooks 扩展机制manager全局 hook 管理器、markers声明式装饰器与specs全部可调用 Hook 的规格定义三个子模块的职责与用法并结合 hooks 规范源码、会话与运行器中的调用点 以及 Hook 实践指南帮助你掌握如何编写、注册、调试 Hook实现在kedro run执行时间线的任意关键节点注入自定义行为。一、Hook 机制与模块总览Kedro 的 Hooks 机制基于 pluggy即 pytest 所用的插件系统构建一个 Hook 由Hook specification规格与Hook implementation实现两部分组成。Kedro 在源码中预定义了一批规格声明在哪个执行节点可以注入行为用户或插件则提供实现声明在该节点做什么。在kedro.framework.hooks包见 kedro/framework/hooks/init.py中公开导出了_create_hook_manager与hook_impl并包含三个子模块其分工如下表模块描述kedro.framework.hooks.manager提供工具函数用于在 Kedro 执行进程中创建全局hook_manager单例并完成 Hook 的注册与插件入口点加载kedro.framework.hooks.markers提供声明式 markershook_spec/hook_impl用于标记 Kedro 的 Hook 规格与实现kedro.framework.hooks.specs包含 Kedro 执行时间线上全部可调用 Hook 的规格定义5 个规范类共 12 个 Hook此外CLI 层面还有一组独立的 CLI Hooksbefore_command_run/after_command_run定义于 kedro/framework/cli/hooks/specs.py命名空间为kedro_cli。二、markersHook 的声明式标记markers.py 是整个机制的基石全文只有三个关键定义import pluggy HOOK_NAMESPACE kedro hook_spec pluggy.HookspecMarker(HOOK_NAMESPACE) hook_impl pluggy.HookimplMarker(HOOK_NAMESPACE)HOOK_NAMESPACE kedro所有 Kedro Hook非 CLI 类共享的命名空间。pluggy 依赖该命名空间将规格与实现进行匹配因此实现方法名必须与规格方法名完全一致。hook_specHookspecMarker实例用作hook_spec装饰器声明这是一个 Hook 规格。hook_implHookimplMarker实例用作hook_impl装饰器声明这是一个 Hook 实现。与之对应CLI Hooks 使用独立的命名空间与 markers见 kedro/framework/cli/hooks/markers.pyCLI_HOOK_NAMESPACE kedro_cli cli_hook_spec pluggy.HookspecMarker(CLI_HOOK_NAMESPACE) cli_hook_impl pluggy.HookimplMarker(CLI_HOOK_NAMESPACE)这意味着kedro.hooks与kedro.cli_hooks是两套相互独立的插件体系分别由各自的 manager 管理。三、specsKedro 执行时间线上的全部 Hook 规格specs.py 定义了 5 个规范类namespace每个类用hook_spec标记其方法构成 Kedro 运行生命周期的 12 个核心 Hook。3.1 命名约定非错误类 Hook 遵循before/after_noun_past_participle约定before/after与past_participle表示执行时机例如before something was run、after something was creatednoun表示被注入行为的组件例如catalog、node、pipeline、dataset、context。错误类 Hook 遵循on_noun_error约定noun表示抛出错误的组件。3.2 KedroContextSpecs上下文生命周期after_context_created是一次 Kedro 运行中最早触发的 Hook在KedroContext创建完成后立即调用见 session.py 中的调用。规格签名hook_spec def after_context_created(self, context: KedroContext) - None:其中context是刚创建的KedroContext实例携带credentials、config_loader、env等有用信息适合在此完成全局初始化如注入外部服务客户端。3.3 DataCatalogSpecs数据目录生命周期after_catalog_created在数据目录创建后触发接收catalog以及KedroContext._create_catalog的全部入参调用点见 context.pyhook_spec def after_catalog_created( self, catalog: CatalogProtocol, conf_catalog: dict[str, Any], conf_creds: dict[str, Any], parameters: dict[str, Any], save_version: str, load_versions: dict[str, str], ) - None:catalog已创建的目录实例conf_catalog用于创建目录的配置conf_creds用于创建目录的凭据配置parameters目录创建后注入的参数save_version目录中所有数据集save操作使用的版本号load_versions目录中各数据集load操作使用的版本号。典型用途在目录就绪后统一添加监控、记录数据集清单或按版本号做审计。3.4 NodeSpecs节点生命周期节点级 Hook 围绕单个节点的执行展开全部调用点位于 kedro/runner/task.py。before_node_run在节点执行前触发且允许通过返回值改写节点输入hook_spec def before_node_run( self, node: Node, catalog: CatalogProtocol, inputs: dict[str, Any], is_async: bool, run_id: str, ) - dict[str, Any] | None:inputs键是数据集名称值是已加载的实际数据而非数据集实例is_async表示节点是否以异步模式运行run_id是本次运行的 ID返回值None或数据集名 - 新值的字典若返回字典Kedro 将用它更新节点输入从而实现输入覆写例如注入测试数据或 mock。after_node_run在节点执行成功后触发额外提供outputs输出数据字典可用于数据质量检查、指标采集hook_spec def after_node_run( self, node: Node, catalog: CatalogProtocol, inputs: dict[str, Any], outputs: dict[str, Any], is_async: bool, run_id: str, ) - None:on_node_error在节点抛出未捕获异常时触发其签名与before_node_run一致并额外携带errorhook_spec def on_node_error( self, error: Exception, node: Node, catalog: CatalogProtocol, inputs: dict[str, Any], is_async: bool, run_id: str, ) - None:常用于失败告警、错误上报或失败现场转储。3.5 DatasetSpecs数据集加载/保存生命周期数据集级 Hook 在目录对单个数据集执行load/save前后触发签名均在 specs.py 的 DatasetSpecs 类 中定义hook_spec def before_dataset_loaded(self, dataset_name: str, node: Node) - None: hook_spec def after_dataset_loaded(self, dataset_name: str, data: Any, node: Node) - None: hook_spec def before_dataset_saved(self, dataset_name: str, data: Any, node: Node) - None: hook_spec def after_dataset_saved(self, dataset_name: str, data: Any, node: Node) - None:其中dataset_name为数据集名称data为实际加载/保存的数据node为触发该操作或刚刚运行完的节点。这类 Hook 适合做细粒度的数据血缘记录、敏感数据脱敏或 I/O 审计。3.6 PipelineSpecs流水线生命周期流水线级 Hook 在整条流水线运行前后触发调用点位于 session.pyServiceSession同构见 service_session.py。三者共享同一份run_params字典其完整 schema 在规格 docstring 内联定义{ run_id: str, project_path: str, env: str, kedro_version: str, tags: Optional[List[str]], from_nodes: Optional[List[str]], to_nodes: Optional[List[str]], node_names: Optional[List[str]], from_inputs: Optional[List[str]], to_outputs: Optional[List[str]], load_versions: Optional[List[str]], runtime_params: Optional[Dict[str, Any]], pipeline_names: Optional[List[str]], namespaces: Optional[List[str]], runner: str, only_missing_outputs: bool, }before_pipeline_run在流水线运行前触发run_params携带本次运行的全部参数可用于权限校验、参数审计hook_spec def before_pipeline_run( self, run_params: dict[str, Any], pipeline: Pipeline, catalog: CatalogProtocol ) - None:after_pipeline_run在流水线运行成功后触发额外提供run_result流水线运行输出可用于结果落库、指标聚合hook_spec def after_pipeline_run( self, run_params: dict[str, Any], run_result: dict[str, Any], pipeline: Pipeline, catalog: CatalogProtocol, ) - None:on_pipeline_error在流水线抛出未捕获异常时触发签名与before_pipeline_run一致并携带errorhook_spec def on_pipeline_error( self, error: Exception, run_params: dict[str, Any], pipeline: Pipeline, catalog: CatalogProtocol, ) - None:3.7 CLI Hooks命令生命周期除上述运行期 Hook 外Kedro 还定义了 CLI 级 Hook在 CLI 命令执行前后触发调用点见 kedro/framework/cli/cli.py其规格在 kedro/framework/cli/hooks/specs.pycli_hook_spec def before_command_run(self, project_metadata: ProjectMetadata, command_args: list[str]) - None: cli_hook_spec def after_command_run(self, project_metadata: ProjectMetadata, command_args: list[str], exit_code: int) - None:project_metadataKedro 项目的元数据command_args本次使用的全部命令行参数包含命令与子命令本身exit_codeClick 应用完成后的退出码仅after_command_run有。官方文档明确指出kedro-telemetry插件正是依赖这两组 CLI Hooks 来收集 CLI 使用统计的。四、manager全局 Hook 管理器manager.py 是 Hooks 的调度中枢提供四个核心工具函数/类。4.1_create_hook_manager()创建全局管理器def _create_hook_manager() - PluginManager: manager PluginManager(HOOK_NAMESPACE) manager.trace.root.setwriter( logger.debug if logger.getEffectiveLevel() logging.DEBUG else None ) manager.enable_tracing() manager.add_hookspecs(NodeSpecs) manager.add_hookspecs(PipelineSpecs) manager.add_hookspecs(DataCatalogSpecs) manager.add_hookspecs(DatasetSpecs) manager.add_hookspecs(KedroContextSpecs) return manager关键点基于 pluggy 的PluginManager命名空间为HOOK_NAMESPACE一次性注册全部 5 个规范类当项目日志级别为DEBUG时启用 pluggy 的 tracing将每次 Hook 的执行过程写入调试日志便于排查但会显著增加日志噪音、拖慢流水线文档建议生产环境保持INFO及以上。KedroSession在创建时即调用该函数持有全局hook_manager见 session.py。4.2_register_hooks()注册项目自定义 Hookdef _register_hooks(hook_manager: PluginManager, hooks: Iterable[Any]) - None: for hooks_collection in hooks: if not hook_manager.is_registered(hooks_collection): if isclass(hooks_collection): raise TypeError( KedroSession expects hooks to be registered as instances. Have you forgotten the () when registering a hook class ? ) hook_manager.register(hooks_collection)两个值得注意的行为重复注册保护若 Hook 已被注册则直接跳过避免用户多次调用注册逻辑导致崩溃实例校验Hook 必须以实例而非类形式注册忘记加()会抛出上述TypeError——这一点被 tests/framework/hooks/test_manager.py 中的test_register_hooks参数化测试明确覆盖[ExampleHook]报错、[ExampleHook()]通过。4.3_register_hooks_entry_points()加载插件 Hook_PLUGIN_HOOKS kedro.hooks # entry-point to load hooks from for installed plugins def _register_hooks_entry_points(hook_manager, disabled_plugins) - None: already_registered hook_manager.get_plugins() hook_manager.load_setuptools_entrypoints(_PLUGIN_HOOKS) disabled_plugins set(disabled_plugins) plugininfo hook_manager.list_plugin_distinfo() # ...根据 DISABLE_HOOKS_FOR_PLUGINS 逐一 unregister 指定插件通过kedro.hooks入口点自动加载已安装插件声明的 Hook这是 Kedro默认开启的自动发现机制对disabled_plugins中列出的插件按dist.project_name匹配而非入口点名调用unregister()禁用其 Hook并记录插件名-版本到调试日志。4.4_NullPluginManager空实现兜底class _NullPluginManager: def __init__(self, *args, **kwargs): ... def __getattr__(self, name): return self def __call__(self, *args, **kwargs): ...这是一个吞掉一切调用的空管理器当没有实例化任何hook_manager时它让 runner 仍可正常运行所有 Hook 调用被静默忽略。测试test_null_plugin_manager_returns_none_when_called验证了其调用返回None的行为。五、编写并注册一个 Hook 实现5.1 最小实现示例在项目src/package_name/hooks.py中用hook_impl标记实现方法名必须与规格同名参数可取规格参数的子集得益于 pluggy 的 opt-in arguments 机制未声明参数会被省略# src/package_name/hooks.py import logging from kedro.framework.hooks import hook_impl from kedro.io import DataCatalog class DataCatalogHooks: property def _logger(self): return logging.getLogger(__name__) hook_impl def after_catalog_created(self, catalog: DataCatalog) - None: self._logger.info(catalog.list())5.2 在 settings.py 中注册在src/package_name/settings.py的HOOKS键下注册实现见 kedro/framework/project/init.py 中_HOOKS校验器的定义# src/package_name/settings.py from package_name.hooks import ProjectHooks, DataCatalogHooks HOOKS (ProjectHooks(), DataCatalogHooks())同一规格可以注册多个实现按LIFO后进先出顺序调用即元组中靠后的实现先执行。KedroSession创建时依次调用_register_hooks(hook_manager, settings.HOOKS)与_register_hooks_entry_points(...)见 session.py因此自动发现的插件 Hook 先执行settings.py中指定的 Hook 后执行。5.3 禁用插件的自动注册 Hook通过settings.py的DISABLE_HOOKS_FOR_PLUGINS键对应 project/init.py 中_DISABLE_HOOKS_FOR_PLUGINS校验器禁用指定插件的自动 Hook# src/package_name/settings.py DISABLE_HOOKS_FOR_PLUGINS (plugin_name,)5.4 两个重要注意事项不要为 Hook 参数设置默认值由于 pluggy 传参机制带默认值的参数会收到默认值而非 Kedro 传入的真实值。例如def before_pipeline_run(self, run_params: dict {})中run_params恒为空字典必须写成无默认值的形式。ParallelRunner 下的行为差异使用ParallelRunner时catalog、context、pipeline级 Hook 会在主进程中正常执行但dataset与node级 Hook不会在并行 worker 进程中执行。若项目依赖这些 Hook请改用SequentialRunner或ThreadRunner。六、执行时间线与调用链验证Kedro 在kedro run生命周期中按下述顺序触发 Hook右侧为仓库中的真实调用点顺序Hook调用位置1after_context_createdsession.py2after_catalog_createdcontext.py3before_pipeline_runsession.py4before_dataset_loaded/after_dataset_loadedrunner/task.py5before_node_runrunner/task.py6before_dataset_saved/after_dataset_savedrunner/task.py7after_node_run成功/on_node_error失败runner/task.py8after_pipeline_run成功/on_pipeline_error失败session.py整个生命周期可通过文档配套的示意图直观理解关于参数匹配的完整性仓库测试 test_manager.py 中的test_hook_manager_can_call_hooks_defined_in_specs会对全部 12 个 Hook 逐一断言其参数集合与规格一致test_hook_args_doc_table_matches_specs更是直接解析 docs/extend/hooks/introduction.md 中的参数表格确保文档与源码签名不漂移issue #4564。这说明上表所列参数与规格签名即为可依赖的地面真相。七、调试与最佳实践小结开启 tracing将项目日志级别调至DEBUG即可看到每个 Hook 的执行记录排查完毕后恢复INFO及以上以避免性能损耗。保持实现无副作用顺序依赖除tryfirst/trylast参数外Hook 之间的执行顺序不保证业务逻辑不应依赖特定次序。聚合相关实现建议用类把相关 Hook 实现分组放入项目中的hooks.py模块名任意不强制。利用返回值覆写before_node_run的返回值可用于替换节点输入是实现测试替身、数据注入的官方入口。插件化复用将 Hook 打包为插件并在pyproject.toml中声明kedro.hooksCLI 类为kedro.cli_hooks入口点即可自动注册若要在settings.py之外彻底理解入口点加载与禁用逻辑可继续阅读 manager.py 与 kedro/framework/cli/hooks/manager.py。综上kedro.framework.hooks通过 pluggy 将规格声明specs markers与注册调度manager解耦规格定义好 12 个运行期 Hook 与 2 个 CLI Hook 的完整签名与参数语义manager 负责创建单例、校验注册、加载插件入口点并按settings.py的配置组织执行顺序从而为生产级数据流水线提供了稳定、可插拔的扩展点。【免费下载链接】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），仅供参考