scale.bythebay.io: Karl Skucha & Yan Yang, Real Time ML Pipelines in Multi-Tenant Environments
start today I hope this is necessary yet so my name is Carlos good habit and the director of engineering at Salesforce Einstein and I'll be presenting together with jian-yang was the lead engineer on this project and we'll be talking about multi-tenant machine learning pipelines this is something I would deal with a lot in Salesforce we have a very heavy amount of mocha tendency so I think there's four stalks by our team at this conference so I will be giving the first one so I'll give some overview of Salesforce what why we're doing machine learning at Salesforce how we're solving problems for customers and then I'll give a quick primer on machine learning and on real-time type lines maybe just forego stretch how many of you have worked on machine learning before great maybe left hand how many of you have worked on the real timely arrive lines before okay great so some for calibration how many of you know what Scala is awesome good idea so then so I'll go through it sort of quickly and then Yin will join if you'll talk about what we've built what we actually built and what are some considerations into our design process and then we'll talk about many things we have we have to do so so Salesforce is a enterprise SAS b2b company I think I got the order wrong all those thing bus things but basically like those are fancy terms for we store data for other companies many other companies from different different industries different sizes and we provide them with a service in sort of domains that we're good at and those domains are sales service marketing a few other things we call those things clouds and for example in our sales cloud makes sales people of our customers more efficient by providing tools for to to sell better and you know service cloud powers you know support centers so on so forth so one thing about our our customers as many of them are not tech companies many of them are small companies and many of them don't have the resources to hire data scientists and build data science pipelines especially for their sales and marketing efforts so this is so as you know like you know to to create a pipeline you have to hire hire data engineers to set up data warehouses to set up ETL systems you have to hire data scientists to you know build models of the mouse models and you have to hire a whole team of dead of Engineers to you know to set up the infrastructure and to maintain it so at Salesforce you know we like to say we're building the world's smartest urm our team is building the backend of much of this platform and really we're trying to give take the burden on and do the machine learning for our customers by providing them with out-of-the-box integrated particular applications some examples are which I'll talk about a little bit today is predictive lead scoring so you know basically this is taking a bunch of potential customers and predicting whether they will become real customers there's also case classification which is helping routes support tickets to the right person so that the end consumers of our customer will get a better service and finally there's a few others but one exciting one is last week at Dreamforce we announced Einstein's prediction builder which actually allows admins Salesforce admins to predict any any sort of field on any custom their model they have within Salesforce so so so far our team has focused on batteries cases of this platform many of our flows run once per hour but of course you know I keep getting you know harangue it can you go a little faster can you do 30 minutes for my customer and I say of course no I can't do it but of course I can and eventually they convinced me that this is really important and I tell them sure you can do 30 minutes and then they of course they come back and say walking the 15 minutes and then I get sweaty but but you know like like after much discussion we eventually do 15 minutes and then someone comes in and says well we have a whole different use case while the customers on the phone we need to predict something we need actually at well under a minute and then you know we realize that we have to do something else completely so before we talk about what we built I want to give sort of a quick introduction to machine learning and to how we think about going to real time so let's and they'll take you through a data sets and this is a DSS I will use so for generating a model let's take the leader objects again this is just a fancy name for a potential customer and we're trying to predict whether this potential customer becomes a real customer and the is converted field is basically an indication of that force means that they didn't become a customer so it means they did and blank means we don't know yet so in this data set we have you know only two interesting columns title and what elite came from the source of course the really the object a salesperson has you know many more fields but I won't bother with that so generally when we bought the model we take a bunch of historical data we run it through some elt process where where maybe we put it into a data warehouse or data like I will omit those details but but essentially what we what we end up doing is some sort of feature engineering steps this could be you know anything from normalizations on numerical features to 100 coding on categorical features or something that requires more domain knowledge but it's something that was predetermined by our data scientists and then we go through a model training step where we might train one model one or more models we can five different hybrid parameters and we will go through an evaluation where we try to figure out or try to determine whether the model that the best model of the model that we actually was the best is good enough to be deployed it's good enough that it will serve good predictions and once we're sure I can limit it for this model and we have our model so then for our simple pipeline to serve predictions let's say that our leads we get changes from a lead object in this case I think max who became a director and Yan is a new lead that we get and we take this data from the same feature engineering step and this time we use the model to perform a scoring step on the data and then we get some scores and these course are indicators of maybe probability of conversion or just normalized numbers to to help the salespeople use them in their business flow to maybe prioritize the more likely leads or whatever so so one thing I want to mention is this is a really simple pipeline it's not hard to make faster the thing that's really simple about it is that all that the features come from the payload everything that we need to predict the lead comes from the media objects and it's not really hard to make this process faster it's you know well not trivial it's also possible to convert all these components to streaming and then you have you know very low latency things get more interesting when you add other sources so for example our data scientists are clever and they say well actually we want some data from a different object we want a number of email replies as a feature and great you know we have this in the activity object we elt it into our fixture engineering step but we know as oh no we have to join our lead data with potentially a very very large activity object these fact tables tend to be very big and this is a very expensive process while I can give us more more more accurate scores it's it's not something that we can easily scale to faster speeds and of course just becomes compounding the worse because most most of you probably reply to email more often than you get remoted so so not only do we have to do this for every change in allele object but for every change in the activity object we have to join it to the entirely object because we remember we need our source to mean our the other features so so as you can see you know as we add more sources for every change in any source we have to join it to all the other sources so this becomes computationally difficult to basically to do and even scaling a process with multiple sources to an hourly basis as difficult so typically the way these problems are solved is you instead of determining the features when you when you need to use them you actually pre compute them whenever the data comes in and you catch them in what's typically called a feature of store this is usually a key value of store where for example if we store them in this in this way where the leader ID and the feature name are the key as new data comes in for simple features we just update them in the future store and for for aggregate features we increment the manager store and this way we can do our scoring decoupled at any point after because because data is guaranteed to be sort of up to date in the future story and as you can see this only scales linearly with with more sources we just have to do you know for n sources we have to the end times more computation so so then go into streaming you know architectural e this pipeline looks the same I'm not saying it's trivial to go swimming but in general like if you could convert your eye the savings and you convert your processes to streaming processes then you know this this architecture looks pretty similar so that was my quick overview of what this looks like in and you know in an ideal world of course it's not so simple in real life and even at Salesforce we have some more considerations and we won't talk about the usual details like data management or compliance or or just ETL being card we will talk a bit about how to reconcile batch batch data with real time data the data is fundamentally different batch data you know has has generally more more data in it in real time so how do you compute the features in the same way and also in the Salesforce has over a hundred thousand customers we want to build this for multiple sources and multiple entities and multiple applications how do we deal with the true multi-tenancy of this problem so so with that en will come over and he'll talk about what we've what we building so far all right thank you car so I will go into some of the architectural components in this system and as the time is limited I would mostly focus on the future star and some other features that actually supports supported so to be new is let's look at the high-level architecture of the Einstein platform we have a typical batch pipeline which ingest data in in a batch manner deposit is a files into a source some sort of data Lake and we have separate training and the scoring process to consume this the Attorney General output and the mirroring that process we will have a real time scoring process that is really for scoring purposes only it actually grabs the model file generated in the offline training process but it ingest data in a streaming fashion and output the result back to the end user as Carl has already indicated the feature star in the real time story is pretty critical because it allows you to cache intermediate results and you don't need to do expensive things again and again so this is an ongoing project so I would mostly go through some of the considerations and Cavett's we have encountered along the way and the some of things would like to do in the future so before we can delve into the future star there is essential pillar in this whole picture that is the common feature definition between the patch training badge scoring and the patch and the real-time scoring so this common feature definition is meant to abstract away the process of generator from raw data in a sense it's trying to address this question so how do we know features used in training and scoring are generated in exactly the same way so just bear with me to take a step back to go through some of the basic definitions so what is a feature this is a Wikipedia definition of feature as will he be there always is it's 100% accurate but utterly useless so a more practical or though nurse trainee also a little bit less rigorous definition with the be that a feature is more like a transform Colin data sets that you will plug there only into training Oskari there it's usually a very wide gap between your raw data and your feature set and you have to do think mounting things like the field extraction data cleansing and it's slightly smarter things like feature engineering and transformations and so in the context of Spock ml before I go into that let me just that so our platform uses a Optimus Prime library that we build in-house that is built on top of stock ml so all our machine needs are party to vampire spark ml in the contest by ml if we have a raw file containing a lot of Deeds for example the training process is basically feeding a data frame of the features into a so-called estimator and the end result of a training would be a transformer or more likely than not a sequence of transformers and the scoring process taking a similar data frame of objects in this case needs and apply this transformers one by one on them to generate a data frame of prediction results which in this case we call them scores so as you can see the input data is of the same type it's lead but they actually ingest very different type of data's the training process usually generates a huge chunk of historical data with a label on it so let's say six months of data or our transformer the scoring process actually takes in most most recent data like past one hour a past one day's data and again which you want to actually make predictions with so so the question is of course how do we ensure the fifth generation consistency how do we make sure that israeli same logic is performed on the raw data to generate the features we need to feed into the two different processors of course the idea is simple we can reuse the code and make sure that it is this this particular part is the same and we can separate separate it from the reader that actually reads in the data but remember that Einstein platforms meant to be a platform that on top of which users can write their own applications so we do not want to burden user with this explicit understanding of ensuring the distances between two steps that would mean that we will have to have a very simple and a composable API for users to define the feature generation processes so this is so in optimus prime we actually identify some key features in future generation first one is of extract function which basically takes care of all the dirty jobs and transforming that input data into some intermediate state but it includes things it's a simple map function that in usually encompasses actions like field extraction data cleansing etc and the middle of the featured definition is aggregator which aggregates this particular field of the input data along a particular time horizon optimus prime piggybacks on the twitter algae bird amyloid aggregator so it frames the aggregation in the form of a monoid so basically for those of you who are not familiar with Archie bird it basically contains methods which allow you to transform the input into intermediate state and reduce intermediate state and the present the intermediate state as a final outcome the third and last the defining feature is an aggregation window over which we actually perform the aggregation so once we have these three things pinned down we have a unique process of generating features from a particular type of input data so as an example here we actually generate a feature which is total number of activities in the past seven days as you can see that this is a perfectly composable api where we identify the input type output type the extract function aggregate function and the window over which we are actually aggregating of it and similarly if we want to predict particular another feature that is the number of conversions in the past one day we can perform exactly the same way and the api's look very similar to each other one critical thing is that at the end of the fifth definition we need to we need to explicit aspera phi that whether it's a predictor feature or response feature which is a label used in training process and because our common pitfall for beginners in machine learning is that you accidentally include the response feature in your in your predictors and you will gather like did that machine many models with very almost perfect performance and in real life is not like and it doesn't work in in real data so this will ensure that Optimus Prime internally will safeguard against disease and the responses feature and any derivative fish out of it will not be able to be used in any predators so this common feature definition allows us to extract the abstract away some of the common definition so this is what it defines a feature so this will allow us to put it feature values actually into the future store which car has already touch about a bit so for feature stores the main purpose of course is to cache the features I would like to touch based on two things first is how do we ensure future caching in a multi-tenant environment and two is how do we actually make this feature cache avoid the computation so multi-tenancy in a in a key value store environment is fairly straightforward you would say all we need to do is put an ID in the key and secure the API so that nothing and no query can can request data from different multiple tenants that's basically the gist of it and that's not the plan of it in in practice there are still some Cavett's are on top of that which basically we want to skill up and skill down what does that mean so in your life when you have a multi tenant application so your till your clients are usually not very uniform you will most typically have a few gigantic organizations with huge amount of data and a field a few big organizations and a long tail are very small organizations with only a handful a handful of data to handle these different types of load properly you would need to be careful about your key design and in making sure that say single challenge data is even issued across your entire cluster so that you don't overload a particular node in this case we put the technical type record ID as well as a unique feature ID identifies the is that feature into the key so that even even if you have big talent if you have a tenant with a very big object it's more or less guaranteed to evenly distribute across your cluster so that's what I mean by skill up on the other side since the isolation is at the software level for a very small tenant it may only amount to a few hundred rows in the same table so we don't need to provision special infrastructure for the small tennis for very small talents and that'sthat's what I mean by skill down so those are some of the important considerations in achieving multi-tenancy with multi talents if taken care of let's look at the actual usage of the future star that is to avoid recomputation car has already touch upon a part of it that is we key different features by the same key and I will just this line it will iterate this for example this leader feature is keyed by its own ID five well the activity feature is hidden not by its own ID but by the leader ID five so that feature story will start three features all of which kid by five and the downstream scoring application when they present the ID five they will get all three features together just as if they are coming from the same entity while in reality they come from two different places of course this is this looks like we are actually outsourcing the shuffling step into an external commando store and the important part of course is that we don't have to perform the first stage aggregation over and over again if this joining on is only performed once this is this does not actually give us any efficiency gain but the in in practice there are a lot of scenarios where you will have to perform the same joint again and again again and Fisher Kashyap will avoid expensive extraction computations at the first step so the second the second area where the features stall will save you computation time is easy aggregation we have already talked about aggregation achieved by a monoid aggregator so I will provide just a very simple and straightforward example let's see our feature star contains this data which is basically the user count of a particular the counter particular user at the count of user activity of a particular user and in while the it while the feature is defined very different needs internally feature star will keep track of some daily aggregations across the southern horizon so now when the client actually asks for a feature of a count of last seven days the feature star s DK would be able to smart to be start smart enough to utilize the free common feature definition we defined early to computer on the fly what is the actual aggregated menu and if a new data camp rose in the intermediate state will be consumed will be will be updated and when the client requests the same magic again the official store SDK would be smart enough to know that there is a sliding window so November nice they had now forced out of the window and we should update the cont so this allows us to continuously update the future store with in the in the context of real-time streaming where we only have the latest data and we can catch the intermediate data again like continuously so that every time we query for a particular aggregation we will the feature star is able to return return so of course this example seems a bit contrived in the sense that it is very special light for this particular access pattern there there is of course this question that what if we need other aggregation windows that is the problem that we are actively solving and we have a few ideas floating and one of them is to roll up data in fixed schedules basically we keep daily aggregates up to a certain degree and then we roll up them into monthly and then here did yearly aggregates this way we can support a different aggregate patterns access patterns like last seven days last month year to day our last 12 months of course it's not for free it means that you are actually making the granularity closer for older data so the query the access pattern for older data will become less and less flexible as a price to pay but most of the common queries you want to use are actually against more recent data a benefit of this approach is that you limit the size of the intermediate state you want to you you want to start your feature store for each feature a second idea is to perform exponential weighting which basically means that you weight more recent data more heavily and the benefit of this is that if you have aggregation over a very long we know like say all historical averages if you perform an exponential waiting on your feature it actually allows you it doesn't save you time but it allows you to truncate the window to a very small size without losing too much accuracy since all the tails are less important now the apart from feature caching purposes this kind of weighting is actually quite common for for machine learning purposes as well because in a lot of practical progress you do want to weigh more recent events more heavily of course there's always come to examples so in practice and to handle different types of access pad access and aggregation patterns we will most likely combine these two together so that's all the point most points I want to touch upon some of the key takeaways as Kara has already mentioned that real-time pipe ayamaro pipelines are already hard it's even harder as a sales force due to all that much tendency requirements and the common feature definitions are critical for correctness if you have you generate the features differently for your training for your scoring and fairly low time scoring over time the logic tends to drift apart and you are actually applying a model that is not applicable to your data and the multi-tenant isolation has to be incorporated in the design from the very beginning and by experience if you haven't working applications and you want to add multi-tenancy as a afterthought it usually doesn't work very well and of course without saying you have to maximize these sharing and minimize recomputation the last point is quite interesting we found that some boundaries like the aggregation patterns we allow has to be set for a real-time application and a generic platform if you allow cut client applications to perform arbitrary aggregations it would be very difficult if not impossible to guarantee to at the same time guarantee the real charm latency and performance so of course this these are only the things that we occur we have already done and we are current doing there are a lot of very interesting and challenging problems on the horizon first the top of that is is shared infrastructure for super large scale model serving currently out there there are a few frameworks capable of serving like thousands of machine many models at real time but in a truly multi talent setting say if we have 100 applications running our platform and we have 10,000 talents that will amount to 1 million models to serve them in real time would be a very big challenge both in terms of infrastructure and in terms of like software design also we with a more systematic way of defining the features and cashing the features it's possible to systematize the heuristics in scoring which means that we don't have to trigger scary scoring every time a feature changes we can set heuristic do say this feature has to be changed by more than 10% before we want to trigger is going this type of future heuristics is quite common but mostly implemented the inga very ad hoc way so a common feature definition and the few occasions have ability will allow this to be become more systematic and also we will not be talking about real-time streaming applications if we don't have a streaming data source and the streaming corporate databases is always a very challenging problem change data capture is a very hot topic in a number of sico and the no sequel databases currently and it's it's particularly hard because traditional psycho databases are not designed with this type of capabilities in mind so there are a lot of interesting and challenging problems lastly we would like to have a few registry where we can further abstract away our future definition so that we allow client applications to register their feature score body to enhance the discoverability and the shareability of our features so that's only a very light primer if you you are interested to learn more about instant platform we have several tops delve into more depths of different parts of the system tomorrow and Matthew will be talking in depth about this Optimus Prime framework which which is the heavy lift which is heavy lifting all our machine learning what froze and the Thomas will be talking about our execution engine and how we managed to run a huge number of SPARC jobs in parallel and now on Saturday and I will be talking about the dev upsides of things and the Chris and the deal will tell you how to make complex machines and in pipelines easy and if those are not enough we are always actively hiring thank you [Applause] yes I guess any questions yes the first question is do we measure the performance of the model in production yes one step that I didn't put on the on the architecture diagram is that we have a specific evaluator that runs to evaluate the performance of the performance of the models it's before the problem for now after Department we have a different set of like different set of metrics yes and we have a metric system that their feedback these metrics sorry the fidget is me sorry I can so for now the the scoring process is made mostly using the latest the feature value as well other aggregated value right so we are not actually keeping the history in check but of course this is something that if we have global registry of features we can keep a longer history of features and to track these movements sorry I think that second part the first question is what database are we using and I assume you're referring to the future star yeah so for now we are using contender for the given the song for now we are not using that yes but that's something yeah in monetary monetary yes for feature story is only used in scoring because it's mostly meant to achieve the cashing purposes so that you need you really need a scarf in in a very short time period yeah okay one second this way with a face [Laughter] yes so that's that's where the actual Education intermediate aggregation stage comes into play of course we can argue that a week if we allow us to make the intermediate state that officially complex we can pay you can effectively store the entire history but that is of course what something that you don't want to do so it depends on the exact allocation we may need to approximate we may need to do some exponential waiting to ignore some of the data and perform the aggregation on the fly over a subset of the history department it all depends on what type of application you want to do yes it's a tradition results so for now we are using a simple model update caches or cache the model in the in the flow yeah but of course this is this in the long term we need to have a more scalable solution to achieve that thank you [Applause] you