SBTB 2023: RJ Nowling, Streaming Feature Extraction for a ML Pipeline using Memphis.dev and Bytewax.
um so I'm a professor of computer science at a poly Technic so we're institution with about 3,000 students and we offer various sorts of engineering Majors um I used to be in ml engineer in industry and so this talk kind of comes out of out of that um as we know machine learning is used for all sorts of real-time prediction right so we have apis where requests get sent and and we want to make predictions um you know today I had when I Ed my my app for Uber I had it telling me okay this is how long we estimate until we connect you with the driver this is how long the distance is to where you're going and how long it'll take given current traffic conditions you know things like that so we are incorporating machine learning into everything and all sorts of end points but there's a small challenge um when we get a request to an endpoint like a rest request like this it's going to contain a record of data right so in this case we want to predict the home price the sale price but machine learning is about approximating a mathematical function so this function takes in a vector of numbers but that's not what our records look like right in fact we want to convert our record into this Vector of number so this is feature extraction now there's all sorts of aspects of this some of it is simple as one hot en coding categorical variables right so that we can easily draw planes right to separate out different different values uh calculated features such as you know body mass index is calculated from your weight and your height for example that allow us to get more predictive power than the raw input data uh we might cluster data if we had geographical data we might cluster it right um dates and times so machine learning models are generally speaking pretty bad at inferring periodicity so we need to extract the sort of different periods of the day so that the model can learn okay in the mornings my traffic level is going to be higher than in the midnight you know at midnight uh similarly when dealing with periodic signals um if you just sort of represent them as a single value there's the sharp break when they loop back around that machine learning models can't handle so you may want to project it into a two-dimensional space uh if you have a Bunch like complex records so say a bunch of clickstream data for users right you looked at what products they looked at or you have medical records do you have an arbitrary number of Records you want to collapse those down into a finite number of features so a fixed number of features actually and so you may compute various statistics on either the signal or you may uh do various sort of summary statistics things like that to uh you those as features um another example here is we have multiple uh algorithm models used for classifying email as spam or not so we have logistic regression random forest multi-layer perceptrons and naive bays and we see that using not just the body of the email but the domain of the sender and the user agent information uh increases the accuracy right so there's a lot of cases where we want to bring in complimentary data we don't just want to take the record we're given we want to pull other data sets in that allow us to uh make more accurate predictions one example of useful data the American Community survey right so if you happen to have ZIP codes in your data set then you can grab the median household income the population size age ranges household sizes poverty levels the fraction of people owner rent their home educational attainment things like that that can be incredibly Rich data right that can help uh improve the the accuracy of your models you may be working with a problem that changes with the weather right so there's a data set out there the bike demand data set from the uh capital d uh Bike Share program and if you look at their data you see oh look when the weather's bad there's far fewer people renting bikes right to commute to work so you might want to incorporate the most recent you know current weather data um there's startups out there and companies working on Precision agriculture right where they're basically trying to monitor the the growth of plants and help predict uh the when you should sort of put on fertilizer and water and things like that to try to maximize your yield and so they might incorporate all sorts of uh geospace information and satellite images and things like that in addition to uh basically taking all of that and and bringing in all that data and transforming it all into these numerical vectors we also have to handle missing or outlier data right so on the left I have a distribution of square footages of homes and on uh with a bunch of them having a value of zero so whoever was entering those records uh didn't know the square footage so they just put in zero so we need to replace that with a more reasonable value so that's imputation so this is part of the data cleaning process and it turns out if we don't do a good job of data cleaning it can be very costly I mean this is an extreme example but does anybody remember when the Mars climate Orbiter crashed it didn't land right this is because one of the contractors was using uh imperial units and NASA was using scientific units and their software just modeled numbers not the units right so they ended up having using different units the software was incompatible uh with the hardware and it crashed because they couldn't control the entry trajectory correctly so once we've done all that we have this nice numerical Vector right um and we can apply our model now the the question is do we uh do all this ahead of time right so do we do offline feature calculations there's advantages and disadvantages to this right so if I'm uh doing offline feature calculations I probably have some sort of batch job that runs periodically to recompute all the features uh for my records or objects that are coming in and I have some sort of key value store I'm using to retrieve those values at prediction time right now this works if you know all of your objects ahead of time right so if it's say a database of real estate and you know every sort of every house that's on the market that's fine but what if you want users to be able to input things that you've never seen before right they just give you an address for a property that you maybe have never seen well that doesn't work so well if you're relying on all the feature calculations being done in the background and just doing re lookup in real time um it's also expensive right these pipelines take a long time to run and you may be Computing uh features for objects that you very rarely see at prediction time so it's not really worth doing all that computational work and with the cost of cloud right um the other aspect of this is is sort of delayed response to changing environments okay so what happens when you know if you think of AD targeting for example I happen to be a fan of math books and so what happens when I go and visit uh a Publisher's website very shortly after that I see ads for books from that company so they're responding very quickly to my browsing habits in near real time and so if you if depending on where you sort of how you design your architecture if you're putting a lot of the feature engineering or feature extraction in sort of a backend offline system it may take longer to have those that updated data available at prediction time so what I teach my students is to implement uh their services using a pipeline like this right so we have a restful service or you know something similar and we get a request and the first thing we're going to do is we're going to pull in enrichment data so this is that you know American Community survey data right if we have a zip code or ways of identifying things that are located nearby geospatially using latitude and longitude so we have all this enrichment data that we pull in and we augment the record with then we clean the data to remove outliers uh or handle invalid value vales and then we go through the feature extraction stage to produce a vector of numbers at that point we can pass into the model and get a prediction that we can pass back to the user right so this is basically the architecture of what an online prediction pipeline looks like okay so how am I going to go about implementing this on the one hand I definitely going to need some key value storage to pull the data from I'm going to need some pipelines to keep those enrichment stores up to date so ideally I the data I'm storing in the enrichment store are things like information coded by ZIP code that are not unique to a given object so it's a much smaller set of data than if I had everything already joined with all of my objects uh then you know we need to calculate we either Implement all of this in a single restful service or maybe we have a we we're not doing it synchronously like that we have an asynchronous format so we can use a message broker right so we can run separate services and pass them through a message broker and we can use uh streaming pipelines for this right so uh Memphis who's sponsoring our talk today is developing a event streaming and processing platform so the idea is that it's has durable q's and Pub sub systems right like a traditional message broker but it's durable like kofka or Pulsar uh the main difference is that it's designed to be low overhead very Cloud friendly and designed for ease of operation and use so it's supposed to be um instead of just it does technically what the other products do but it's designed to be much easier to manage and operate and use on the other hand we also have bite wax which is is a streaming pipeline framework right so think of it as an an alternative to Patchy spark the main difference is that for me when I'm teaching my class ml production systems it takes me about six lectures to get folks to a point where they can be comfortable using spark takes me one lecture with bite wax right so they're much more much easier to use um it's all in Python they can just do a pip install whatever right so I use Memphis to hold my records waiting to be processed and then I use bite wax to process Implement a stage of the pipeline process some data and then move it into the next queue B wax has what you expect sort of standard stateless Transformations like a filter or a map but it also has a really rich library of stateful Transformations so one example is a stateful map let's say that you have uh clickstream data so you have a you know every time the user views a product you get a message for that and maybe you're calculating features off of that right you want to look at how many products are they looking at maybe that's an indication of they're searching for something but not finding what they want or how many times do they look at each product maybe if they keep going back every few days and looking at a product like a laptop that's an indication of intent buy it right and so we can use these stateful transformations to basically keep track of the state for each object and as we receive new messages update that state so we don't have to recompute it from scratch every time and then when we do we can emit those updated States as messages and maybe store them in say our enrichment store or something so they can be pulled later so there's this the more you can move to the online prediction pipeline is fantastic but you may not be able to move everything there and so having the ability to do incremental computation is a huge benefit for these architectures because it avoids having to implement batch pipelines that reprocess all the data every time okay now when you combine Memphis with bite wax you get this uh very cool opportunity to uh see into your system so Memphis has this overview where it tracks each application that's connected to one of its what it's calls a station and so you get a view of your entire system with these purple boxes being stations the Yellow Boxes being producers and the green boxes being consumers so it allows you to actually see overall holistically your system how many different pipelines are you running and what are their states I don't know about you but when I was working in Industry I think one of the hardest problems was that all the Frameworks I used were designed to basically um sort of treat each pipeline as a separate entity separate Deployable entity and it you was easy to get detail get lost when you had 20 30 different pipelines running in production right that's a lot to keep track of and manage so having a platform that provides the ability to actually see your system at a glance and see all the components in their interconnections is a huge step up right so um yeah I'm GNA pause here for a second are there any questions before I continue how can can sorry can you give him the how does bite wax handle faults with a stateful map recovering from a failure on a stateful map okay so uh bite wax actually each worker will write incrementally write out uh data to a SQL light file on the local node and then the current practice is to back up those sqlite files periodically S3 so I agree if you're storing if you have State running in the pipeline right you want to be able to recover so you can control um how often that that writing of State occurs and if you basically if you develop your own operators um you basically provide functions that enable you to initialize the state and to update the state and so it stores that so that that you can use it can recover the state in the future how many of you have used online feature extraction or like have you how many of you by show of hands have worked on say production ml systems some sort of endpoint that's using a model okay did you use um online feature calculations or did you mostly do offline online what were some of the challenges that you fac B in doing that yeah so what I'm really excited about these Technologies and and where this kind of was a big sort of mind-blowing aspect for me is when I worked when I was working in Industry we were using batch pipelines for this because we had hundreds of millions of different objects that we were doing prediction on and actually most of our team's time was spent on sort of of dealing with uh failures in that batch process it was very complicated very heavy difficult to recover from and so being able to do more of this online really makes it much easier to handle those faults um and and because it's done sort of later in time and you only need to do the updates for things that come in and related to that is like one of the patterns I found really useful was using the cues in Memphis as a sort of way to handle incremental computation right so I could actually write out state to a Quee and then pull that state later update it with new information and write it back so I could use Memphis or similarly kofar or Pulsar as a database rather than just a temporary place to put messages so that another process or another service could read them um I found that to be a useful approach for kind of handling the state outside of the computational framework right from a storage perspective which also has its ups and downs with respect to recoverability and lookup time so all right any questions all right well thank you folks I appreciate your time um hope you learn something and have a good day