Kedro自定义DataSet疑问：如何将Catalog属性传递至DataSet？

核心问题分析 #

你当前的错误源于对Kedro数据集的使用逻辑理解偏差：Kedro会自动从Catalog实例化数据集并注入到节点，不需要你手动创建RSSFeedExtract对象。

步骤1：修正自定义DataSet类 #

你的RSSFeedExtract类需要符合Kedro的DataSet规范：

• __init__方法接收url参数（Kedro会自动从Catalog传入）
• _save和_load方法必须以self作为第一个参数
• 只读数据集的_save要抛出明确的DataSetError

修正后的类代码：

from kedro.io import AbstractDataSet, DataSetError
import feedparser
from typing import Dict, Any

class RSSFeedExtract(AbstractDataSet):
    def __init__(self, url: str):
        self._url = url

    def _save(self, data: Any) -> None:
        # 只读数据集，禁止保存操作
        raise DataSetError("RSSFeedExtract is a read-only dataset")

    def _load(self) -> Dict[str, Any]:
        dictionary = feedparser.parse(self._url)
        # 在此添加你的数据预处理逻辑
        return dictionary

步骤2：修正节点函数 #

不需要手动实例化数据集，Kedro会将Catalog中配置的rss_feed_extract数据集实例注入到节点函数参数中，直接调用其load()方法即可：

from typing import Dict, Any

def extract_rss_feed(rss_feed_extract: RSSFeedExtract) -> Dict[str, Any]:
    raw_rss_data = rss_feed_extract.load()
    # 可在此对原始RSS数据做初步处理，返回给下一个节点
    return raw_rss_data

步骤3：修正流水线节点定义 #

节点的inputs必须指定Catalog中数据集的名称rss_feed_extract，这样Kedro才会自动完成数据集的实例化与注入：

from kedro.pipeline import node

node(
    func=extract_rss_feed,
    inputs='rss_feed_extract',
    outputs='rss_feed_for_transforming',
    name="extract_rss_feed",
),

关键逻辑说明 #

• Kedro的Catalog是数据集的配置中心，当你在节点inputs中指定数据集名称时，Kedro会根据Catalog里的配置（包括type和url参数）自动实例化对应的DataSet对象
• 你之前使用内置APIDataset无需手动传参，正是因为Kedro完成了实例化和注入的过程
• 该节点需要配置输入，因为rss_feed_extract是你的数据源，Kedro通过输入参数完成注入，而非让你手动创建对象

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