Building Real-Time ML Pipelines with Feature Stores and MLOps Frameworks

[00:00:01][Music] hello so uh i hope you hear me well i'm i'm iron i'm a co-founder and cto for a company called iguazio which is uh developing solutions for accelerating machine learning in i2 production what we're going to speak today about is how to build a real-time machine learning pipelines in a simpler way than you would usually do and we have a packed agenda for those uh 45 minutes i believe uh we'll start with what are the real challenges in productizing ai and machine learning uh we'll talk about uh mlabs and feature stores are they the solution what what are we talking about um we'll introduce an open source project that we're uh we've developed called ml run that has those features built in i will go into a real-time churn prediction example to understand how do you model a problem how do you break it into uh building blocks and then sort of how you develop it uh and we'll see a different demo of real-time fraud detection or using again machine learning models and and real-time streaming data uh in order to detect a fraud and do some action based on that um so with that let's uh first understand what are the real challenges uh when we want to move machine learning to production and the key problem is that the research or the development environment is very different than the operational pipelines that we run in production so you can see here the left side and the right side where on the left side this is typically how a machine learning workflow would look like you know you take some data it may be like csv files or some extractions from database or historical data you run some analysis you know using jupyter building aggregations and looking at the data running some training in small scale and then doing manual evaluation and so on this is how you're building the model as you're going to production the world looks a bit different and first you need to bring the data and the data doesn't come from static resources like a csv file it may actually come from operational databases and you need the etl it may come from streaming kafka and so on it may arrive from web apis or any other apis to external services once you bring all the data you need to prepare that data at scale using different technologies batch processing same processing and so on and as you do that now you want to run a training on that data set and even that training you want to run it in scale across multiple algorithms multiple parameter sets you know with hyper parameters and you may need a cluster that you know can run and and work on all that large data as well as all the different permutations finally you create a model and that model needs to be running behind some real-time endpoints or even a stream and access the real-time features and also that model need to be modded monitored so when we see that there is a drift or an accuracy accuracy problem we have to uh retrain and build a new model and serve it and so on and we also need to integrate our model with different apis if we're building a web application or some event-driven application it needs to integrate and interact with the model in order to bring real business value and that flow is very different the typical flow of taking your jupiter notebook and building some model and then creating some report usually in many organizations this is done by separate teams you know the data science teams may build a model in their notebooks and then different teams of data engineers devops ml engineers and so on will bring the right side and this essentially brings a lot of complexity and slows down projects significantly sorry so the key challenge that we have in operational machine learning are those that you see here you know the first one we discussed is essentially working in silos different teams work with different tools some work with sparks some jupiter some kubernetes you know deep learning and tesla flow everyone has his own framework and he know his own ways to track experiment to do ci to do development and so on so we need to find a way to collaborate across it's a very long process as we've seen and it you know going to production is not just building the things that we mentioned is also for every component you have to think about packaging and testing and scaling and security and monitoring logging and you know ci and continuous deployment and so on and that's a long process with a lot of development a lot of testing um another critical thing that we're going to touch upon here is how to access features you know if you create features in the offline by just having a csv and now you need in production to create the features from a kafka stream or from a transactional database and so on that requires a lot of logic there's also political issues around the security issues who has access to the database you don't want to interfere with the database operational databases as you do machine learning so there are a lot of challenges that because of that you'll see that customer the companies don't really have uh all the right features that they need in the production pipelines because of all those reasons the complexity and others and finally we have to think about the model accuracies because you can train a model on some test data but as you go into production and the data is different the behaviors are different your model is not going to produce the same results we need to know how to monitor that activity as well as remediate like do retraining build ensembles and so on um so the typical flow today in most organization is what you see here there is the research uh pipeline where you know someone throws the data into a data lake uh then you do all sorts of batch transformation whether it's using sql language or python and pandas you run some uh you store the results in insert a data set csv a parquet file for training and then you run the training as you grow and you need a stronger training more scalable more automated then comes the machine learning engineers or the amalops engineers they take the code from data scientists from the jupiter refactor it to run on a cluster typically a kubernetes cluster or on some cloud machine learning service so that's the the research pipeline now as you finish with that pipeline you go to a second pipeline where you you build a more of an operational pipeline that taps into the operational databases uh run online transformation it's very different than batch transformations so it's sometimes it's stream processing it needs to work at scale you need to be partitioned and so on and and then you store the results in some database for the scoring and you run the serving and finally we have finally we also have the governance pipeline which is where we track every activity in the model the inputs the outputs the feature vectors the latency the performance and and we take that data and this is how we analyze drift accuracy problems we may want to explain the results later on so we may also do analysis on the data that was gathered from production in most organization those are separate implementations done by separate teams and brings a lot of complexity and slows down the process also again a lot of delays because you're moving parts between one guy and another you have accuracy gaps because the pipelines you're building for development for building the model and for using the models are different pipelines and this really calls for a different architecture and we're going to dance with how it works but essentially the general idea is to think more in a form of automation with you know everything needs to be automated into a production pipeline so the first steps of development need to also be instrumented in a way that you can reuse that and not refactor and not redo your your entire code but also you need to think about the data as the center piece in that workflow so this is really where you have the concept of a feature story that is a concept that was originally coined by uh you know the big web web providers you know netflix uber etc they all have their feature stores and um and now there is more open source and commercial solutions which are feature stores what is the future store feature store is essentially a way to build and catalog features uh but also not just the build and catalog features uh also a way to describe the transformation describe a recipe for building those features that will run automatically what's also unique about feature stores is that they know how to produce features for real time and for batch so for training use cases for serving use cases and for monitoring and analysis use cases all you're doing it once and then it's used across every um every every different application so that's the general idea about that and we're going to see how it's uh how it's designed uh what i'm going to to show you is using doing all of that with the framework and open source framework uh called ml run and so you can just go to amaron.org and see the architecture is a lot of documentation the way that it works in order to automate all those pieces that we're presync here is uh it has two underlying layers the first underlying layer is the way to manage catalog and build features this is through the feature store and connectivity to various data sources the second thing that you have is a way to take a simple piece of code and run it as a fully managed service on kubernetes with security telemetry auto scaling logging and so on on top of those two baseline layers you have different components that allow you to build you know the data ingestion pipelines the training pipelines the real-time production pipelines as well as all the monitoring and it's fully managed with the central metadata and orchestration so you can go into the link and and see the core com the core technology behind it is essentially serverless computing what does it mean serverless computing one of the big challenges we saw in moving to production is taking a piece of code building containers around it scaling it out addressing security parallelism and so on so this is done automatically in amazon so the idea is that you can take a jupyter notebook or some piece of code with a single command line you or a single python command you essentially make it a fully managed microservice that runs and scales automatically it also extremely high performance because it has a lot of parallelism built into it there are some benchmarks that were published in towards data science and other places showing how you can actually run your code faster and also as i mentioned it's auto scaling so one of the problems about data analytics deep learning machine learning you want to run across multiple containers or multiple virtual machines so those serverless functions are horizontally scalable scaling so they can run this against a much larger set of of data and computational resources by distributing the workload in an efficient way and and you can also create those functions and store them in a marketplace your own marketplace or a public marketplace and reuse them in the pipelines the other core concept we mentioned is a feature store is essentially a way to to build those ingestion pipelines that will produce the features and persist the features in multiple mechanisms for example offline parquet files used for training or real-time nosql databases used for serving but the pipeline is designed and defined only once you don't need to do it multiple times and we'll see a couple of examples of how you build those and run those pipelines so once you build those pipelines the features are generated along with all their metadata with all their statistical information and you can just go and choose those features from a catalog and then plug them into your training or serving workload with two lines of python code so again very easy way to build features and use features and this is already ready for production without any extra devops i.t engineering work so let's take an example that let's assume we want to build a real-time pipeline and i i take an example real-time turn prediction so think let's assume you have a mobile app or a game or any anything like that that you want to create a pipeline that will predict the churn of a customer obviously you can do that in batch you know in many cases churn problems uh are done on a monthly basis you want to predict if someone is going to keep on paying with this example we also had a webinar on then you can see that we you wanted to address the turn problem in a in a real time fashion because most gamers and most application users will churn within the first 24 hours so we wanted to address that so what what is really uh you know the data that we can gather for such a mobile app we have some static data about the users when they register like the username the birthday uh geolocation or zip code which can tell us some information about the social economical situation the zip code and so on uh affiliation memberships and so on but there's also a lot of real-time data that comes in like user activities a user goes and plays a game goes and clicks a button and so on uh their transactions like user goes and and buys credits or tokens or you know pays a monthly bill and so on and then maybe other types of events like social interactions a user does a like user shares something social interaction can tell us something about churn of a customer so let's assume we have all of that data and we need to build feature this is an example of the features that we want to build in order to train our model and those features will be things that for example uh aggregate usually when you have uh real-time data you want to build aggregations like how many times you clicked in the last hour uh how much activity how much money you spent within a time window and so on so you can see that you can take those sources that i mentioned and build features now you need to join those features together into a feature vector because uh each one comes from a different source and they also need to be joined based on some timing information like i want to if i'm looking at the transaction today i want to also look at the activity of the user in the last hour relative to this time point in time of today so we need to do something called time traveling is essentially join the different features based on on timing information so let's assume i want to build one of those lines just one or you know or two couple of those lines and the only tool i have at my disposable at my disposal is sql so this is what it's going to take me to build a single feature this is a real example of writing a select within a select you know within a select to do time aggregation and things like that it takes obviously a lot of time to run a single that for a single feature uh and and that's very complicated how many of you want to go and do all that coding uh it's very slow very resource intensive because it's scanning the database multiple times and so on and you it won't work in serving because in serving when you're going to get the transaction from the wire you want to immediately go and react you may get thousands of transactions per second you can't run thousands of those sql queries per second in a decent latency so we need a different way and and there are different ways there is something called the kappa architecture which is essentially processing the data as a continuous flow or as a stream so typical tools in that will be like spark streaming or apache beam or flink or used to be something called storm no not used too frequently these days so you have all the information that you have that you want to build you know all the transaction log as an example we want to take uh that transaction log and essentially process it in two different ways one is more as a cumulative processing like a stream processing um and the other one is more sort of a storytelling in a csv or rk or batch files it would be used for for some batch processing so that's the general idea of how how you do real time how you do processing on real time in real time data and this is also a slide from apache beam let's assume i have the the customer data customer transaction data and the user the transaction user is the the main key that we want to uh use as the for the feature vector so what we typically do we throw all those events if we want to scale we do something called sharding or partitioning essentially divide the entire flow of data into uh partitions by by users and then we have some stateful processing of that what the why do we need stable processing because in many cases we accumulate stuff if we want to do uh you know average or some of the transactions in the last hour that means that we need to accumulate the the transactions per user in the last hour or the last week and so on so this is in general how you build stream processing but this is very tedious a lot of coding by the way usually in java most of those frameworks are java frameworks and if you're a data scientist it's going to be very hard to build something like that so so part of amaron we built this is by the way called an actor model this way of processing we we built a mechanism to run those pipelines in pure python very simple abstraction and you can run exactly the same pipeline in batch or real time so you implement it only once and you have the same implementation for the training or the survey so let's start with a simple example let's assume i have some csv or parkl file or a database that has some transactions i want to fit them into this transformation pipeline and write the result into a csv or per k file that will be used for training so again we can take and build a batch job that will take the source data we'll run the transformations and store it as we mentioned we do need state to accumulate stuff and we need to persist the state in many cases if we want to distribute the workload and handle failures and you know and so on so this is the baseline architecture but let's assume we also want to do the same from a kafka stream or an http then it could just work we just need to change the way we we feed the data and run the same pipeline but in most cases we want to have the data ready for training but also for serving or maybe even to build dashboards out of it et cetera so the only thing we need to add is essentially syncs you know targets that will write the data in multiple formats that will be suitable for the different use cases so that's the the general idea of how to build a solution that takes that data and we can fit any of those tables that i mentioned before so we can take that data feed it through pipeline and uh generate the results and then use the results in the end so and this is much simpler much faster and much more scalable than the mechanism that you're using today now the way that we want to also do it is in a way that is ready for production and as you start developing with your native tools whether it's a jupiter notebook or pycharm you can go and test it in a small scale and then test it in a larger scale and then go in one click deploy to production so this is the overall theme of what we're trying to do with with the mri essentially you can develop in your native ide get some something working test it on a bigger scale on a cluster on usually on as a kubernetes job or service and then as you have that single single thing working you may want to create workflows which comprise of multiple steps uh for example in the bad side you may want to grab some data run training run evaluation do test deploy on the model see it actually works or in production you may want to have a pipeline like the one i mentioned before of gathering streaming data aggregating it doing some model inferencing and so on now everything that you're doing you need telemetry if it's the badge or the training side you need experiment tracking and all of that if it's a production side you need to see things like cpu utilization errors failures drift analysis accuracy problems and so on so one of the nice things about this framework is that telemetry is built in you don't need to do anything to make to see it all works and and also everything is reusable so as you build features or functions or models you can just store them and then use them in a different pipeline that's a general idea and you can again go to amazon.org there's a lot of documentation and examples and notebooks that will show you how to do those things one of the key features in ml run is the ability to run real-time graphs there are dags or graphs certain pipelines focus on batch there are many of those like kubeflow or you know jenkins or ci tools and so on but it's very hard to find a simple tool to build uh sort of real-time graphs so enamel ranger is a sort of something called a graph server that is used in different things it's used for serving it's used for ingestion and for for different tasks and we can build real-time complex real-time pipelines that span the data engineering part along with the machine learning part so we can build a single pipeline as you can see here that starts from an event event could be a kafka stream an http endpoint a batch job and so on run some enrichment some pre-processing some exception handling go into a model ensemble that has multiple models that will be used for first for creating the result and then some post-processing uh so you know most serving engines the only thing they allow you to do is create a model serving end point but in in real application you need something more like what you see here in this diagram and in order to describe this thing it's just you know going and writing this piece of python code uh simulating it with you know a couple of lines of code and when it's ready just saying deploy and that will build the entire real-time pipeline for you in a fully managed way on top of a kubernetes cluster this example by the way is also documented if you go into demo run documentation and you go into the serving and data pipelines you'll see several of those this is this example which i mentioned and there's also another cool example of the nlp pipeline distributed nlp pipeline uh that you can see you know paragraph you know document processing entity extractions etc so it's not really limited to machine learning cases it also addresses deep learning and nlp pretty well now once we build those pipelines the next thing that we must do is is also monitor the activity of the of our models and our pipelines so the way that it works is you know we have that feature store that stores all the features the features also have a lot of metadata all the statistical analysis of each individual feature and we're going to see in a minute how we can actually use that statistical analysis information in our serving engine which is very powerful for imputing for value computing so we want to feed our model serving pipelines with those features that are coming from the feature store now the model serving pipeline will write everything that happens you know the incoming features the outgoing features you know sort of the predictions the latencies the performance data into an output output stream and on that output stream we have another serverless function which is sort of built-in function and it uses the same concept of stream processing to essentially produce the real-time analysis of the production data and we'll see in a minute how it looks like but essentially it analyzes drift and accuracy problems and and custom matrix and and so on and writes back the production data into the feature store so later on when we want to do retrain we can just trigger a retraining event and it will upload and upgrade all of our models so it will grab this it grabs the statistical information from the feature store to understand the expected behavior of features and it writes back all the the feature real-time feature statistics and the feature logs so we can detect drift and also do retraining and things like that this is by the way a snapshot of you know this is more like the built-in ui and there's also a grafana plug-in you can actually go buy model and see the real-time production data so again very simple and intuitive so with that let's now go into a demo and i'll show i'll share some some notebooks and i'll taking an example of predicting financial fraud okay so what we we have in this example is three different data sets you know so we may have user records uh we have the account activities you know login logout and so on we have real-time transactions you know someone paid for something and so on we also have the label data but what you'll see that we're producing few different features that from that the transaction data is actually translated to three different data sets one data set is the road transaction data the other one is transaction data by category and the third is the labeled data which could be essentially when you're doing the training you can figure out the the label data like when there was a fraud from the transaction and the transaction so as we go and ingest all of those different real-time streams uh into the system it's going to be stored in the feature store then we can just go into some interactive exploration on those uh feature sets and feature vectors later on we may want to build a complete training pipeline that says she create takes a feature vector creates a training set run automl and deploy and test the model um and then deploy the production model and deploy the model into production once you have the model in production it works against the feature store to get real-time data as needed and also all the data is being monitored by the model monitoring piece and written back into the feature store so we can do post-production analysis for like explainability and so on so this is the entire pipeline i'm going to to demonstrate and again a lot of those things are you can look into the different features for example there's like a stocks analysis demo this is also like a real time healthcare data analysis for like patient deterioration analysis all based on real real examples real customer deployments so but let's go into the the fraud example which is slightly simpler let me zoom out a little comment okay so um i want to build those pipelines that again usually i need data engineers and data scientists and devops and you know a lot of different people to build those complex features that i mentioned so i'll try and do it in a single notebook to build all of those different features i just need to like import and do some setup [Music] for sql demo now i have some data this is sort of we've taken some data from kegel and some other data that we have that shows has like um monetary transactions okay so and it's stored in our public data repo so this is the example of this transaction transactional data [Music] now we want to build a pipeline in this pipeline we also demonstrate how we build custom classes for the pipeline but this is actually a library class which does date extraction we want to take the timestamp and extract like time of day hour of week etc and so we can even implement like custom python classes will be plugged into the pipeline and here we're essentially defining our real-time pipeline this is the entire definition there's no no other code for it so we create a feature set which is essentially defining sort of a pipeline of creating multiple features we define the the entity know the source of the transaction it also uses a column for timestamp give it some description and then we build a graph i graphically serve that and the dag you see this is the that of computation the first thing we need to do is extract dates uh hour and day of week and then we may want to do some aggregations on two hour 12 hour 24 hour windows across the amount field okay so this is the only thing we need to do in order to build that feature set and we can ingest the data we can adjust it from the notebook for sake of development or we can ingest it as a real-time job or as a batch job with the change of a single parameter now we can also view visualize our pipeline and visualize the data that is produced by our pipeline in the notebook so we don't need to go and run some some spark and some other things and data engineers and they'll send us the csv with the results and then we want to iterate here we can iterate very quickly we go change the code change the pipeline rerun we see the result of our features now let's take another set of features which is the activities there you see like user login logout timestamp etc and we build another pipeline here we want to do uh one of the encoding of the different activities like login logout and so on so we just create a feature set we define a pipeline that does some mappings and one of them coding on those mappings you can see that this is creating the onenote encoding with the result again a few lines of python code um okay and i'm it's already ingested when i say it's already ingested then it's uh actually if you go to the amazon ui you'll you'll go into the for example transactions you can see the different uh transaction how to use this uh you can see that the different features uh the transformation pipeline that we created a preview of your data all the statistical analysis of all the features and so on and this is done automatically the minute you've done this ingestion or preview and now we can also build another type of data set which uh groups the the transactions by different activities like someone buying something in the supermarket or buying some shoes and so on and we need to group the transaction because there may be an anomaly based on a category if someone buying uh shoes uh five times a day it's probably an anomaly if it's going to the supermarket every day it's probably not an anomaly so we also want to group the results of based on some category and create a pipeline that will produce a feature set which is grouped by categories and also the last thing that we may want to do is is actually ingest the label data and the label data as i mentioned before is derived from the original data because there is a flag in the original data indicating a fraud and we need to do things like potentially time shifting and other things on that data so the last feature set that we created is the label a feature set that stores like true or false if it was a fraud in the relevant time step okay so this way we just created all those different feature sets the next thing that we want to do is just uh create a feature vector so we um we go and again build a feature vector essentially saying you know i we need all those features from those feature sets bill give me a vector this is the labels this is for training we can go and explore you see all those fields are joined automatically and we can explore how our feature vector looks like and again iterate quickly on the notebook then we want to run training so we use emerald's ability to run automated jobs and schedule jobs so just giving it the the feature vector as an input telling it go training on multiple algorithms because we want to do serve automl it will run it will choose the best algorithm provide us all the different analysis of the features of the algorithm rc curves etc by the way again if you want to see that in the i don't run ui you just go and you can see the training jobs you know what was the source of the training job the future vector the artifact that were used by the were produced by that training job the results for every epoch or every algorithm depending on deep learning or machine learning and so on so all that again was done automatically for you you just said here's the data go run some uh library function that does and training and the last part is now going to production this is usually a part where it takes at least six months to get everyone together and build a real-time pipeline with real-time data engineering and so on so here i'm going to do it in a couple of minutes we have a model serving class which is essentially loading the model we'll see that we produced a minute ago it's doing uh pre-processing and what is uh done here in the pre-processing in may in some cases we get null values now values because the aggregates didn't aggregate anything there was no transaction or and we may want to substitute values from the from the real-time stream with some some imputing computed values so the feature vectors they also store statistical information about every feature in the in the future vectors so here essentially what this code is doing is replacing and null values with the mean values of that feature or with some other default values so again the complex stuff usually computing and so on you you can do it in few lines of code but in the next version all that is going to collapse into the feature feature get function so as i built it now i want to build a pipeline i don't want to just create a model endpoint i want to build an ensemble comprising of the three of the models that i generated rml generated multiple models and i want to serve an ensemble of three of those models so we can just go in and build build a pipeline add a list of models we essentially query the model database we get the different models we build a function with a for loop we create a function with an ensemble router and multiple models we create a simulator for our real-time pipeline and we can just go for example to see how it works with the real-time features i can just go in and load the feature vector and do a get essentially here i'm going to give you the idea of the user and you gave me give me all the different features in the vector you see with one command i can just create a feature vector uh don't need any data engineering uh practices don't need stream processing knowledge all that is happening for you and if i want to test the simulator the simulated pipeline i can just ingest a sample sample data and see the response the response is the the output is essentially zero meaning there was no fraud on that transaction if i want to deploy it into production the only thing i need to do is just say deploy and that brings containers and serverless functions on top of the kubernetes cluster and giving me an endpoint to the real-time pipeline i could use kafka or kinesis or http to access that that pipeline and we can also test against the real invoke means testing against the real endpoint and you can see it produces the same the same results so you see with three notebooks i build something it usually takes organization six months to build because of all this automation and all this uh serverless processing and this restore capabilities so so this is essentially what we've seen in the demo i didn't show all the like fully automated pipelines but this can also run as a ci pipeline uh using kuflow or other things like github actions or for gitlab and so on and just anyone that wants to there is now an envelope for a good hackathon that we're running together with microsoft and supported by odsc and other organizations uh it's about building automated sort of machine learning solutions for different uh good you know things were good like going to the link all sorts of protecting kids and protecting against uh fraud and other problems so people can join into teams and then uh make submissions you can see judges from all you know leading in the industry leaders in that space so everyone is welcome for that and also you can see a lot more content we're running a bi-weekly webinar called envelopes live and there's a lot of different sessions from different companies and different individuals leading individuals in the space so with that that's my session and let's open it up to two questions and see if you have any