Passing entry to the pipeline via API body request

[Akarinokadze]

Hi everyone!

disclaimer: Sorry for not using slack, it's unavailable for me unfortunately..

I'm trying to make pilot microservise app that use docker and makes prediction via API.

Here is pipeline:

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                func=get_input,
                name='api_data',
                inputs='user_entry',
                outputs='output'
            )
        ]
    )

Here is catalog.yaml description:

user_entry:
  type: MemoryDataSet

As I seen on some YouTube videos, I'm trying to make it run trough separate runner:

from fastapi import FastAPI
import uvicorn
from kedro.framework.session import KedroSession
from kedro.framework.startup import bootstrap_project
from pathlib import Path


bootstrap_project(Path.cwd())
with KedroSession.create() as session:
    context = session.load_context()

    app = FastAPI()


    @app.get("/model")
    async def predictor(entry: str = 'some default text here'):

        context.catalog.save('user_entry', entry)
        output = session.run(pipeline_name='api')
        return output

    if __name__ == "__main__":
        uvicorn.run(app, host="127.0.0.1", port=8000)

Here is the problem: once I run the runner it runs server, get request, runs the pipeline, but pipeline cannot read user_entry from catalog with error:

kedro.io.core.DataSetError: Data for MemoryDataSet has not been saved yet.

What am I doing wrong?
I there some easier method to pass some value (from API for example) to a running pipeline?

p.s. process pipeline as I see it:

1. create nodes and pipelines that use stored model and a user_entry as a MemoryDataSet declared in catalog
2. get user inqury as get or post request
3. save user entry to declared earlier user_entry MemoryDataSet (or pass it to pipeline some other way?)
4. run pipeline and return answer by the same get/post request as answer

[noklam]

You may want to read the documentation about Data Catalog https://docs.kedro.org/en/stable/data/data_catalog.html

Can you share the link of your video? If your data is expecting data from an API, you can use the API Dataset for the request instead of declaring a MemoryDataSet(which isn't required at all, all dataset not defined in catalog are defaulted to be Memory Dataset)

[Akarinokadze]

You may want to read the documentation about Data Catalog https://docs.kedro.org/en/stable/data/data_catalog.html

Can you share the link of your video? If your data is expecting data from an API, you can use the API Dataset for the request instead of declaring a MemoryDataSet(which isn't required at all, all dataset not defined in catalog are defaulted to be Memory Dataset)

Hello! thanks for documentation, will read it, once again :)

Here is the video: https://www.youtube.com/watch?v=z7MIq-B4hPA&t=1s
As you may see he uses API call just to run evaluate pipeline, without passing anything via API.

As for MemoryDataSet, I've read about the need to declare it, but since it wasn't working, I've decided to make it explicitly :)

Regarding APIDataset, this was the first clue. In the proposed schema API will start the pipeline, but the dataset will attempt to read data only after request is already done, so it waits for another request through the same API, which will run another pipeline and so on. The problem is how to simultaneously run the pipeline at the API request AND pass body of this request to the pipeline inside the same request so that answer would include result of the pipeline.

After all I was quite confused that Kedro have a lot of instruments to train and evaluate, but where is the inference instruments?

[Akarinokadze]

Changed a little bit my initial scheme, and decided to try to pass inquiry via parameters at session creation.
Here is runner.py:

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from kedro.framework.session import KedroSession
from kedro.framework.startup import bootstrap_project
from pathlib import Path


class Entry(BaseModel):
    user_entry: str = 'Default entry text'


bootstrap_project(Path.cwd())
app = FastAPI()


@app.post("/")
async def get_category(entry: Entry):
    with KedroSession.create(
            extra_params={'inference_options': {'user_entry': entry.user_entry}},
            save_on_close=False
    ) as session:
        output = session.run(pipeline_name='api')
    return output

if __name__ == "__main__":
    uvicorn.run('runner:app', host="127.0.0.1", port=8000, reload=True, log_level='debug')

and initial parameters yml:

inference_options:
  user_entry: dummy entry

so at every creation of the session my inquiry pis passed to the pipeline and node:

from typing import Dict
import time
import logging

log = logging.getLogger(__name__)
start = time.time()
print(start)


def some_node(parameters: Dict):
    output = parameters['user_entry']
    return output


end = time.time()
print(end)

from kedro.pipeline import Pipeline, node, pipeline
from .nodes import some_node


def create_pipeline(**kwargs) -> Pipeline:
    return pipeline([
        node(
            func=some_node,
            inputs='params:inference_options',
            outputs='output from pipline',
            name='dummy_node',
        )
    ]
)

So basically it works. But now about the time issue.
To run 10 times it takes 0.8s according to the log:

2023-06-01 15:29:34,992 [uvicorn.access] INFO:     127.0.0.1:58453 - "POST / HTTP/1.1" 200 OK   **LAST OPERATION PRECEDING**
2023-06-01 15:29:35,045 - kedro.framework.session.session - INFO - Kedro project fast-api   **FIRST OPERATION OF THE CYCLE**
2023-06-01 15:29:35,061 - kedro.io.data_catalog - INFO - Loading data from 'params:inference_options' (MemoryDataSet)...
2023-06-01 15:29:35,061 - kedro.pipeline.node - INFO - Running node: dummy_node: some_node([params:inference_options]) -> [output from pipline]
2023-06-01 15:29:35,061 - kedro.io.data_catalog - INFO - Saving data to 'output from pipline' (MemoryDataSet)...
2023-06-01 15:29:35,061 - kedro.runner.sequential_runner - INFO - Completed 1 out of 1 tasks
2023-06-01 15:29:35,061 - kedro.runner.sequential_runner - INFO - Pipeline execution completed successfully.
2023-06-01 15:29:35,061 - kedro.io.data_catalog - INFO - Loading data from 'output from pipline' (MemoryDataSet)...
2023-06-01 15:29:35,077 [uvicorn.access] INFO:     127.0.0.1:58454 - "POST / HTTP/1.1" 200 OK   **LAST OPERATION OF THE CYCLE**
2023-06-01 15:29:35,114 - kedro.framework.session.session - INFO - Kedro project fast-api   **NEXT CYCLE**

As I can judge, about 52ms goes to create session, 16ms goes to read params from MemoryDataSet and 16 ms goes to close session.

Am I doing it right or is it work some other way? Is there some way to speed up the process?
Or.. Is there any need to do this after all and everything goes well enough?

p.s. this is my first Kedro service and I have no idea if it's OK or not.

[Akarinokadze]

Looks like I have the same issue as in API discussion.

There is no possibility to store context and pass it to KedroSession.create() method, and it's loaded every run. That's convenient at learning stage, but not at inference, just a waste of time.

So I would suggest it as an idea for improvement.

[astrojuanlu]

@Akarinokadze Thanks for sharing your thoughts openly here. Would you mind opening an issue about being able to store the context and pass it around, with a succint description of your use case? It will have more visibility than this discussion.