Oh, sorry, and another question: I would like my project to allow running the pipelines on new datasets added by users to data/01_raw/custom-dataset. While I can just account for new datasets via dataset factories in the catalogue, actually using them in pipelines seems a bit more difficult, since the dynamic dataset name cannot be passed as pipeline parameter (as far as I know). If possible, I would like to avoid having users required to modify the conf/params, or even editing the source code for pipeline creation to add pipelines for their dataset. How can I do this in Kedro? Thanks you!

Hi @Bjarne Hiller, If I understand correctly you want to run a pipeline processing new datasets added to a folder (avoid processing existing datasets ?) and you do not want to manage passing parameters etc. Have you considered looking into IncrementalDatasets and PartitionedDatasets which are designed to tackle these usecases ? Thank you

Hi again @Ravi Kumar Pilla. I am familiear with

PartionedDatasets

, and I am not sure if they fit my use case exactly. What I want is basically to let the user run something like this

kedro run -p benchmark --params model=model_a --params dataset=dataset_b

. I guess I could use a PartionedDataset and a param to select the right loader for the pipeline - however, the node would then return results only for this dataset. Would I again return a partioneddataset? Also, it feels weirdly contradictionary to kedro's philosophy, as this is actually not really a

PartionedDataset

(i.e., data spread across multiple files), but a set of individual Datasets.

Hi @Bjarne Hiller, okay. You are right that the PartitionedDatasets do not fit in this usecase as they are basically 1 dataset split into multiple files but your case is having new datasets. I will get back to you shortly. Thanks for your patience !

Hi @Ravi Kumar Pilla, thank you for your reply. I am familiar with the article, and I am already using some of its techniques, like dataset factories and dynamic pipeline generation. The use case I have in particular is that other users might add more datasets to the project. I could avoid making them edit the data catalogue by using matching dataset factories. However, I cannot dynamically generate a new pipeline for this dataset, since I don't know the name of the new dataset at pipeline generation time, unless I access the DataCatalogue manually.

Hi @Bjarne Hiller, So if I am understanding this correctly, 1. You have a pipeline say user_ds_pipe 2. Everytime a user adds a new dataset, you want to run the user_ds_pipe taking in the new dataset 3. You are unaware of the dataset name to make this as input to your pipeline > since the dynamic dataset name cannot be passed as pipeline parameter Let me try this solution which seems close but the resolution might not work

Hi @Ravi Kumar Pilla, sorry for the late reply. I'm not sure if this is what I'm looking for. Since I have a lot of different datasets/models, the pipeline registry has a lot of entries which becomes confusing. Ideally, I think I would like users to pass a parameter `kedro run predict --params dataset=dataset_a", etc. However, this does not appear to be trivial to do in kedro, since I would need the dataset name during

register_pipeline

. Maybe something like this would work? 1. user adds a new dataset folder to

01_raw

2. It is picked up by the data catalogue via a dataset factory 3. The user configures a

dataset

parameter to select the dataset to run the pipelines on 4. A

after_context_created

hook reads the

dataset

parameter from the kedro context 5. The

dataset

parameter is passed down to

register_pipelines

, where it is injected as input to the

predict

pipeline Would this work? What do you think?

Hi @Bjarne Hiller, Worth a try. I am not sure. @Jitendra Gundaniya can you give this a try to see if this works ? I would suggest testing dynamic pipelines (if there is a chance to know the dataset name before hand), unless you are very particular on passing the dataset name as parameter. i.e., running something like

kedro run --pipeline predict_dataset_a

# src/my_project/pipeline_registry.py

from pathlib import Path
from kedro.pipeline import Pipeline, pipeline
from my_project.pipelines.predict import create_pipeline as create_predict_pipeline


def _discover_custom_datasets() -> list[str]:
    root = Path("data/01_raw/custom-dataset")
    if not root.exists():
        return []

    return sorted(
        p.name
        for p in root.iterdir()
        if p.is_dir()
    )


def register_pipelines() -> dict[str, Pipeline]:
    pipelines = {}

    for dataset_name in _discover_custom_datasets():
        pipelines[f"predict_{dataset_name}"] = pipeline(
            create_predict_pipeline(),
            namespace=f"predict_{dataset_name}",
            inputs={
                "raw_input": f"custom.{dataset_name}.raw",
            },
            outputs={
                "predictions": f"custom.{dataset_name}.predictions",
            },
            parameters={
                "params:predict": "params:predict",
            },
        )

    pipelines["__default__"] = sum(pipelines.values(), Pipeline([]))
    return pipelines

# conf/base/catalog.yml

"custom.{dataset_name}.raw":
  type: pandas.CSVDataset
  filepath: data/01_raw/custom-dataset/{dataset_name}/input.csv

"custom.{dataset_name}.predictions":
  type: pandas.ParquetDataset
  filepath: data/07_model_output/custom-dataset/{dataset_name}/predictions.parquet

# This gives you good lineage because the graph contains concrete datasets such as:
custom.dataset_a.raw
custom.dataset_a.predictions
custom.dataset_b.raw
custom.dataset_b.predictions

If you really want the --params way to do, may be create an additional node within the pipeline -

# src/my_project/pipelines/predict/pipeline.py

from kedro.pipeline import Pipeline, node


def create_pipeline(**kwargs) -> Pipeline:
    return Pipeline(
        [
            node(
                func=load_selected_dataset, # this node takes the param and selects the dataset to process
                inputs=["params:dataset"],
                outputs="selected_raw_input",
                name="load_selected_dataset",
            ),
            node(
                func=predict,
                inputs=["selected_raw_input", "model"],
                outputs="predictions",
                name="predict",
            ),
        ]
    )

Hey @Bjarne Hiller, I actually tried both of these and they both work. 1) Pick the dataset at run time (your idea): A small

after_context_created

hook reads the

dataset

param and hands it to

register_pipelines()

, which builds the pipeline for that one dataset:

kedro run -p predict --params dataset=dataset_b

This keeps your registry clean (just predict). Two things to watch out for: - On newer Kedro version, have the hook read

context.config_loader.runtime_params

, not

context.params

. Reading

context.params

loads the pipeline registry too early, and you'll get "Failed to find the pipeline named 'predict'". - Since the pipeline only exists once you pass the param, it won't show up in

kedro registry list

or kedro-viz. 2) One pipeline per folder (Ravi's idea):

register_pipelines()

scans

data/01_raw/custom-dataset/

and

creates a predict_<name>

pipeline for each folder, and the dataset factories handle the catalog:

kedro run -p predict_dataset_b     # just one
kedro run                          # all of them

Drop a new folder in and its pipeline shows up on its own, no code changes. Everything's visible in registry list and kedro-viz, and the lineage stays clean. The only real downside is that the registry gets long if you have a lot of datasets. Both cover the main thing you asked for: people add a folder and nobody has to touch code or config. But 1 is more on workaround side.