钩子(Hooks)简介¶

概念¶

钩子(Hook)由钩子规范和钩子实现两部分组成。

钩子规范¶

Kedro为特定的执行点定义了钩子规范，用户可以在这些点注入额外的行为。目前，hooks中提供了以下钩子规范：

after_context_created
after_catalog_created
before_pipeline_run
before_dataset_loaded
after_dataset_loaded
before_node_run
after_node_run
before_dataset_saved
after_dataset_saved
after_pipeline_run
on_node_error
on_pipeline_error

非错误钩子的命名约定为__，其中：

和表示钩子执行的时间点，例如 before was run（在<某操作>运行之前）或 after was created（在<某对象>创建之后）。
指代Kedro执行时间线中该Hook添加额外行为的相关组件，例如 catalog、node 和 pipeline。

错误钩子的命名约定是on__error，其中：

指代Kedro执行时间线中抛出错误的相关组件。

hooks 列出了完整的规范，您可以通过提供实现来注入额外的行为。

该图展示了在kedro run期间钩子的执行顺序： kedro run hook execution order

CLI 钩子¶

Kedro 定义了一小组 CLI 钩子，用于在 Kedro CLI 命令执行时注入额外行为：

before_command_run
after_command_run

这是kedro-telemetry插件在底层所依赖的功能，用于收集CLI使用统计数据。

钩子实现¶

要将Hooks添加到您的Kedro项目中，您必须：

创建或修改文件 src//hooks.py 来为特定的Hook规范定义一个Hook实现，该规范描述了您想要注入额外行为的位置
在src//settings.py文件中的HOOKS键下注册该Hook实现

定义钩子实现¶

钩子(Hook)的实现应与规范同名。钩子必须提供一个具体实现，使用对应规范参数的一个子集(不需要使用全部参数)。

要声明一个Hook实现，请使用@hook_impl装饰器。

例如， after_catalog_created() Hook 规范的全部签名为：

@hook_spec
def after_catalog_created(
    self,
    catalog: DataCatalog,
    conf_catalog: Dict[str, Any],
    conf_creds: Dict[str, Any],
    save_version: str,
    load_versions: Dict[str, str],
) -> None:
    pass

不过，如果您只想在数据目录创建后使用此Hook来列出其内容，您的Hook实现可以非常简单：

# 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())

注意

包含Hooks实现的模块名称可以任意命名，不限于hooks.py。

我们建议您在项目中创建的hooks.py文件内，将相关的Hook实现按命名空间（最好是一个类）进行分组。

向Kedro注册钩子实现¶

钩子实现应通过src//settings.py文件中的HOOKS键注册到Kedro。

您可以为同一规范注册多个实现。它们将按照后进先出（LIFO）的顺序被调用。

以下示例设置了一个钩子，以便每次创建数据目录后，都会调用after_data_catalog_created的实现。

# src/<package_name>/settings.py
from <package_name>.hooks import ProjectHooks, DataCatalogHooks

HOOKS = (ProjectHooks(), DataCatalogHooks())

Kedro默认启用了自动发现功能。这意味着任何已安装并声明了Hooks入口点的插件都会被自动注册。如需了解如何为自定义插件启用此功能，请参阅我们的插件开发指南。

注意

自动发现的Hooks将优先运行，随后执行在settings.py中指定的Hooks。

通过插件自动注册的钩子¶

你可以通过创建一个Kedro插件来自动注册一个Hook（可通过pip安装）。Kedro提供了kedro.hooks入口点来轻松扩展此功能。

禁用自动注册插件的Hooks¶

自动注册插件的Hooks可以通过settings.py按如下方式禁用：

# src/<package_name>/settings.py

DISABLE_HOOKS_FOR_PLUGINS = ("<plugin_name>",)

其中是已安装插件的名称，必须禁用该插件的自动注册Hooks。

钩子执行顺序¶

钩子遵循后进先出（LIFO）的顺序，这意味着第一个注册的钩子将最后执行。

钩子按以下顺序注册：

settings.py中的项目钩子 - 如果你有HOOKS = (hook_a, hook_b,)，那么hook_b会在hook_a之前执行
在kedro.hooks中注册的插件钩子，按字母顺序排列

通常情况下，Hook的执行顺序无法保证，您不应依赖它。如果需要确保某个特定的Hook最先或最后执行，可以使用tryfirst或trylast参数来修饰hook_impl。

底层原理¶

在底层实现上，我们使用pytest的pluggy来实现Kedro的Hook机制。我们建议阅读其文档以了解更多关于底层实现的细节。

注意

当项目的日志级别设置为DEBUG时，钩子将使用pluggy的追踪功能记录每个钩子的执行情况。这对调试很有帮助，但可能会产生大量日志并拖慢管道执行速度。您可以通过将日志级别设为INFO或更高来禁用此功能。

插件钩子¶

插件钩子通过importlib_metadata的EntryPoints API进行注册。