Dynamic Pipeline

[noklam]

Introduction

We had a discussion about dynamic pipelines from #1993, also partly related to #1963, this issue is to summarise the discussion and lay out the work that we need to do.

Related Issues:

• Dynamic pipeline from current data #2626

A high-level, short overview of the problem(s) you are designing a solution for.

Background

Dynamic Pipeline has been one of the most asked questions, there are various solutions but often they are case-by-case. As a result the solutions comes with all fashion and it has been asked whether Kedro can provide a feature for that.

What is "Dynamic Pipeline"

When people are referring "Dynamic Pipeline", often they are talking about the same thing. We need to make a clear distinction between them before we start to build a solution for it.

We can roughly categorise them into 2 buckets

1. Dynamic construction of Pipeline
2. Dynamic behavior at runtime

Dynamic construction of Pipeline (easier)

Examples of these are:

1. Time series forecasting - Pipeline make prediction for Day 1, next pipeline requires Day 1 prediction as input.
2. Hyperparameters tuning
3. Combined variable length of features - feature engineering combine N features into 1 DataFrame
4. A list of countries - each need to be saved as a catalog entry, the data are then combined in a pipeline for further processing

Dynamic behavior at runtime (harder)

Examples of these are:

• 2nd order pipelines - pipelines generated from some node's output

  • I have a scenario that I would like to run a model training and model evaluation based on labels on dataset. Each Label
    would trigger an indiviual pipeline.

  • A pipeline that make prediction on 1 user, Fetch a list of N users, then run pipeline on each of them.

• Running node conditionally - Run A if B does not exist, otherwise run C

Why is it challenging for Kedro Users?

It needs experience with Kedro, often you need to combine advance features, i.e. TemplatedConfig + Jinja in Catalog + doing some for loop in your pipeline.py.

In addition, each of the use cases need different solutions. As part of the Kedro's value proposition is the standardisation. There are no well-known pattern for these solution, they are hard to reason and debug with Jinja

What's not in scope

• Non-DAG pipeline - i.e. Github Action, CircleCI type of pipeline.
• Skipping nodes - i.e. if A exist, don't run B and C (a workaround with hooks is possible)
• Dynamic node generation during a run

These two types of pipelines are fundamentally different from the "data-centric" approach of Kedro

What are some possible solutions?

Follow-up

• Create one or more blog posts about Kedro & "dynamic pipelines" that sets out the various requirements, solutions and links to docs kedro-devrel#7

Reference

• https://miro.com/app/board/uXjVMSX0s6s=/

[stichbury]

Thanks for this @noklam -- I was looking back at the technical discussion recently and wanted to flag it again, so I'm pleased you did. Would you be comfortable about writing this blog post? Maybe we can put it in a sprint in the next couple upcoming?

[noklam]

@stichbury Yes! I am happy to write something about this.

[desmond-dsouza]

The Miro board linked above seems to be not viewable by public?

[noklam]

@desmond-dsouza Sorry about that! You are right the link is private, as this is still in draft and there are some internal discussion in the Miro board. I will try to post some screenshot here when we discussed this.

Do you have any view on this? The board is mainly examples, I try to go through all the Slack & Discord question to summarise what are the user problem.

[astrojuanlu]

Another question on dynamic pipelines https://www.linen.dev/s/kedro/t/12647827/hello-thank-you-for-this-great-library-i-am-a-ds-working-in-#d170c710-8e6a-4c56-a623-058c3ec33da7

[etape 1] > [etape 2] > [if score_etape2 < X ] > [etape4]
                      > [if score_etape2 >= X ] > [etape5]

[noklam]

@astrojuanlu This sort of runtime generated DAGs is not supported currently. There are few possible solutions to get around this:

1. Embed the if-else condition inside a node
2. Use hooks or dynamic injection

We usually advise try to avoid these kind of conditional DAGs as much as possible, because it gets very complicated once you have multiple switches and it is difficult to debug. Having a conditional DAG is not much different from having a if-else node. i.e.

if score >= 0.5:
  return do_a()
else:
  return do_b()

The challenge here is the return dataset may be different or not compatible at all.

I haven't done it before, it should be more flexible to do this with the Python API. The code may look like this

# deployment_script.py
result = session.run(tag="some_pipeline")
if result["score"] >= 0.5:
  session.run(pipeline="deploy")
else:
  session.run(pipeline="retrain")

It does mean that you may lose certain feature and cannot use the Kedro CLI to run this, so use it sparingly.

[datajoely]

For me, if we're going to do this properly we need some sort of

• Conditional node
• Some pre-flight checks to validate things
• To push forward with the session store so that the reproducibility concerns can be managed by good logging/time-machine stuff
• [not entirely sure] A way of accessing the catalog in the pipeline registry

[noklam]

X-posting some solution our users is using.

1. Create a after_context_created after_catalog_created hook
2. Replace the register_pipelines() function with a custom register_dynamic_pipelines(catalog: DataCatalog) function
3. If you can pass the catalog to create pipelines, you can access datasets and parameters to dynamically build your pipeline!
   Use create_pipeline(catalog: DataCatalog) functions to create your pipelines !

Same Hook as above. But, instead of creating pipelines by changing the parameter and dataset names directly we create namespaced pipelines. Along with dataset factories: https://docs.kedro.org/en/stable/data/data_catalog.html#example-3-generalise-datasets-using-namespaces-into-one-dataset-factory - have a look at the catalog.yml.
This way , not only is the pipeline dynamic but also your catalog!

Detail solution can be found here.

[Lodewic]

@noklam To pitch in here, thank you for sharing my solution to this issue tracker! You asked for an example repository, you can find that here: https://github.com/Lodewic/kedro-dynamic-pipeline-hook-example

[inigohidalgo]

• [not entirely sure] A way of accessing the catalog in the pipeline registry

@datajoely we are currently looking for a workaround to be able to access parameters from the pipeline registry. In one of our projects (not sure exactly what kedro version was current when it was implemented, def <0.16) we had this as a workaround:

class ProjectContext(KedroContext):
    """Users can override the remaining methods from the parent class here,
    or create new ones (e.g. as required by plugins), indicate in settings.py
    """

    def _get_pipelines(self) -> Dict[str, Pipeline]:
        params = {} if self._extra_params is None else self._extra_params
        return create_pipelines(**params)

We are looking for a current "kedroic" way of handling this. I am looking into @Lodewic 's implementation to see how well it fits our usecase, it looks very promising, but a more out of the box solution might be preferable as it requires a certain amount of knowledge about kedro's inner workings which we probably shouldn't expect from general users.

[datajoely]

This is neat @inigohidalgo !

[noklam]

This look like the 0.16.x or 0.17.x style of creating pipeline, I actually don't know what happened and why we moved away from this. It used to be possible to access paramaters in create_pipeline, thus the template is still create_pipeine(**kwargs).