如何通过命令行传递路径参数运行Kedro流水线

你可以通过扩展CLI参数 + 钩子注入模板变量的方式实现需求，具体步骤如下：

1. 配置TemplatedConfigLoader #

首先在项目的src/<your_project_name>/settings.py中指定使用模板配置加载器：

from kedro.config import TemplatedConfigLoader
# 导入后续自定义的钩子
from <your_project_name>.hooks import CommandLineParamsHook

CONFIG_LOADER_CLASS = TemplatedConfigLoader
# 注册自定义钩子
HOOKS = (CommandLineParamsHook(),)

2. 实现自定义钩子 #

在src/<your_project_name>/hooks.py中添加钩子类，实现两个核心能力：给kedro run命令新增自定义参数、将命令行参数注入到配置模板的全局变量中：

from kedro.framework.hooks import hook_impl
from kedro.config import TemplatedConfigLoader
import click
from typing import Any

class CommandLineParamsHook:
    @hook_impl
    def add_cli_parameters(self, command: Any) -> None:
        # 仅给kedro run命令添加参数
        if command.name == "run":
            # 添加--input参数
            command.params.append(
                click.Option(
                    ["--input"],
                    type=str,
                    help="输入数据集路径",
                    required=False
                )
            )
            # 添加--output参数
            command.params.append(
                click.Option(
                    ["--output"],
                    type=str,
                    help="输出数据集存储路径",
                    required=False
                )
            )

    @hook_impl
    def after_context_created(self, context: Any) -> None:
        click_ctx = click.get_current_context(silent=True)
        # 非run命令不处理
        if not click_ctx or click_ctx.command.name != "run":
            return
        
        # 读取命令行传入的参数
        input_path = click_ctx.params.get("input")
        output_path = click_ctx.params.get("output")

        # 将参数注入到配置加载器的全局变量中
        if isinstance(context.config_loader, TemplatedConfigLoader):
            # 保留原有globals配置，仅覆盖传入的参数
            if input_path:
                context.config_loader.globals_dict["input_path"] = input_path
            if output_path:
                context.config_loader.globals_dict["output_path"] = output_path

3. 配置默认值（可选） #

你可以在conf/base/globals.yml中配置默认的路径，当命令行没有传入--input或--output参数时，会自动使用默认值：

input_path: "default/input/data.csv"
output_path: "default/output/result.csv"

4. 验证使用 #

直接使用你期望的命令格式运行即可，catalog中的${input_path}和${output_path}会自动被替换为命令行传入的路径：

kedro run --pipeline="my_pipeline" --input="path_to_input_1" --output="path_to_output_1"

如果需要支持多组输入输出，只需要参照上述逻辑新增对应的CLI参数和全局变量即可。

内容的提问来源于stack exchange，提问作者Isy89