如何以编程方式在kedro 0.18中注册Kedro数据目录？

Question

由于各种原因(主要是动态构造文件路径的能力)，我喜欢编程定义数据目录，而不使用yaml文件来定义数据集。

DataCatalog(
    {"products": ParquetDataSet(filepath=f{PREFIX}/products.parquet") 
...
})

在kedro0.17中，曾经是通过register_catalog钩子简单地注册目录，以便在管道定义中使用它的数据集。

但是在0.18中，这个钩子是不存在的，并且在钩子RegistrationSpecs中没有规格说明节。

在kedro0.18中有什么方法替代register_catalog？

我搜索了文档，并试图在yaml文件中实现目录，但这不符合项目的要求。TemplatedConfigLoader可能是一个选项，但我不允许实现自定义逻辑，构造数据集的路径。

Answer 1

不确定这是否是您要寻找的确切功能，但我一直在使用after_context_created和after_catalog_created挂钩的组合，以编程方式添加数据集。

只需为所需的数据集创建一个"add“方法，并使用这些文档来查看需要什么args。在下面的示例中，我的数据在S3中，所以我创建了一个方法，从credentials.yml中获取我的creds并将它们传递给PickleDataSet。

import logging
from kedro.config import ConfigLoader
from kedro.framework.project import settings
from kedro.framework.hooks import hook_impl
from kedro.extras.datasets.pickle.pickle_dataset import PickleDataSet


class ProjectHooks:
    @property
    def _logger(self):
        return logging.getLogger(__name__)

    @hook_impl
    def after_context_created(self, context):
        self.project_path = context.project_path
        self._logger.info(f"Project path: {self.project_path}")

    def _get_credentials(self, key):
        conf_path = f"{self.project_path}/{settings.CONF_SOURCE}"
        conf_loader = ConfigLoader(conf_source=conf_path, env="local")
        return conf_loader.get("credentials*")[key]

    def add_pickle_dataset(self, name, folder, layer=None):
        self.catalog.add(
            data_set_name=name,
            data_set=PickleDataSet(
                filepath=f"s3://root/data/{folder}/{name}",
                credentials=self._get_credentials("dev_s3"),
            ),
            replace=True,
        )
        if layer:
            self.catalog.layers[layer].add(name)
        self._logger.info(f"Added dataset '{name}' to the data catalog.")

    @hook_impl
    def after_catalog_created(self, catalog):
        self.catalog = catalog
        datasets = self.catalog.load("params:datasets")
        for dataset in datasets:
            self.add_pickle_dataset(
                name=f"{dataset}",
                folder="07_model_output",
                layer="Model Output",
            )