如何自动生成Kedro流水线？复现方法遇参数错误求助

问题背景 #

参考DataEngineerOne的《How To Use a Parameter Range to Generate Pipelines Automatically》视频，实现Kedro流水线自动生成，针对带通滤波器多个中心频率执行网格搜索，每个频率对应运行simulate流水线。已完成pipeline_registry.py、grid_search/pipeline.py、grid_search.yml及grid_search/nodes.py代码编写，但执行kedro run --pipeline grd时触发错误：

ValueError: Pipeline input(s) {'params:pipeline_400000'} not found in the DataCatalog

按视频内容，这类参数应作为内存数据集处理，但实际报错。需明确是否因Kedro版本更新导致差异，是否必须在catalog.yml中逐个定义参数（失去自动化意义），并寻求问题原因及解决方法。

问题原因 #

核心原因是Kedro版本迭代后，动态生成的params:前缀参数不再默认被DataCatalog识别为内存数据集。旧版本（如0.17.x及更早）中，params:前缀的动态参数会自动被当作内存数据集处理，但Kedro 0.18+对DataCatalog的输入验证更严格，要求所有流水线输入必须在Catalog中有明确注册或配置定义。

解决方法 #

1. 动态注册内存参数数据集 #

在流水线构建逻辑中，为每个目标频率创建对应的MemoryDataSet并注册到DataCatalog，确保DataCatalog能识别动态生成的参数输入。示例代码如下：

from kedro.io import DataCatalog, MemoryDataSet
from kedro.pipeline import Pipeline
from .grid_search.pipeline import create_simulate_pipeline

def register_pipelines():
    frequencies = [400000, 500000, 600000]  # 可从grid_search.yml读取
    grid_catalog = DataCatalog()
    sub_pipelines = []

    for freq in frequencies:
        # 生成动态参数名称并注册内存数据集
        param_key = f"params:pipeline_{freq}"
        grid_catalog.add(param_key, MemoryDataSet({"center_frequency": freq}))
        # 构建对应频率的子流水线
        sub_pipelines.append(create_simulate_pipeline(freq))

    grid_pipeline = Pipeline(sub_pipelines)
    return {
        "grd": grid_pipeline,
        "__default__": grid_pipeline
    }, grid_catalog  # 新版本Kedro支持返回流水线与对应Catalog

2. 用参数命名空间批量配置 #

通过grid_search.yml的参数命名空间批量定义参数模板，避免逐个在catalog.yml中配置：

parameters:
  grid_search:
    frequencies: [400000, 500000, 600000]
    pipeline_params:
      "{{freq}}":
        center_frequency: "{{freq}}"

在流水线代码中，通过参数解析逻辑将模板参数映射到动态生成的流水线输入，示例如下：

from kedro.config import ConfigLoader
from kedro.framework.project import settings

def create_grid_search_pipeline():
    config_loader = ConfigLoader(settings.CONFIG_SOURCE)
    grid_params = config_loader.get("grid_search*")["grid_search"]
    frequencies = grid_params["frequencies"]
    sub_pipelines = []

    for freq in frequencies:
        # 引用命名空间中的参数
        param_key = f"params:grid_search.pipeline_params.{freq}"
        sub_pipeline = create_simulate_pipeline(param_key)
        sub_pipelines.append(sub_pipeline)
    
    return Pipeline(sub_pipelines)

这种方式无需手动逐个定义参数，保持自动化灵活性，同时符合新版本Kedro的参数管理规范。

3. 版本回退（不推荐） #

若需完全沿用视频中的旧逻辑，可降级到Kedro 0.17.x版本，但该方式会错过新版本的功能迭代与安全修复，仅作为临时应急方案。

关键注意事项 #

• 动态生成的参数必须以params:前缀开头，符合Kedro的参数命名规范。
• Kedro 0.18+中，所有流水线输入必须被DataCatalog明确识别，动态生成的数据集需通过代码注册或配置预定义的方式加入Catalog。

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