Kedro 插件开发完整指南基于 pluggy 扩展 CLI、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 的插件plugin机制允许开发者以独立 Python 包的形式为 Kedro 添加新功能、向 CLI 注入新命令并自动注册 Hook 实现且无需修改 Kedro 项目本身。本文以 docs/extend/plugins.md 为主线结合当前仓库的 CLI 源码kedro/framework/cli/utils.py、kedro/framework/cli/cli.py与 Hooks 实现kedro/framework/hooks/manager.py系统讲解从零编写一个可安装、可运行、可维护的 Kedro 插件的完整路径掌握entry_points配置、click命令组约定、global/project命令分类、kedro.init初始化入口、懒加载优化以及 Hooks 与 CLI Hooks 的自动注册方法。插件机制底层pluggy entry pointsKedro 的扩展机制构建在pluggy之上pluggy 是诞生于 pytest 生态的成熟插件管理库。pluggy 依赖 Python 的 entry points 机制——包通过pyproject.toml或setup.py声明入口点其他包如 Kedro CLI则借助importlib.metadata在运行时发现并加载这些入口点。在 Kedro 内部所有入口点分组被集中定义在 kedro/framework/cli/utils.py 的ENTRY_POINT_GROUPS字典中ENTRY_POINT_GROUPS { global: kedro.global_commands, project: kedro.project_commands, init: kedro.init, line_magic: kedro.line_magic, hooks: kedro.hooks, cli_hooks: kedro.cli_hooks, starters: kedro.starters, }加载过程由load_entry_points()完成kedro/framework/cli/utils.py它先通过_get_entry_points()用importlib.metadata.entry_points().select(group...)获取该组全部入口点再调用_safe_load_entry_point()逐个加载。值得注意的健壮性设计是单个入口点加载失败例如插件代码抛异常不会导致整个 CLI 崩溃而是记录Failed to load ... commands from ...警告后跳过该入口点其余插件照常加载。第一个示例插件kedrojson原文档给出了一个打印流水线 JSON 的最小插件。在插件包kedrojson内创建kedrojson/plugin.pyimport click from kedro.framework.project import pipelines click.group(nameJSON) def commands(): pass commands.command(nameto_json) click.pass_obj def to_json(metadata): Display the pipeline in JSON format pipeline pipelines[__default__] print(pipeline.to_json())关键点pipelines来自 kedro/framework/project/init.py是当前项目的流水线注册表pipelines[__default__]取默认流水线其to_json()方法定义于 kedro/pipeline/pipeline.py将流水线序列化为 JSON。命令接收metadata参数当你在 Kedro 项目内运行 CLI 时KedroCLI 会把项目元数据作为obj注入 click 上下文见 kedro/framework/cli/cli.py因此click.pass_obj能拿到它。声明 entry point从 Kedro 0.18.14 起项目统一使用pyproject.toml替代setup.py。在插件的pyproject.toml中添加[project.entry-points.kedro.project_commands] kedrojson kedrojson.plugin:commands该配置的含义是在kedro.project_commands入口点组注册一个名为kedrojson的入口点指向kedrojson.plugin模块中的commands对象即上面定义的click.Group。安装插件后即可运行kedro to_json运行kedro info时Kedro 会列出所有已安装插件及其版本与入口点分组见 kedro/framework/cli/cli.py 的info命令实现可用于验证插件是否被正确发现。与 click 协作Group 的合并规则插件命令必须以click的Group形式提供。Kedro 主 CLI 是一个click.CommandCollection见 kedro/framework/cli/utils.py 的CommandCollection类它会将插件的 Group 与内置命令合并进主 CLI Group。合并时的两个限制必须牢记Group 自身的 options 会丢失Group 回调函数callback中执行的处理逻辑也会丢失。也就是说插件 Group 上通过click.group()注册的回调不会在命令执行时被调用因此不要把初始化逻辑放在 Group 的回调里——那部分逻辑应当转移到子命令内部或使用下文介绍的kedro.init/ Hooks 机制。从 kedro/framework/cli/cli.py 的global_groups与project_groups属性可以看到完整的合并与覆盖优先级Global 命令[cli, *load_entry_points(global), global_commands]Project 命令[user_defined, *plugins, project_commands]若项目存在自定义cli.py或[*plugins, project_commands]无自定义 cli.py覆盖顺序后者覆盖前者内置命令 插件命令 项目自定义cli.py。这意味着插件可以覆盖 Kedro 的内置命令而项目自身又可覆盖插件命令。在插件中访问项目上下文插件运行时可能需要读取当前项目的信息。标准做法是创建一个KedroSession并加载其 contextfrom pathlib import Path from kedro.framework.session import KedroSession project_path Path.cwd() session KedroSession.create(project_pathproject_path) context session.load_context()KedroSession的实现位于 kedro/framework/session/session.py其抽象基类在 kedro/framework/session/abstract_session.py。通过session.load_context()返回的KedroContextkedro/framework/context/context.py可访问catalog、config_loader、params、credentials等核心对象是插件与项目深度集成的入口。初始化入口kedro.init如果插件需要在 Kedro 启动之前完成某些初始化例如注册全局配置、打补丁、设置环境变量可以在pyproject.toml中声明kedro.init入口点[project.entry-points.kedro.init] my_plugin my_plugin.plugin:init该入口点必须指向一个无参数函数但为了未来兼容建议用**kwargs声明def init(**kwargs): # 初始化逻辑 ...在 kedro/framework/cli/cli.py 中main()入口函数会先调用_init_plugins()它加载所有kedro.init入口点并逐个调用随后才构造KedroCLI并执行命令。这保证了初始化发生在任何命令解析之前。global与project命令Kedro CLI 支持两类插件命令命令类型entry point 键可用范围globalkedro.global_commands项目内、项目外均可使用projectkedro.project_commands仅在检测到 Kedro 项目当前目录存在pyproject.toml时可用判定逻辑在 kedro/framework/cli/cli.pyKedroCLI.__init__用is_kedro_project(project_path)检测项目bootstrap_project()加载元数据project_groups属性在无项目元数据时直接返回空列表。因此project命令在非项目目录下不会出现在帮助信息中此时运行项目专属命令会得到 Kedro project not found in this directory 的提示。建议的命令命名约定社区与官方一致推荐kedro plugin-name command的形式即kedro plugin-name作为顶层命令组kedro my-plugin do-something这种结构是可选约定但能避免不同插件之间的命令名冲突也让用户一眼看出命令来源。进阶懒加载插件命令如果插件包含大量 CLI 命令或依赖体积庞大、导入缓慢的第三方库应当考虑懒加载lazy loading。这能显著改善插件与 Kedro CLI 本身的启动性能。从 Kedro 0.19.7 起Kedro 内置命令本身就采用了懒加载命令组见 kedro/framework/cli/cli.pyrun、jupyter、new、starter等均通过LazyGroup的lazy_subcommands延迟加载。LazyGroup的实现位于 kedro/framework/cli/utils.py它是一个click.Group子类构造时接收lazy_subcommands字典{命令名: 模块.对象}list_commands在常规命令之外追加懒加载命令名get_command遇到懒加载命令时才执行import_module并获取对应命令对象。这样命令对象与相关 import 会被延迟到真正调用时才执行。回到kedrojson示例。假设插件有两个子命令kedro to_json pipelines与kedro to_json nodes其中nodes命令需要导入大型库、使用频率较低则可如下实现在kedrojson/plugin.pyimport click from kedro.framework.project import pipelines from kedro.framework.cli.utils import LazyGroup click.group() def commands(): pass commands.group( nameto_json, clsLazyGroup, lazy_subcommands{ nodes: kedrojson.plugin.nodes, pipelines: kedrojson.plugin.pipelines } ) def to_json(): Convert Kedro nodes and pipelines to JSON pass click.command(namenodes) def nodes(): Convert Kedro nodes to JSON import some_large_library print(Converting nodes to JSON) ... click.command(pipelines) def pipelines(): Convert Kedro pipelines to JSON print(Converting pipelines to JSON) ...这里kedrojson.plugin.nodes与kedrojson.plugin.pipelines指向独立模块可分别放在kedrojson/nodes.py与kedrojson/pipelines.py中各模块内部再导入各自需要的大库。nodes与pipelines命令及它们的依赖导入都会被推迟到对应命令真正执行时才发生从而加快kedro命令本身的响应速度。Hooks插件中声明 Hook 实现并自动注册插件可以开发 Hook 实现并在安装后由 Kedro 自动注册到项目上下文中。实现方式是在pyproject.toml中声明kedro.hooks入口点[project.entry-points.kedro.hooks] plugin_name plugin_name.plugin:hooksplugin.py中声明 Hook 实现注意hooks必须是类实例而非类本身import logging from kedro.framework.hooks import hook_impl class MyHooks: hook_impl def after_catalog_created(self, catalog): logging.info(Reached after_catalog_created hook) hooks MyHooks()底层机制hook_impl标记来自 kedro/framework/hooks/markers.py命名空间为kedro。Hook 规范spec定义在 kedro/framework/hooks/specs.py按生命周期分为五组DataCatalogSpecsafter_catalog_created接收catalog、conf_catalog、conf_creds、parameters、save_version、load_versionsNodeSpecsbefore_node_run、after_node_run、on_node_errorPipelineSpecsbefore_pipeline_run、after_pipeline_run、on_pipeline_errorDatasetSpecsbefore_dataset_loaded、after_dataset_loaded、before_dataset_saved、after_dataset_savedKedroContextSpecsafter_context_created。注册流程在 kedro/framework/hooks/manager.py_register_hooks_entry_points()通过hook_manager.load_setuptools_entrypoints(_PLUGIN_HOOKS)加载kedro.hooks组的全部入口点_PLUGIN_HOOKS kedro.hooks并支持基于插件项目名dist.project_name禁用特定插件的 Hooks。此外_register_hooks()会校验注册对象必须是实例若传入类会抛出TypeError并提示Have you forgotten the () when registering a hook class?。值得注意的是新生成项目的模板pyproject.tomlkedro/templates/project/{{ cookiecutter.repo_name }}/pyproject.toml本身就预留了[project.entry-points.kedro.hooks]空段说明这是官方推荐的、每个项目都具备的扩展位。仓库内置的测试插件 features/test_plugin/plugin.py 也演示了同样的模式它定义MyPluginHook.after_catalog_created并在 features/test_plugin/pyproject.toml 中以test_plugin plugin:hooks注册。CLI Hooks扩展 Kedro CLI 行为除了运行期 Hooks插件还可以开发 CLI Hooks 来扩展 Kedro 命令行行为。当前可用的 CLI Hook 规范参见 kedro/framework/cli/hooks/specs.py包括before_command_run(project_metadata, command_args)在某个 CLI 命令执行前触发接收项目元数据与全部命令行参数含命令及子命令本身after_command_run(project_metadata, command_args, exit_code)在命令执行结束后触发exit_code表示退出码。在插件的pyproject.toml中注册[project.entry-points.kedro.cli_hooks] plugin_name plugin_name.plugin:cli_hooksplugin.py中的实现示例import logging from kedro.framework.cli.hooks import cli_hook_impl class MyCLIHooks: cli_hook_impl def before_command_run(self, project_metadata, command_args): logging.info( Command %s will be run for project %s, command_args, project_metadata ) cli_hooks MyCLIHooks()与运行期 Hooks 类似CLI Hooks 使用独立的命名空间kedro_cli见 kedro/framework/cli/hooks/markers.py由 kedro/framework/cli/hooks/manager.py 中的CLIHooksManager管理它以_CLI_PLUGIN_HOOKS kedro.cli_hooks为入口点组加载插件实现。调用时机在 kedro/framework/cli/cli.py 的KedroCLI.main()中命令执行前调用before_command_run命令以SystemExit正常结束或抛出异常后分别调用after_command_run异常路径的exit_code为 1。这让插件可以统一实现命令审计、日志记录、指标上报等横切能力。提交贡献插件发布流程当插件开发完成并准备发布时官方建议按以下流程操作独立仓库为插件创建独立仓库并按命名约定命名kedro-plugin-name选择命令策略确定提供global和/或project命令——所有global命令放在一个clickGroup 中所有project命令放在另一个clickGroup 中两组均通过 entry points 机制声明编写 README在README.md中说明插件功能及全部依赖GitHub 标签为仓库打上kedro-plugin标签便于官方与社区发现。如果插件希望进入官方维护的插件列表其许可证必须与 Apache 2.0 兼容。官方支持的插件Kedro-DatasetsKedro 数据连接器合集是AbstractDatasetkedro/io/core.py 中定义的具体实现集合Kedro-Docker将 Kedro 项目打包并放入容器运行的工具Kedro-Airflow将 Kedro 项目转换为 Airflow 项目的工具Kedro-VizKedro 流水线可视化工具。社区维护的插件速览社区中已有大量活跃维护的插件完整清单收录于awesome-kedro仓库。以下是其中一小部分代表仅列举用于说明插件的典型应用场景kedro-mlflow将 MLflow 集成到 Kedro 项目提供模块化配置、参数自动追踪、数据集版本化、流水线打包与 serving以及训练/推理流水线自动同步配套教程仓库与独立文档kedro-kubeflow通过 Kubeflow Pipelines 在 Kubernetes 集群上运行与调度流水线kedro-airflow-k8s在 Kubernetes 集群上用 Airflow 运行 Kedro 流水线kedro-vertexai与 Vertex AI Pipelines 服务集成kedro-azureml与 Azure ML Pipelines 服务集成kedro-sagemaker与 Amazon SageMaker 服务集成kedro-partitioned扩展分区数据处理能力kedro-dagster将 Kedro 项目转换为 Dagster 项目附示例项目与文档。这些插件正是本文所述机制的实践样本无论是云平台集成、编排系统对接还是实验追踪与可视化都以独立 Python 包 entry points 声明 click Group / Hooks 实现这一统一模式融入 Kedro 生态。你也可以参考本仓库的 features/test_plugin/含 Hooks 与自定义 Starter 的完整最小示例作为开发自己插件的起点。【免费下载链接】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），仅供参考