如何运行Kedro Pipeline中的算法？适配自定义数据集疑问

一、核心概念快速理解 #

• Node：Kedro 中的最小任务单元，本质是「业务函数+输入输出数据绑定」。比如一个数据清洗函数，绑定输入的原始数据集、输出清洗后的数据集，就构成一个 Node；模型训练函数绑定训练数据、输出训练好的模型，也是一个 Node。每个 Node 明确了「做什么」以及「数据从哪来、到哪去」。
• Pipeline：按依赖关系串联起来的 Node 集合。Kedro 会自动识别 Node 间的数据依赖，按顺序执行，避免重复运行无依赖变更的节点。
• Catalog：数据集的配置中心，定义了数据的存储路径、格式（CSV/Parquet 等）、加载参数。你的预处理数据集需要在这里配置，让 Kedro 能识别并调用。

二、正确运行算法的命令 #

终端命令（项目根目录下执行） #

• 查看项目所有 Pipeline：kedro pipeline list
• 运行完整 Pipeline：kedro run
• 运行指定 Pipeline：kedro run --pipeline <pipeline_name>
• 运行单个 Node：kedro run --nodes <node_name>

Jupyter Notebook 中运行 #

先初始化 Kedro 上下文，再执行任务：

from kedro.framework.startup import bootstrap_project
from pathlib import Path
from kedro.framework.session import KedroSession

# 确保当前路径是项目根目录
project_path = Path.cwd()
bootstrap_project(project_path)

# 运行完整 Pipeline
with KedroSession.create(project_path=project_path) as session:
    context = session.load_context()
    context.run()

# 或运行单个 Node
with KedroSession.create(project_path=project_path) as session:
    context = session.load_context()
    context.run(nodes=["<目标node名称>"])

三、适配自有预处理数据集的步骤 #

1. 配置 Catalog
   打开项目 conf/base/catalog.yml，添加你的数据集配置，示例：

my_preprocessed_data:
  type: pandas.CSVDataSet
  filepath: data/03_primary/my_preprocessed_data.csv
  load_args:
    index_col: 0

my_preprocessed_data 是数据集的唯一标识，后续 Node 会通过这个名称调用数据。

2. 修改 Pipeline 依赖
   找到项目中定义 Pipeline 的文件（一般在 src/<项目名>/pipelines/<pipeline名>/pipeline.py），找到算法对应的训练 Node，将输入替换为你的数据集标识：

# 原代码示例
node(
    func=train_model,
    inputs="original_train_data",
    outputs="trained_model",
    name="train_model_node"
)

# 修改后
node(
    func=train_model,
    inputs="my_preprocessed_data",
    outputs="trained_model",
    name="train_model_node"
)

3. 验证与测试

• 运行 kedro viz 打开可视化界面，检查 Node 间的依赖链路是否正确，确认你的数据集已关联到目标 Node。
• 先单独运行目标 Node：kedro run --nodes train_model_node，验证数据加载和算法运行是否正常，再逐步扩展到完整 Pipeline。

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