MLOps with FastAPI, RabbitMQ and Kubernetes using Skipper

[Music] hey uh hello everyone very nice to see you here and let's start this session my name is andree and i'll talk about envelopes with fast api rabbit mq and kubernetes okay so agenda what you should expect from this session first i'll introduce myself in few more details so that you would have a bit more context then i'll talk about skipper this is our open source tool that we are developing and i'll explain why we are developing and why it's maybe useful for you and i'll talk in general in general about envelopes micro services then we'll see live demo and if there'll be still time left i'll go through the code and to try to explain the main concepts and also talk about infrastructure okay so a few things about me because um yeah i'm kind of it's my first time talking on python conference in lithuania uh so first of all um i'm from lithuania from vilnius uh back in 2007 i finished venus university math and informatics faculty since then actually i joined local company which was called maybe not very creative but it was uh literally called information technology it was the name of the company uh it doesn't exist anymore it was acquired by a few other companies but it was great company and the focus was on oracle and this is how my it career started uh we focus on oracle and then from like 2005 to 2018 i was focused on oracle and um then i started my own company red samurai with oracle consulting and was traveling uh a lot in different places and uh mostly working abroad not not lithuania but i was still based here but then in 2018 i thought okay maybe i need to change and i decided to completely change the way i work i still manage oracle business but myself i don't do any oracle work anymore and i switch completely to open source and python and the main reason for that was because python is very popular language and but obviously it's not not the main reason but the main reason is that there are lots of uh open source libraries available to implement ml solutions with python and we decided to build our own product yeah so that's this what we're doing now and i'm trying to because there are lots of open source uh codes uh tutorials videos about python and this is what i'm this is how i'm learning about python but at the end i'm also trying to give back to the community and i have my youtube channel and whatever i learned during week or whatever i implement during week i try to publish online as video or as a blog or some sample code because based on my experience working with java and oracle it works great when you contribute back to the community and this allows you also to learn something new when you actually contribute back because when you build a sample or create some tutorial at this time at the same time you're not only teaching or sharing knowledge but also you're learning yourself as well okay so this is just a few words about me and the main topic of the session skipper so yeah we decided to build our own ml product and then we started to think like how okay we'll build this product but how we would run it in development and in production and we did the research uh spent quite so quite a lot of time and we saw that there are existing solutions like a cube flow air flow that this solution is allowed to run containers either on premise or on cloud and you could orchestrate your ml workflow or ml pipeline and they were great but then still i was thinking okay those those solutions are kind of heavy and if you want to be to build something lighter more flexible more configurable you would need to spend a lot of time in researching um like a q flow or any other solution and kubeflow is very heavy in terms of resources also so if you want to test it on premise not on cloud then you need a very powerful machine lots of gigabytes of ram and so on so i thought let's uh let's uh take those open source libraries that are available and let's combine them together and build some sort of a template um i would not call it like a product it's more like a template with a predefined set of scripts and instructions that anyone could take and anyone could modify containers that are included and add their own containers remove containers and quickly spin up on premise or on cloud and it should run either with python virtual environment with different set of virtual environments or should run with plain docker with different containers or if you need more scalability it also can run on kubernetes either on local one or on any cloud vendor and it's completely open source on github and this is for us this is the infrastructure that we are using to build our products that some some use cases this is like a foundation and anyone can use it as well and to give you high level architecture overview about what i'm what i'm talking about so uh skipper and this uh mlops i think is focused on microservices and why microservices if you talk about a machine learning domain uh very often you would see developers they go and they create a jupiter or call up notebook they do research they build model they transform data the model is great the model is trained very well and then they run predict functionality everything is working fine and they end up having like one big chunk of code which works fine but then as as time comes to deploy it or integrate with the main system they face a challenge that it's kind of not flexible and even worse it comes later when you need to do maintenance and if uh data each structure is changing continue to go and update the logic uh to make sure that data parsing is updated and so on then it's the risk comes that you make may break entire pipeline and so on and this is actually we faced ourselves when for one of the customers we were building uh some forecasting solution based on ml models we had one big monolith application and at the end when uh data structure was changed and we needed to update it after half a year it was a big challenge to find the place where change needs to be done and so on so after that i learned that uh to build proper ml solution we need to use uh different containers different services and then the word microservice doesn't mean that service should be like very small no it can be big it can be like some data transformation logic which [Music] maybe have a fair amount of code but the main thing you should split a code based on a use case so for example data transformation preparation logic should be one container then model training should be another and model serving and inference prediction should be another one and this would allow to maintain the code better and easier so yeah then we started to look into infrastructure as they said and we created three different services data training and serving but then we need to communicate somehow between those services for example when requests would come to train the model then this container which trains the model it somehow needs to ask a container which creates data please give me the data and send me the data i'll use this data to train the model so this means there should be some communication between services so there are multiple patterns in microservices how you implement communication between services and i think what at least in our case the optimal one was to use rabbitmq there is also kafka but i kind of uh i prefer rabbitmq because of very simple and clear python interface and on top of existing rabbitmq python interface we created our own wrapper skipper leap which makes a calls to rabbitmq very generic and if in future someone who would be using skipper would like to use kafka for example they could replace rabbit cube with kafka and [Music] everywhere in all the containers where api calls are made for communication layer the code would stay the same because skipper leap hides the specifics of rabbit mq python api okay so communication is done through rabbitmcube between all the containers and we have on the left side you see genetic engine and this this block is also based on multiple containers all the yellow boxes are separate actual docker containers as you see here so this central part is fast api and by the way the creator of fast api here is in the room so thanks thanks for that and fast api allows to easily implement rest endpoints and it's a very light tool doesn't need lots of resources and it's very flexible it comes with a very nice ui which displays the rest api structure also you can do groupings if you have a rest api you expose rest api and you you know that there'll be certain groups so you can group the group create groups for to improve readability of your api and these engines in front as http proxy to surf api to the outside then this salary com container which helps to implement asynchronous to process asynchronous requests this means when request comes to fast api and if you know that this request is like for training for model training which is long long one then we execute this request through salary we put it into the salary queue and that means that salary would take care for the request execution in a synonymous way when a request will be completed then salary internal storage will be updated and it will have the information that this task with this specific idea is completed and at any point of time we can check the status of the task and see if it's completed or not then there's a workflow and logger those two uh custom components and if someone is using skipper it's uh he's free he can either use this this workload and logger or uh they can replace with their own components uh the idea of workflow that it should have a logic which would have would be able to understand if this specific task is should be executed asynchronously or not for example and logger provides the end point where you could submit any action from from any of the containers and basically lock them into the central place yeah and all of this runs as docker containers or on kubernetes yeah so this uh yeah just to summarize so the engine um containers engines act as http proxy then there's a fast api as a rest api controller for workload implements navigation then there is a salary for asynchronous uh course processing and logging to lock all the events and all this stuff is implemented with python then we have communication container which we're using a standard image of rabbitmq in a separate container and this wrapper skipper lip which is a reusable library published on python library index and also this another version of implement implemented with javascript is published on npm and it allows to have to use different languages in in this infrastructure like python or javascript so for example your model training uh code can be implemented with javascript on node.js for example and you can still plug in this container into this infrastructure that implemented with python and there's a sample set of microservices for data training and serving and those are created just as example to show you how the infrastructure works and obviously in your use case you would need to replace that with your own components but you can follow the example and see how api calls implemented how the scripts are set up and you could add your own containers or use existing ones so it's up to you okay and all the code is on the github and it's implemented as a it's set up as public template and the idea is that you could either clone it if you want or you could create your own repository and you can map your repository with the template and then in future when you will see that there are some changes uh we implement in this main template you can if you if you want you can sync them uh into your own repository and uh this way you could you could you could stay up to date because if you would clone it then you you on its own and you you would not get updates that we would implement in the future but it's completely up to you so in terms of infrastructure as i already mentioned that to just to repeat it to be more clear that there are multiple runtime options uh first one you could run all those containers as a separate processes on your own computer with virtual environments another option you could use docker containers or kubernetes but there are some specific cases where if you your model for example you want to train it on gpu then it's not so so simple to do it through the docker container because when you run uh code inside docker container it's tricky to leverage and utilize gpu on on your main machine so you may uh like model training you may run for example not on docker container directly on your own machine but communication with other containers still would work because rabbitmq runs as independent container and for rabbitmq there is no difference where from where the request comes either from another container or fro from the code the transfer on locally on the machine yeah and as i said that this skipper library of wrapper library for rabbitmq api is is published on python library index and npm and you can install it directly from from there also all this all this this template and all the scripts are tested with different cloud vendors and uh like with google cloud engine and with oracle uh i didn't test it yet with amazon but the reason for that is because i prefer to run kubernetes on google it's for me it's somehow easier and more flexible to run it on google okay then let's let's see how it works and most of the session will be not with slides but i actually will show the stuff so yeah just to give you a heads up what i'll be showing um uh like kubernetes infrastructure setup uh probably a bit and uh also show how asynchronous api call is executed and handled by fast api and salary then obviously rabbitmq runs and it orchestrates all the events across spots and containers so this is included as well we will see rabbitmq dashboard and we'll try to execute multiple request requests very quickly to make sure that asynchronous requests are executed executing in parallel and uh interesting part will be kubernetes port scaling i'll scale one of the ports to multiple instances and we'll execute few requests and we'll see that first request handled by one port second request is handled by another port so we have uh some sort of clustering and yeah if there'll be still time left i'll show you the source code and it'll explain main main things about the source code okay i will switch to the desktop and i should record the screen because later this will be this session will be available online and the the screen recording will be included so let's okay recording is on so first of all uh i'll run everything through common line i think the font is visible i was trying to increase it so it should be visible from from the back as well and the first thing um this demo i'm running on local environment and i run it on docker on docker desktop so it's both engines are up and running docker and kubernetes there are multiple ways how you could run kubernetes locally but i prefer docker desktop because it simplifies usage of the docker images because there are other engines but but then you would need to publish your docker images on some on some external for example regis docker registry and consume from there but if you're using docker desktop then kubernetes that runs inside docker desktop it out of the box understands and it it can see docker images that you created inside docker obviously okay so first first of all uh to to able to access rabbitmq dashboard i open proxy connection to the rabbit mq and then i could go over here and test yeah the page is accessible and that's because infrastructure runs in kubernetes and by default uh you don't have uh external communication to the to kubernetes port from the outside so you need for the development purpose you need to explicitly open the board rabbitmq inside kubernetes port for for this port to be accessible from the outside okay then we can check that we can double check that infrastructure is running and what we have here is yeah by the way maybe first i should go to the github and just briefly show you the structure of the skipper so all the all the environments uh all the infrastructure that i'm showing to you it's basic it's very easy to start because all the scripts are available and whatever i'm showing to you is you can spin up for example if you go here you can see that there are instructions for docker compose or for kubernetes so what this is what i'm using here now and in case of kubernetes you install engines ingress controller then you build images with docker and then you run kubernetes setup script which creates everything all the posts all the infrastructure and then api becomes fast api application is accessible on this internal url and yeah so this is the main point of skipper is is that it's light simple and you don't have to do any complex things to spin up your containers you just run script and so on and if you want to change it's easy you go to the script and change it okay so in terms of infrastructure we have now different kubernetes spots and kubernetes port is the simplest the smallest object in kubernetes that you can have and kubernetes spot can run either one or multiple containers and those containers are created from docker images so um as we see here most of the ports they they run single container like fast this is fast api port this is cell report this is a log report then there's a workflow training service have it's a single port with two containers and kubernetes have a very nice concept of a sidecar container and sidecar container concept means that in a single port you can run main container and helper container and it's very convenient because when two containers run in the same port they share the same infrastructure the same network the same uh storage so in this in this case uh helper container uh what does it it's managing the model which is trained so when main container trains the model builds the model file the helper container takes this model file it's it's it makes the serialized copy of this model file and it's posting uh this model to the rabbitmq so that all the subscribers that need to get this model file they'll get it and this works when model file is small when probably work on tabular data but if you build work with complex models or you're fine tuning some models like from hugging phase they're usually huge so then it would not work to pause the model uh through rabbit mq and distribute it through rabbit mq message to other containers then you would need to use some external storage either on premise or or on cloud and you would post the model to external storage so that all other uh containers would consume this model file not through internal rabbitmq messaging system but from from the external file file storage okay and we have spinned up here too to pause for the serving service because usually in real life the the model training would execute uh you would not run multiple model trainings at the same time probably but when model training model is created uh you need to serve this model to the public to the users to your api to the to the application and at the same time model could be used by multiple people by multiple sessions and you need some scalability here so for that reason uh in default example of skipper i included two instances of serving port just to show how uh request is being executed by first port and by second port and this way the clustering works okay so now let's open logs so late in the next step when i when i execute api call you would see the information printed on the log and it would make sense how the request is being uh invoked so this one is [Music] look for a fast api application then over here we have a lock for a salary container which as i said earlier allows to implement asynchronous requests okay then this is port for the model training and over here we have first port for the server for the serving service and just i should double check that the instance name is correct it ends with x double x dv so yes this one is correct okay it's up and running and is for the second one is tr yeah that's correct one it's over here and the last the last lock window would be okay i think i closed my mistake so let's let's open let's copy them let's make it wider okay this one was closed okay this one training and this this one is serving first okay this is the okay serving seconds fine no actually this one we don't need and this one is for mobile net okay so all the locks are open seems like now we can go and see if fast api in general fast api endpoint is responding so we go to this default ui page that's out of the box generated by fast api tool we refresh it and this [Music] like very basic endpoint to test if infrastructure is running in general so you get it out of the box from skipper so we click execute and we see that api is running that's fine and we go to to this lock window and if i clean it up and execute few more times we see that the lock is printed so the port which is responsible to run fast api application is responding uh fine okay so the next the next thing would be we need to train the model if you want to allow users to run inference then we should have a model and we train the model in this in in default skipper example i'm using boston housing data it's a small data set just for example purposes is not supposed to be some complex ml model which trains for a long time it trains just in a few iterations very quickly and the moment the main point is to show that the model trains and it's not uh there is no point to show some sophisticated model and by the way model training is done with tensorflow and keras but in the same way you could install pytorch and build models with python if you want doesn't matter so we go over here and to say that we use eighty percent of the data for model training and twenty percent for uh validation and we execute the request we get we get back task id and if you go to to the salary lock we see that task is already completed this is on purpose i'm not running very long model training just because i want to save time mainly for during presentation and but this request is asynchronous because the idea of model training that probably it would run long and this is the reason why when we execute the request we get back task id in response we don't get information that model is trained for example because model is supposed to train maybe for half an hour for for one hour and for so far and so on so i can take this id and i could scroll this another standard endpoint from skipper that allows to check the status of task based on this on the id of the task okay i'd copy paste the id execute and get the status task completed if the task would be still running then it would be a task in progress okay and now i could go and simulate a couple of requests in parallel so for that let me go and clean up the lock for the salary window and by the way uh before we run uh if i switch to another lock uh where training port is running we see that lock was printed out for model training and this means that training was actually executed in the spot and there was no errors so now i cleaned up log for the salary window and i go and click on execute multiple times quickly like one two three four and then we go to the salary and all those tasks are executed in the back end asynchronously response was returned quickly to the fast api from salary without blocking them without blocking the original response and this is in particular important for web applications because when user runs a long task user don't want to wait until like this request maybe for 10 minutes maybe for half an hour would execute and user don't want to see like a blob ui blocked or whatever so for that reason you want to implement asynchronous request and return response instantly and you want to let salary to handle the request in the back end and later when request will be completed you may have some ui logic or backend logic which checks uh which polls for example or maybe some through the web socket you may check task status id and inform the user that task was completed and then user can come and see the results so now we can see that i clicked like four times to execute and we get back uh four four statuses returned in the back to show that all those four tasks were completed successfully okay so we trained the model and if we check we check the rabbit mq we see that under cues section then we have like a q name for skipper training serving mobile net mobile net data and so on this means that when request comes to fast api and then it's routed to if it's a synchronous one it's routed to salary and from salary we call rabbit in cuba send the message that uh hey we want to execute model training and then rabbitmq goes and checks who are the subscribers for this message if there are any subscribers for this queue and it finds that this port training model training port actually subscribes to the queue and rabbitmq delivers message to that port and in case uh what is great about rabbitmq is it supports like clustering out of the box uh message distribution to multiple subscribers so if you would decide to spin up training port and create two two instances three instances or more uh then uh rabbit when each instance is created rabbitmq would automatically get information that there is a new subscriber and by default when there are multiple subscribers to the same message when the message arrives rabbitmq would distribute it to the first subscriber then when the second message comes it would go to the second one and to the third one and so on so it makes round robin out of the box and this allows to distribute your load on application evenly and there are some use cases when you want to like fan out the message to all the subscribers at once so when message comes you need all the subscribers at the same time to get the message and then you just need to change the property of rabbitmq for this specific queue for the specific queue and say that you want to use fun out mode and then it will be distributed to all the subscribers at the same time and this uh rabbit mq functionality plays very well with kubernetes because on kubernetes it's easier to spin up uh port to create more instances or reduce instances and since rabbitmq automatically is able to track all the subscribers you don't need to do any extra job any extra implementation to support kubernetes the way how kubernetes scales because when new instance of kubernetes support will be created automatically rabbitmq will register a new subscriber and will implement will allow communication to with that subscriber so it comes out of the box okay so now we train the model so the next step would be to try and use the model and this is a separate endpoint called predict for that we click try it out and before i would execute it i would go to uh to the to the lock which displays output from the first port and this uh second uh second window which displays output from the from the second port so let's clean it up and if i click execute i get back the result now prediction was executed the model which was trained was used to process this payload and this is very it's a sample payload you should again it's created just for demonstration purpose and the idea of the model that it should try to predict the house in boston based on some historical data based on and data is constructed on different parameters like a number of rooms of the house the when the house was constructed and so on and this is done just for demonstration purpose and i don't have a goal to execute proper prediction here but whatever it returns it returns some price and some ratio over here and then if you go back to the lock we see that in this case this request was handled by the second port because we lock was printed out when this model prediction was executed it was printed in the second port and if you click once again then uh it gets uh the lock also printed in on the first port this means first request was handled by the first one by the second port and the second request was handled by the first port so this load distribution is working as is expected okay and this one this another service it runs with javascript with node.js and using tensorflow.js and tensorflow.js allows to implement model training and inference with javascript and it runs in a separate container and the reason why i included it into this default skipper template is to show that we could also integrate containers with javascript and they would run could run in the same infrastructure which runs on top of python ok so let me close this one and let's open this last endpoint and i executed and if i go back to look we see that request arrived and what this model does is using standard mobile net model it gets image and it tries to predict what is the in image and in this sample there is some picture with orange and it returns probability that 98 percent this is orange okay but now if we look into the infrastructure we see uh that this one port for this mobile net service and what we could try to do now we can we could scale scale the number of ports to two and would have like two ports to handle mobile net so for that let's let's go to the let's copy the instruction how to scale up the port we go to the folder which implements service a sample set of services and we go specifically to mobile nut service service and each service in skipper have a readme file and this is created for that reason because maybe you would like to run just one service and see how it works and you don't want to want to run entire infrastructure so you could go to the actual service and see the steps over here so what you want to do now you want to scale mobile net service to multiple instances and with kubernetes is done very easily just one line of code we provide the name of the of the port mobile service and we say how many replicas we want to have in this case we want to have two so we execute and it says that service the port was scaled and we could review the in the infrastructure and we see that new port was created just now and it's already up and running okay then we could go and open one more lock window which would display the lock for the second mobile net port for that we need to copy the id of the port which was created just now this is the id okay we'll go over here delete the old id which was created before okay it's up and running i can clear up the lock and now if i go back to fast api endpoint i click execute once and we see that this request was handled by the first port there is nothing printed in the second one and we click once again then lock is printed in the second port and this request handled by the first one if you click one more time then this third call is handled again again by the first one so this is default route round robin messaging from rabbitmq because we created a second port second subscriber as soon as this second port was created it registered itself with uh rabbit mq rabbit mq now uh it's informed that there are two subscribers and when requests are coming then they are evenly distributed across the ports to make sure that we split up the load and the beauty of kubernetes in this case that is simple to manage the instances of the ports in this case i was doing manual scaling but kubernetes allows to do auto scaling when when you run in production you can define rules that would allow to auto scale the number of posts based on the traffic so if you would during the day you get more load then the kubernetes infrastructure could auto scale to multiple ports at night it could scale down and so on this is especially important when you when you if you run your solution on cloud because you don't want to use many port instances all the time because you need to pay for that so you could scale down to reduce the cost of the of the runtime and now if i would go back to this lock window i would say that okay i don't need two instances i just want just one i scaled it down if i go now to infrastructure i would see that the instance that was created just few moments ago it's terminating and it will be automatically removed by kubernetes engine okay so this is a runtime and my point was to keep it simple not too over complicated because i believe it's important to show like the main points and if you would find it useful in your for your daily work then it would have it would help to create a better idea in your mind how it works and how it could be usable for your own use cases and maybe now we have some moments left i could uh go and um uh gives uh give brief overview in terms of code how you would use this solution on your own so as i said that in the main readme file you got all the instructions how you could start it with docker compose with kubernetes and you need to if you run local kubernetes you need to install a load balancer by yourself you go through that by that link and it's engines documentation and they describe multiple use cases if you run docker desktop then this is one way to install load balancer if you run other other tools then there is another way and so on if you deploy it on cloud then load balancer usually provision automatically you don't need to install it separately then we then all the components are described and yeah there are references to libraries on the library skip library on python index and npm okay and if you look now into the api which is the entry point uh which entry point it handles um rest calls from the outside and it's the place where initially request is handled and all the other job is is executed from here so we have you will find jam files which would describe infrastructure for kubernetes like how the ports are created and so on then there is a docker file which describes how to create the image and the same the same structure is repeated for all the components like for api for workflow for sample services so the same structure is everywhere and then we got endpoint this is the main uh main python script which is automatically invoked when container starting up then end point is initialized out of the box then because this container implements fast api endpoints so we're using fast api we define the url which is accessible through the host from the outside and then we create multiple groups here so the first api group defined defines uh boston housing related model api and the second group is mobile net related and i created this example on purpose so you would you would see how you would create different groupings and if your api is large it's convenient because then api is not listed just in a single single list but in separate groups and it's easier to navigate this is this is how it looks like this one is default group just the basic call to see if api is running then this another group for boston housing then this another group for mobilenet okay and then we go to api and this if you go to do this each api group is implemented by different script and this is where you define with annotations you define fast api endpoints like this one this is the method handles task training this is this one gets task status and this one executes prediction and it calls all the methods are calling helper libraries from um helper api calls from the skipper library and as i mentioned before this allows to minimize the dependency on rabbit mq and if you decide in future to switch to like kafka or other options then it's kind of easy okay and let's see so we have multiple routers for different api groups then this [Music] worker is defined for salary to implement asynchronous request processing and then if you would go to that's about api if very briefly and if you go to services let's go to serving service the serving service gets requests when you want to execute a prediction based on on model which was trained right so this main entry point which is executed also when container starts and it defines event receiver over here and event receiver is also is coming from the skipper library and event receiver main job is to register this container with rabbitmq and over here we point to the serving service this means then that when event will be received this event will be handled by serving service class service service class is implemented in the same container under app folder serving service and this is a definition for serving service class and we got methods call method which is the standard method which handles [Music] the call automatically when this container gets the message from the rabbit and queue and over here we are parsing the payload that is transferred through rabbitmq message then we open up load the model executes predict functionality and send back the result so this specific service it handles event and it doesn't produce an event and this uh another example of uh in in training service we receive event and we also produce event we produce event because we want to talk to the data service where data will be prepared for us and returned back and ready to for model training so there is the same main enter script which defines receiver then it points to the training service class which also defined uh under this app folder this is the training service class implementation this uh the same style is used over here is this a call method which is invoked when message is received and then when message is received we do different steps and one of them is to prepare data set and over here we call we we use event producer and we call we send the message to the rabbitmq and we say that we want to queue for data container and in this case rabbitmq will receive the message from training container distributed to the data container data will be prepared there and returned back in this response and in in this case this is not a synchronous call this is a synchronous call because we need to wait until response will will be produced we don't want to move next we need to wait and when this request is being executed rabbit mq allows us to do a weight basically and execute it in synchronous way yeah so this is quick summary and yeah so we are using it uh internally uh for ourselves and there are other i know i was in in touch with some people just from online they are also using it to run their own infrastructure so in terms of summary during this hour what i was trying to show and explain this simple ml ops pipeline which you could use to run your own ml services but it's not uh only ml specific you could run any workload this the container the trans the service could run any logic uh it just in terms of keeper we were trying to build a few more examples to show how to run with machine learning use cases as well it's completely configurable customizable everything is open it's easy to change up to your requirements it runs on premise or on cloud and it runs either with docker and kubernetes so thanks and yeah maybe some questions if if there will be any questions now i would be happy to answer okay so thanks