Versioning in Kedro is great. Even though we have an alternative to use different root for KedroWings, I believe it would be much better to support versions out of the box.

We can do that by adding extra parameter to KedroWings, namely versioned: bool = False. When it set True, we can update the KedroWings._dataset_configs. I've tried it by subclassing from KedroWings and it works fine.

First of all, thank you for the great and simple way to checkpoint my pipelines! You already saved a lot of my time.

Now, according to the current implementation, Kedro Wings transforms datasets only for the pipeline which was requested to run. This is okay in the most of the cases. However, it would be better to process datasets from all pipelines.

For example:

• it will allow to use "fixed" datasets inside the kedro ipython
• or to see them in kedro viz
• this also works better with %reload_kedro IPython/Jupyter magic.

At the moment we can hack our way by inheriting from KedroWings, replacing before_pipeline_run to suppress this hook and creating a hook for after_catalog_created:

class EarlyBird(KedroWings):
    def __init__(self, *args, **kwargs):
        super(EarlyBird, self).__init__(*args, **kwargs)
        self._context: Optional[KedroContext] = None

    def set_context(self, context: KedroContext):
        self._context = context

    def before_pipeline_run(
            self, run_params: Dict, pipeline: Pipeline, catalog: DataCatalog, name: str = None,
    ):
        log.info(f'Adding pipeline "{name}" checkpoints with Kedro Wings...')
        super(EarlyBird, self).before_pipeline_run(run_params, pipeline, catalog)

    @hook_impl
    def after_catalog_created(self, catalog: DataCatalog, *args, **kwargs):
        for name, pipeline in self._context.pipelines.items():
            self.before_pipeline_run(None, pipeline, catalog, name)

However, the problem with this solution is that we need to pass context to the plugin. For example (in settings.py for Kedro 0.17.0):

class ProjectContext(KedroContext):
    def __init__(self, *args, **kwargs):
        super(ProjectContext, self).__init__(*args, **kwargs)
        load_dotenv()

        for item in HOOKS:   # in Kedro 0.17.0 hooks are defined as a config variable in the `settings.py`
            if isinstance(item, EarlyBird):
                item.set_context(self)

I personally not very happy with this solution because it breaks Framework API and may cause problems with next Kedro versions (I've just upgraded from 0.16.6 to 0.17.0 without much pain but who knows).

For now we either define "global" root for all checkpoints or have to specify separate root for each type of checkpoints. This is okay but sometimes we need to transfer data between locations.

Possible scenarios:

• checkpoint some of the datasets locally
• or publish reports to the different S3 bucket

For now to support such behavior I am overriding datasets in the catalog. But I believe we can make it much better by introducing locations and tag inputs/outputs with location tags. For example:

• 01_raw/base.csv will be check-pointed in the default root location.
• while 01_raw/base.csv#reports will be saved to the reports location.

I am not sure whether #* suffix is the best API though. Looks like the possible interference with dataset transcoding convention should be considered.

There is a handy feature in Kedro: special tags for datasets according to the data engineering convention. It is quite useful in combination with kedro viz or (as in my case) for creating a UI on top of Kedro and sorting/filtering datasets according to their position in a pipeline.

I think we need to support this feature and provide some way to specify the layer for pipeline checkpoints.

Currently I am subclassing KedroWings and use simple rules based on the first two letters of dataset locations:

class ExplicitLark(KedroWings):
    LAYERS: Dict[str, str] = {
        '01': 'raw',
        '02': 'intermediate',
        '03': 'primary',
        '04': 'feature',
        '05': 'model_input',
        '06': 'model',
        '07': 'model_output',
        '08': 'reporting',
    }

   @hook_impl    
   def before_pipeline_run(
            self, run_params: Dict, pipeline: Pipeline, catalog: DataCatalog, name: str = None,
    ):
        super(EarlyBird, self).before_pipeline_run(run_params, pipeline, catalog)
        self._update_layers(catalog)

    def _update_layers(self, catalog: DataCatalog):
        for dataset_name in catalog.list(regex_search=r'^\d+_.*'):
            layer_code = dataset_name[:2]
            if layer_code in self.LAYERS:
                layer_name = self.LAYERS[layer_code]
                catalog.layers[layer_name] = catalog.layers.get(layer_name, set())
                catalog.layers[layer_name].add(dataset_name)

I think we can provide additional parameter to the KedroWings which accepts dictionary with regular expression -> layer name and use the default convention for XX_* datasets.

Hello,

Thanks for the great plug-in! I am trying to figure out how to save different model outputs into different folders. I followed your tutorial for generating pipelines using modular pipelines and separate namespaces .

 node(
   func=inference_single_model,
   inputs=["test_subset_s3uris", "params:model_name"],
   outputs="07_model_output/testset_predictions.csv")  # not in catalog; Kedro-Wings hook handles saving

However, the namspace prefixes the pipeline_key in front of outputs, so instead of being stored in data/07_model_output/MyPipelineKey.testset_predictions.csv it is saved in data/MyPipelineKey.07_model_output/testset_predictions.csv.