Scale By The Bay 2019: Savin Goyal & Ville Tuulos, Human-Centric ML Infrastructure at Netflix
[Music] all right welcome to scale by the way good morning it's very early in the morning thank you for making it here my name is seven and today along with my colleague Willie we are going to talk about the ml infrastructure that we built at Netflix to handle a wide diversity of problems that our data scientists face now all of you might be wondering what are the actual data science problems that Netflix alts or a lot of you the first contact that you have with Netflix and ml at Netflix would be around recommendations that you see on our service but that's mostly just above the iceberg we are one of the world's biggest studio in the world we have a Content budget that is in the billions so every single day our content executives need to make a decision how much to pay for any given piece of IP how to efficiently produce any piece of content we're in a lot of data science goes and we are also one of the biggest subscription services on the Internet so a lot of decision and goes beyond behind fighting fraud account takeovers we operate one of the biggest CDN services on the internet as well so how do we optimally place our videos on those CDN caches so our data scientists are inherently working on a really really wide diversity of problems and they need an infrastructure that sort of like scales out and handles that I was sitting pretty easily so my team over the last two years we have built this machine learning platform called meta flow which has predominately targeted towards users in the pythonic and the art world and today in this talk we are going to focus on two major pillars of this platform inherently reproducibility and scalability so for the first 15 minutes I'll focus on reproducibility and then my colleague will talk about how we deal with some of the scalability concerns and our philosophy behind it and while all this while we wanted to make sure that our data scientists can focus specifically on data science and all the infrastructure concerns are so like transparent to them so a big focus of this talk as well as our platform is on making sure that the construct the abstractions that are provided to our users are inherently high usable by them so let's begin so a lot of you like the software engineers as well as the data scientists in the room would be familiar with you know like situations like these where you have some amount of code let's say you were training a model on tensorflow and all of a sudden next day it stopped working even though you didn't really take any explicit action its are like mutated the state of your code and what's happening behind the scenes is that your code inherently is interacting with external systems in this case tens of your brain saying myriad of dependencies many of them are unpin and they can change in often unforeseen ways right as software engineers we are all tuned to identify those issues and fix those because that's sort of like our bread and butter but then when it comes to data scientists then they are asked to be sort of like philostrate software engineers and when they have to figure out like what has gone wrong and weird and that's all like asking for a lot from data scientists and that has a clear impact on their productivity if let's say you know a data scientist is following good hygiene they are pending all of their dependencies they're like very well-versed in how to maintain and manage their systems other people will have a really hard time replicating their environment and sort of like reproducing their results and then building on top of that and that has sort of like given rise to this culture of team of one in many data science organizations where every single data scientist I saw like tasks with one singular project and there's not a lot of collaboration happening on the model inherently person so these problems are inherently just rooted in this notion of reproducibility where you want to make sure that you are able to execute what executed before in precisely the same way and expect more or less coherent results and if we saw like go deeper into like what this notion of reproducibility means at the end of the day it's composed of two major portions it impotency which is inherently if you are able to execute your code then you are solid like guarantee consistent effects and repeatability which is merely the ability to actually execute something over and over again and if we look at it a little bit more concretely the notion of edom potency lies in the user code which has been surrounded by a repeatable compute environment and then this entire unit of work is something that you can take and execute on any sort of like physical execution system and sort of like be guaranteed consistent results over into working in the case of machine learning workflows your machine learning hopefully would inherently be a composition of these unit of works and in this case you know you might have a bunch of different steps like data prep training model fitting them and then deploying them into some sort of like production system and you want to sort of like guarantee that each of these nodes of your workflow are sort of like reproducible in certain sense so that then you don't really run into unintended behaviors now if we look into what actually goes on in this unit of work and how do we actually go about thinking about making it reproducible let's peel up the covers a little bit more let's look at like what exactly is in this repeatable environment what are the dependencies of this it impotent queues of code that sort of like needs to be packaged and shipped around in a consistent manner so notionally the two big dependencies that any sort of code has in machine learning like data dependencies and libraries and it's not just the data dependencies these are like fixed in time they're also moving as time progresses so for any sort of like notion of reproducibility we need to figure out how to encapsulate and capture any changes in user code libraries as well as data dependencies and how do we sort of like go about reasoning about those which are like leads us to inherently a versioning problem in the case of machine learning systems so how do we do it with versioning for user code for libraries for data and in the software engineering world we have solutions for all of these three so people are familiar with like portion control it's been around for like more than three decades and you know get started like to divorce for libraries a lot of people use docker to sort like bake their dependencies and then execute their user code on top of that and for data when it comes to sort of like this functional paradigm of data people rely on immutable snapshots of data so that they can just like keep the pointer to data around and then just assume that the eyes are like frozen in time all of these strategies in isolation they work well but then machine learning system is sort of like an amalgamation of the user code libraries and data and we need sort of like a solution that sort of takes care of all of these three things in the same similar manner without actually placing a lot of tax on the user which in this case is going to be a data scientist let's actually walk through like what might be some of the big issues with some of these systems and see if we can solve like come up with our alternator which might so like give us the exact same capabilities that these systems provide but with a much better usability function now let's say I'm a data scientist my work is exploratory in nature I'll be writing some code and then I'll be executing it with a bunch of hyper parameters at the end of it I'll have a model I'll evaluate at using some metrics if I'm dissatisfied then I'll tweak some hyperparameters I'll tweak some code so during any given day I am making a lot of changes to my code and I want to step back and see what my state of code was in any of my previous experiments and one obvious way to do that is every single time I execute my code right before that I'm sort of like doing a kid commit and storing that commit somewhere annotating it but like what were the hyper parameters what was the inherently the context of that execution but now the big problem here is that if I forget to do a git commit at any point in time I've sort of like lost state and we can definitely rely on like some homegrown discipline and data scientists you saw like get around this issue but even then it's not really that usable right like as a data scientist I shouldn't be just looking at like okay what a good push what's a good commits what's a good pull that's just like asking for a lot so one thing that we can do here which is like very similar to how people think about data is just treating data and just reading code and any sort of like hyper parameters as data in itself and what we do in our machine learning platform is every single time there's any execution of code we inherently snapshot that entire code and we ship it around as we ship around data and treat it and exactly the same way let's talk about docker about how do we manage libraries right so this is like one standard wave here a data scientist will come in let's say they are training a model using pandas intensive flow they'll create a docker file pip install pandas pipe installed tensorflow they'll have some CI CD process they'll create an image and then they'll execute their machine learning model on top of that image maybe a training process or some evaluation process all of this is doable it's not really asking for more but still it has a lot of overhead a lot of unforeseen overhead the number one is that ok do you create one docker image for your entire machine learning pipeline as you saw before a machine learning pipeline has a variety of different states you might be training a model you might be deploying a model all of these different states they bring in different dependencies or not you create one docker file with all of those different dependencies if you do then you run into dependency management hell where you constantly trying to figure out how those transitive dependencies are like dealing with one another if you choose to go with an alternative approach where every single node has its own sort of like docker image then you run into an image management tool because now you have to constantly figure out like ok which step has what image is that image being maintained if you have any updates are those being propagated to all of the images or not so one way or the other you have to sort like pick your poison whether you want to be dealing with like many dependencies in one image or multiple images in respective of that there are certain other issues that you will soon run into as a data scientist let's say you had panels and tensorflow and you decided that hey I want a newer version of dense pandas maybe there's some new functionality that you want to use now as soon as you switch out a switch to a newer version of pandas and you decide to well your docker image again you would end up bringing in an entirely new version of tensor feel just because you hadn't pinned your version of tens of you right so you can definitely rely on pinning your dependencies but that sort of like doesn't really get you all the way because all of these packages that you are bringing in they have transitive dependencies and often these transitive dependencies are not really pinned so every single time you build your docker file you are sort of like not really guaranteed the same result so while we are talking about it important seeing user code like we need to make sure that our infrastructure our library declarations are sort of like also add important in some nature also when it comes to machine learning code you are not energy even though you know you might be writing your code in Python your dependencies are not always in the Python equal if you want to use tensorflow GPU then you need to understand what version of CUDA CUDA daeun and you're trying to bring in and at times this means that you're not only in the ecosystem of like trying to figure out what your dependencies are but you need to know so like learn and intimately understand what are some of the other package managers apt-get in this example again most of these problems that I'm referring to our problems that a lot of people deal with these problems have existed for a really long time and people have built solutions as well for these for example Conda is a package manager that works across python as well as system package dependencies it worked really well and as being embraced by a lot of dear scientists but now when you look at it look at this core of the problem let's say you're using Conda then what is exactly docker bringing into the equation then docker is inherently kind of like an overhead that is causing death by a thousand cuts to our data scientist and that's sort of like the approach that we took when we were building meta flow where we inherently came up with this UX where we asked our users to inherently just specify their dependencies using a decorator at Conda in this case specifying what sort of libraries and Python version they want and then behind the scenes we actually snapshot all the dependencies we create an equivalent of the see log file and then we ship this entire execution environment wherever they want to execute whether it's on the local box or whether it's on some sort of container orchestration system the input to their code as discussed before could be any sort of like immutable snapshots which can reside in any sort of cloud environment and their entire code every single time they execute we actually package this entire code the libraries and the pointer to the big data and we just move it all around and that sort of like leads me to the next question like okay what is the physical substrate over which you actually end up executing your code and for that I have villi we'll talk about it yeah thanks Alvin so yeah that's fine so what Simon was talking about is a software solution for having these reproducible units of computation in a machine learning workflow so that's all software but of course I mean the software ultimately it needs to execute on some hardware so it's it's not like a abstract thing so all the metal we need some kind of a server like we do need to execute this code somewhere and now if we take a little bit of step back that like what does it actually mean to execute arbitrary user code on the server of course there's a question that well do we know that the server is big enough do we know that it scales and hence this part is about scalability so now if you if you consider that like you have these units of computation that sabine was talking about there in the middle and in order to be able to execute this whole thing with the user-defined code as well as the dependencies on some server you need a big enough server so you have a circle and like you find a big enough server to execute the code on and now like the question of scalability is that what happens when the circle crows what if you need more memory what if you need more CPU what if you need more cheap use and now as my six-year-old could tell me if you have a big circle and like you need the box around if you need a bigger box so you need a big server to run it is like a big units of computation and and this approach to scalability has been of course around since the very beginning of shooting and it's called the vertical scalability which means that like we just keep finding bigger and bigger servers every time we have a bigger unit of computation now many of you of course I can imagine that the problem is that you can't just like keep finding bigger and bigger servers infinity so it might be problematic so now what people especially in this conference series have been talking and thinking about for like at least the past 10 years is this idea of horizontal scalability where instead of like trying to find always bigger and bigger boxes what if you can divide and conquer but if you can start your problem in much smaller pieces and then you can find a like uniform sized boxes to execute this computation on and this model is particularly appealing if you have a data center that just happens to have let's say 10,000 servers and you know that it's it's very homogeneous and makes operations easier so each one of these servers have 32 gigabytes of RAM so now you just like split any problem to small enough chunks that you can execute them let's say using less than or like using less than 32 gigabytes of RAM and and off you go so technically sounds good well now the question is that like what could be problems with this horizontal scalability approach so for instance like a couple of years back there was this like interesting paper you can google it scalability but at what cost and what this paper was showing that for a certain kind of a graph processing craft querying problems a single threaded implementation performed better than these horizontally scalable systems so how could that be I mean how can you beat with a single CPU a cluster of many CPUs and and the reason for that is simply that the horizontal scalability comes with overhead so every time you you you have network communication you have network overhead like now when you have two kind of a shard your problem in different ways I mean that comes with an overhead and now it might be that the overhead ends up dominating that kind of the total cost of computation in any unless you do it very very carefully and that's only about performance so now there are the kinds of overhead as well so this paper is all about performance but also as you know there's operational overhead in in managing these distributed clusters so any of you have maintained a spark cluster or presto cluster know that well I mean things can go wrong and like like you need to have expertise to know how to deep these things and of course if there's a certain probability that the single server goes down and let's say it goes down once a month and you have a single server then you can expect that it goes down once a month if you have hundred of those servers you can be pretty much sure that like something goes wrong every day and of course like over the past twenty years or so we have learned how to deal with these issues on the software layer but I would argue that one thing that's missing I mean although we technically know how to do it is that like we haven't quite solved the usability question how do we make this disappear horizontally scalable systems usable as usable as as things that are vertically scalable so just to give you an example that um if you had a small enough data set let's say it's a hundred megabytes for a gigabyte so really small now the question is that what do you use horizontally scalable system to analyze this data set and now what you might be thinking that's that like well if you have a spark cluster already and there's no downside using it that's there but I mean let's say that like you're a data scientist this is what you need to do like probably the easiest way what you might consider is that you have your laptop and your laptop has 16 gigabytes of RAM and now you get the 4 gigabyte data set so it's a no-brainer you open a Jupiter notebook you load your 4 gigabyte data set in a panda's data frame and like you have a lot of lot of light Headroom there like to be even copies of the data frame and you can do everything in your notebook you don't even have to think about anything it just works beautifully and like if something goes wrong you can understand the Python stack Tracy's probably the biggest issue when it comes to the hardware is that your laptop runs out of battery and you can understand like what's going on now the question about scalability well the thing is that like if let's say instead of 4 gigabytes if 10 gigabytes now even with the 10 gigabyte data frame you do exactly the same thing you execute the same trooper notebook the same code nothing changes the failure patterns are the same it's all like works beautifully maybe it's slightly slower because now you have twice the amount of data well obviously tech would like you can see where this is going so eventually you have like a big enough data frame that like it doesn't fit in memory anymore and like then you're you're like a notebook fails in a mysterious way you get out of memory error and so forth and now if any of you have worked as the kind of an infrastructure devops person in an environment like in helping data scientists then typically the next discussion that happens system of this kind so you have the data scientist in the blue shirt I mean to you saying that okay the data set is too large for my laptop and could I get the bigger laptop so I can just execute my jupiter notebook because it worked for four gigabytes it worked for ten gigabytes well I mean I could just like kind of a get it running for 20 gigabytes as well well now you as a responsible DevOps person your answer is that look you are doing it wrong I mean like that line of thinking is not scalable so even if I had in my back room like a laptop with 32 gigabytes of RAM and I gave it to you yes you could execute it with 20 gigabyte data set but next time when you have a 60 gigabyte data set the same thing would happen all over again and like I can't keep giving you bigger and bigger laptops all the time so change your approach you should like you SPARC do something differently and and now it's it's fair and technically speaking even the data scientist knows that yeah that line of reasoning kind of makes sense but the fact is that I just wanted to go home and watch Netflix so can I just please execute my superior notebook and can I just get the bigger laptop and in the usability point of view that is exactly right I mean that what that's what we looked like to have and now the problem with this is obviously that well that actually the laptop that I had about ten years ago is actually not that different than the laptop that I have today I mean they just like haven't like it changed so much but there actually are the things that I have changed quite drastically over the past ten years so instead of like running this stuff on a laptop what we do at Netflix this you may know Netflix is pretty much hundred percent on Amazon Web Services so the question is that like well I mean what does this mean in the context of a cloud and here's an interesting chart that I pulled like for you for this presentation and what it shows is that it shows Amazon Web Services since the beginning of time since 2006 when Amazon Web Services launched in in public beta and it goes all the way on the exit x-axis to 2019 which is today and on the y-axis we have the maximum amount of memory that's available on the largest kind of a normal general-purpose instance on Amazon ec2 at any given point in time now that the small disclaimer here is that by general purpose instance I mean that Amazon has these special instances like FPGA instances bare-metal instances so Ford for many practical purposes it might not be like very convenient to use them for like this general compute so let's just focus on the basic instance it's like m instances and our instances and so forth and now in 2006 Amazon launched with one instance type it was the m1 small that had 1.7 gigabytes of RAM and then like about a year later they launched the m18 extra large with 15 gigabytes of RAM then like 2010 m2 a 4x large 68 gigabytes of RAM then finally 2013 like I remember being super excited when they launched cr1 a text large with 244 gigabytes of RAM and I believe I was probably like one of the biggest users like of these instances at my previous job for the longest time because it was only in 2016 and 17 when they finally launched the RF for 16 extra large with 480 80 gigabytes of RAM and now this year they launched the r5 24x large with 768 gigabytes of ram and now a couple of things here well I mean the first thing is that like you can see that this is pretty much following an exponential curve and the other just a thought exercise for you is that like many of you have been coming to this conference like remember for instance that the original Google MapReduce paper came out 2004 Hadoop started gaining traction in 2007 2008 so now if you imagine yourself standing there in like 2007 2008 time frame you might think that the horizontal scalability is the only way to go because all we have available easily in our own data center and in AWS they are they small instances with with 1.7 gigabytes of RAM so really it felt that that's the only way to go but now something that's really important to consider is that now today we have 500 times more memory on these machines than like what it used to be about 12 years ago so big difference there so what does this mean in practice so now imagine this like a very practical use case scenario for for Netflix so we have we have like like many machine learning problems that are often nature that we want to analyze what our what our users mut our members are doing what kind of content they would like to see so imagine that you have this data frame that has every single Netflix member as a row and today we have about 250 million members and then some number of features for every member as a column so big data frame and like for a convenience of the math imagine that you have one byte per column so now if you imagine that like we have thousand features for every user and and like we have kept like every single subscriber in memory the interesting thing here is that every ever like ever since the very beginning of AWS we have been able to keep this data frame in memory so since 2008 all the way now to 2019 we have been able to keep this data frame with all Netflix subscribers and maybe thousand features in memory so that's that's that's pretty interesting and now of course you might think that well I mean yeah but it's only because like one kilobyte is not a lot of data so what happens if we had three kilobytes of data per member well now interesting thing here is that like you can see that around like 2008 2009 we had a situation like where we couldn't keep the data frame in memory and like you might think that if you are a distributed systems engineer like standing here 2008 2009 like you can't keep the data in memory tip the feeling might be that well we just have to change our strategy which we have to change our paradigm because we have like we have reached the point kind of the escape velocity that the amount of data is still large that we possibly can't keep working this way anymore and somehow the idea is that the day that just keeps growing but the machines don't keep growing so it's really that it's like a shooting up in the sky but uh but the machines stay small and there is like a like kind of a slight biasing thinking there because actually the machines are growing exponentially as well in this case looking at the memory dimension and and like all those companies like Netflix are growing pretty exponentially too I mean they kind of a keep keep at the same pace now of course you can imagine that you can keep going here let's say we have four kilobytes of data per user like while definitely getting hotter you can imagine at 10 kilobytes it's like hundred kilobytes this chart is getting all red and it doesn't work so it's easy to have the counter-argument that of course like a vertical scalability isn't the Silver Bullet and you can't solve everything with that approach and that's of course absolutely true but the point that I'm trying to make here is that the number of different and the volume of different workloads that you are able to handle with this very usable vertical scalability approach is actually growing exponentially and of course that Netflix we can't solve every single problem with this approach but what's really important is that we can solve an increase number of problems with this approach so like here the yellow chart it like depicts all those cases where we can like make provide a really nice experience for the data scientist but by saying that look it works exactly the same way as on your laptop so there's no new paradigm you don't have to change anything very convenient so what does this mean in practice in meta flow so in the same way as Simon was explaining that you can define your execution environment you can just define that ok I need 10 gigabytes of RAM to execute this function and like we give you the right size server or like maybe if you need 500 gigabytes RAM we can give you a 500 it server with 500 gigabytes and of course all this works in there in the container management system so like we just provision the amount of fly from that from the cloud and and so forth so very very straightforward and now an important like additional dimension here is that because this is the cloud they are not physical laptops not only we can give you a one laptop what quote-unquote laptop but we can give you an any number of these machines in parallel and now there is a subtle and important difference to the idea of horizontal scalability so where it's like with the horizontal scalability the idea is that you are kind of sharding your problem automatically let's say we put put sparked us with with our DS and like other other other ways of like splitting the problem automatically here we are treating these user-defined functions that as Adamic units of computation like someone was saying so we are not like trying to automatically split them but we are just like having multiple copies of the same computation maybe with different parameter stations and this is very convenient in many machine learning workflows where let's say you want to do things like hyper a parameter search so so that that allows our users both like write their code in a very straightforward way the same way how they would be writing it on their notebook and like when they need more resources they can just say that ok give me a quote unquote a bigger laptop and like let's say they want to like a train to say model for every country in the world or if they want to like it run it in any any way they want to be the fan-out I mean that's that's easily doable and we provide abstraction for that in the in the in the Python code base and now this kind of concludes my talk so so severe like I started by talking about the reproducibility which is that how do we define how do we make it very easy for users to define these closures that both include their user code as well as the full environmental the repeatable executed execution environment that make sure that the results are always reproducible and you can make sure that you are not interfering with production and then I talked about the scalability side of things how do we make sure that like we always have the hardware at our disposal so we can actually execute these units of computation and like that the overarching theme theme here is not to say that we proposing something that's very new you're different and technically what people have done before but really that the kind of the backdrop here is the humans interested the idea is that like how do we make it as usable as possible that's it thank you thank you Vela and Sabine we have time for one question is there any question in the floor none all right thank you very much Thanks [Applause] you