在Kedro中实现嵌套PartitionedDataset作为Catalog条目方案问询

问题解答 #

1. 是否可行？有无类似实现？
   可行。Kedro的Dataset体系具备良好扩展性，已有不少用户通过扩展PartitionedDataset或直接继承AbstractDataset实现了嵌套目录结构的自定义数据集。这种方式完全契合Kedro的设计逻辑，能将嵌套文件夹结构封装为catalog.yml中的固定条目，无需依赖动态配置生成。

2. 自定义类是否只需实现_load、_save和_describe？
   如果选择继承AbstractDataset，确实只需要实现这三个核心方法，这是Kedro自定义数据集的标准要求。若选择扩展PartitionedDataset，则需要基于原有类的逻辑，重写或补充处理嵌套层级的相关方法（比如分区识别、加载逻辑），但核心依然围绕数据的加载、保存与描述展开。

最终实现方案 #

我最终选择直接继承AbstractDataset来封装嵌套文件夹数据集，以下是核心的_load和_save方法代码：

_load方法（加载嵌套目录下的所有音频文件） #

def _load(self):
    # 获取主目录下的所有子文件夹（对应标签）
    subfolder_names = [
        subfolder_name
        for subfolder_name in os.listdir(self._mainfolderpath)
        if os.path.isdir(os.path.join(self._mainfolderpath, subfolder_name))
    ]

    wav_paths_dict = {}
    for subfolder_name in subfolder_names:
        subfolder_path = os.path.join(self._mainfolderpath, subfolder_name)
        wav_files = []
        # 遍历子文件夹下的所有WAV文件
        for root, dirs, files in os.walk(subfolder_path):
            for file in files:
                if file.lower().endswith('.wav'):
                    wav_file_path = os.path.join(root, file)
                    wav_file_name = os.path.split(wav_file_path)[-1].replace('.wav', '').replace('.WAV', '')
                    wav_files.append((wav_file_name, wav_file_path))
        wav_paths_dict[subfolder_name] = dict(wav_files)

    # 加载所有音频文件，返回{标签: {文件名: 音频数据}}的嵌套字典
    partitioned_dataset_dict = {}
    for subfolder_name, sub_dict in wav_paths_dict.items():
        partitioned_dataset = [
            (wav_file_name, SoundDataset(wav_file_path).load())
            for wav_file_name, wav_file_path in sub_dict.items()
        ]
        partitioned_dataset_dict[subfolder_name] = dict(partitioned_dataset)

    return partitioned_dataset_dict

_save方法（将嵌套字典数据保存为标签化的嵌套文件夹） #

def _save(self, subfolders_dictionary):
    # 清空并重建主目录
    if os.path.isdir(self._mainfolderpath):
        for root, dirs, files in os.walk(self._mainfolderpath, topdown=False):
            for name in files:
                os.remove(os.path.join(root, name))
            for name in dirs:
                os.rmdir(os.path.join(root, name))
        os.rmdir(self._mainfolderpath)
    os.mkdir(self._mainfolderpath)

    # 遍历每个标签对应的子文件夹，用PartitionedDataset保存音频文件
    for subfolder_name in subfolders_dictionary.keys():
        subfolder_path = os.path.join(self._mainfolderpath, subfolder_name)
        os.mkdir(os.path.normpath(subfolder_path))

        partitioned_dataset = PartitionedDataset(
            path=subfolder_path,
            dataset=SoundDataset,
            filename_suffix=".WAV",
        )
        partitioned_dataset.save(subfolders_dictionary[subfolder_name])

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