Galileo-Galilei
commented
4 years ago TL; DR: make sure that training_pipeline.outputs() and predict_pipeline.inputs() are identical (except for one value for the data to predict in the second set, which seems to be remote_new in your project.). Note: the namespace may have modified them and causes some interference and it might be a bug.
A bit of context: what is the pipeline_ml function for?
Unlike all the others functions of the plugin, this function is not intended for production: it packages a whole pipeline to make it easy to serve (in api or batch) in one command line. The goal is to share a pipeline very easily and to facilitate test and reuse between data scientists.
For production purpose, you need a tool to schedule / monitor the pipelines (like airflow for instance).
What is its goal?
Under the hood, the PipelineML class (the type of the output of pipeline_ml) is nothing but a kedro Pipeline (the "training" pipeline) which has access to another pipeline (the "predict" pipeline) (you can see it in the code here).
When the KedroPipelineHook detects that the pipeline you are trying to run (with kedro run) is not a standard Pipeline but a PipelineML, it tries to log it in mlflow with the KedroPipelineModel class which is actually the class that contains all the code logic.
This class is inspired from mlflow example on how to create a custom mlflow model. Basically, it is a class that implements a predict method which will be called automatically when serving the model, and a load_context method which manages the inputs your predict method needs. Inded, any logged model that respects this format can benefits from mlflow capabilites, including model service.
How does it works?
If you have read the KedroPipelineModel code described above, you can see that the KedroPipelineModel needs 2 elements to be logged in mlflow :
- the predict function (here, your kedro pipeline used for prediction)
- the data required for the predict function to work (here your lightgbm model, your imputer and a couple of parameters)
You can fill these arguments by hand but it can be a bit tedious and this is where the pipeline_ml comes into play: your predict pipeline is a higher order function, i.e. a function generated by the training pipeline. All the arguments your predict pipeline needs are generated by the training pipeline, and the pipeline_ml binds them together: it should have all the informations needed for logging. When the training pipeline is run, the associated "predict" is logged.
To do so, it applies the following process:
- the
pipeline_mlcheck thattraining_pipeline.outputs()=predict_pipeline.inputs()-{input_name}. This means that allpredict_pipeline.inputs()are outputs of a terminal node in thetraining_pipeline. - when the
PipelineMLwill be logged, aDataCatalogwith the corresponding items as keys will be stored. - the artifacts must be persisted locally (it is not needed to declare them as MlflowDataSet though, the plugin uploads them whatever if they are on the disk) and they will be uploaded to mlflow with the model to ensure they can be accessed.
How to fix your bug
The error comes from the fact that training_pipeline.outputs() and predict_pipeline.inputs() do not have the same entries.
It may either comes from the fact that you do not use the same name for entries in the model (fix that!) or that the namespace introduces a prefix that changes the entries names. In the second case, it may a bug to investigate (I confess I have not tried pipeline_ml with namespaces which are quite recent in Kedro). Your pipeline should look like this (I'm coding on the fly, the code is not tested):
full_pipeline= Pipeline(
[
node(
func=preprocess,
inputs=dict(data="raw_data"),
outputs="preprocessed_data",
tags=["training", "inference"]
),
node(
func=fit_imputer,
inputs=dict(data="preprocessed_data"),
outputs="imputer",
tags=["training"]
),
node(
func=transform_imputer,
inputs=dict(data="preprocessed_data", imputer="imputer"),
outputs="imputed_data",
tags=["training", "inference"]
),
node(
func=train_model,
inputs=dict(data="imputed_data"),
outputs="model",
tags=["training"]
),
node(
func=predict,
inputs=dict(data="imputed_data", model="model"),
outputs="predictions",
tags=["inference"]
),
]
)
### create the pipeline like you. Use kedro viz to see what's going on!
train_pipeline=full_pipeline.only_nodes_with_tags("training") # print the pipeline and check that train_pipeline.outputs={imputer, model}
predict_pipeline=full_pipeline.only_nodes_with_tags("inference") # print the pipeline and check that predict_pipeline.inputs={raw_data, imputer, model}
training = pipeline_ml(
training=train_pipeline,
inference=predict_pipeline,
input_name="raw_data"
)If you still struggle with the function and needs additional help, can you please share :
- the result of
training_pipeline.outputs(),predict_pipeline.inputs()andtraining.input_name(you must pass this argument) - the actual values of
training_pipelineandpredict_pipeline(i.e. the list of nodes). If you can show me code in a github repo it will make things easier to debug.
Additional infos
I will publish a detailed example in the docs this weekend, if you can wait until this it should make the explanation more clear I guess.
laurids-reichardt
Hi @Galileo-Galilei. As I mentioned in other issue, I'm working currently with integrating my training and inference pipelines with MLPipeline. Unfortunately I'm confused with handling inputs and outputs, I can't wrap my head around it.
Context
My training pipeline is built from three other pipelines:
de_pipeline(data engineering),fe_pipeline(feature engineering) andmd_pipeline(training aka. modeling).My inference pipeline is buit from the same pipelines but with predict argument which change their behavior (they're using previously saved models for imputer and prediction.
In my current implementation it looks like this:
My pipelines also getting as input parameters, obtained from kedro configuration (by that I mean
conf/base/parameters.yaml).When I'm trying to glue them together with:
and running my training pipeline I'm getting:
I'm understand the issue here, but I don't know how to proceed with that ("un-free" inputs which should be obtained (automatically?) using Kedro features). I would be glad for any tips.