essentially you are just running the same pipeline with a different catalog, right?

yes exactly

The principle of static pipeline is easy to understand and debug - i.e. when you look at your

kedro-viz

diagram, you should be able to understand the flow

It depends on how complex the things that you are trying to do, it may works just with Jinja to generate the

catalog.yml

with some additional loops

For my usecase I want to run the same pipeline multiple times only changing the seed And then plot the results with a confidence interval I think for my case a post processing step will be better However in your case if you have multiple ingestion pipelines I would try to merge them I would create namespaced pipeline sets for each datasource that can run on their own (ex.

ingest.<ds>

) in the

<ds>

namespace with a

prediction

pipeline that sources them and then merge them together to a complete pipeline That way you can run all pipelines independently to make sure they are correct, access all the artifacts using ipython and view them in kedro viz

Look up the code I posted for the dataset hooks in 30 Aug for inspiration so that your catalog hook can always produce the same datasets, with versioning support

In my case I have a const.py file that lists the names of datasets and algorithms I want pipelines for It's good enough for what I'm doing

Hi @antheas , I've started with the refactoring for my multiple source pipelines, following the namespace approach that you described. I find the namespaced config parameters ("params:xyz") quite useful for this purpose, and I am curious if you are also using this and how you structure your configuration. More precisely, if I come back to your cars.cnn4.train example, do you follow the same namespace ordering (cars -> cnn4 -> train) or do you start the namespacing with the pipeline namespace, i.e. train -> cars -> cnn4? Personally I find that using the later makes the configuration structure cleaner, because each pipeline (ingest, feature engineering, train_eval, infer, ...) has it's own isolated parameter space. Thx

As for parameters for me it's more complicated. I don't use kedro for that anymore. I only use it to generate the top level dictionary parameters, then feed that to a custom data structure I made that holds my hyperparameters And I essentially namespace like this:

xml

alg:
    <insert your overrides here>

default:
    ...

<dataset>:

    tables:
        <table>:
            ....

    algs:
        <alg1>:
            ...
        <alg2>:
            ...

Each of my datasets gets a top level node in the xml file with its name and a full set of hyperparameters. In it, an

algs

tag allows tuning per algorithm This is in addition to the

default

tag, which provides the defaults for the project so I don't have to list them per dataset. I essentially merge the

default

dictionary with the

<dataset>

dictionary to form my hyperparameters when I start. I also allow for the top level to do overrides, so if I insert

alg.lr = 0.2

then the current algorithm will be sent the learning rate 0.2 With a custom cli I then do

kedro p cars.cnnv4.train alg.lr = 0.2

to run my project with a hyperparameter override each time. I also store an xml version of the hyperparameters in mlflow for future review If you want some code snippets for ex. the dictionary merging ask. In my code the following are merged to the same dict:

alg.lr: 0.2

alg:
    lr: 0.2

There are multiple ways to assemble the parameters/options based on the namespace, such as programmatically by building the catalog with code. An alternative which I have been trying is to use parameter mappings in pipelines factories such as:

def create_pipeline(
    ingestion_ns: str,
    features_ns: str,
    input: str,
    has_custom_options: bool = False,
) -> Pipeline:

    parameters = {
        f"{features_ns}.default_options": f"{features_ns}.default_options",
    }
    if not has_custom_options:
        parameters[f"{features_ns}.custom_options"] = f"{features_ns}.default_options"

    return pipeline(
        pipeline(
            pipeline(
                nodes,
                namespace=features_ns,
                inputs={
                    "input": "input",
                },
            ),
            inputs={
                "input": "input",
            },
            namespace=ingestion_ns,
            parameters=parameters,
        ),
        inputs={
            "input": input,
        },
        namespace="feature_engineering",
    )

In the above strategy, the nodes expect some default and custom options that are mapped inside the factory to the proper location in the parameter space. If there are no custom_options in the catalog then the default_options are mapped also to the custom_options. custom and default options are then merged in each nodes using a node decorator

So many ways to do it, it's hard to make a choice. I will try to use as much as possible the tools of the framework, but I am definitely interested in some code snippets if you have, they are worth thousands words 🙂

*now you might ask, feed a static value to a node with kedro? How to do that? I think that's one of the biggest oversights in kedro. Some times I need to pass in a fixed string to a node that's not based on a parameter and I can't. Use a lambda you say, well, two issues with that. 1) lambdas have no name so kedro viz will show $lambda as the node name. 2) lambdas can't be serialized, so ParallelRunner will crash. 3) If you use a

functools.partial

and you change the

__name__

it won't persist through serialization and deserialization, mutating the name of the node and crashing ParallelRunner. So I made

gen_closure

for that. Took a good afternoon to get all the issues sorted. Works just like

partial

does, but you can also feed in

_fn

to change the function name. I also included a little lazy load function that you can replace node function `import __`s with. Kedro, by making node funcs be actual functions, requires you load the node's module which might include pytorch, xgboost, jax, tf or . Making kedro help/ipython take 6s to load in my case (now 3), and if you use multiple gpu frameworks (one each run) probably causes ram issues/kedro to crash. So if you feel like optimizing your startup time (esp. with parallel runner where the main process doesn't need to load libraries/most processes don't need gpus), you can use that to help you.

@datajoely you might find some of this useful.

python
from functools import partial
from itertools import chain
from typing import Callable, TypeVar

from ...utils import get_params_for_pipe

A = TypeVar("A")


def list_unique(*args: list[A]) -> list[A]:
    return list(dict.fromkeys(chain(*args)))


class gen_closure(partial):
    """Creates a closure for function `fun`, by passing the positional arguments
    provided in this function to `fun` before the ones given to the function and
    by passing the sum of named arguments given to both functions.

    The closure retains the original function name. If desired, it can
    be renamed using the `_fn` parameter. If fn contains `%s`, it will be
    replaced with the function name"""

    def __new__(cls, func, /, *args, _fn: str | None = None, **keywords):
        self = super().__new__(cls, func, *args, **keywords)

        if _fn:
            self.__name__ = _fn.replace("%s", func.__name__)
        else:
            self.__name__ = func.__name__

        return self

def _lazy_execute(anchor: str, module: str, fun: str, *args, **kwargs):
    from importlib import import_module

    module = import_module(module, anchor)

    return getattr(module, fun)(*args, **kwargs)


def lazy_load(anchor, module: str, funs: list[str] | str):
    if isinstance(funs, str):
        return gen_closure(_lazy_execute, anchor, module, funs, _fn=funs)
    return (gen_closure(_lazy_execute, anchor, module, fun, _fn=fun) for fun in funs)

I guess we'd love to hear your rationale, the problem it solves and what you would like from kedro in the future if this was natively supported