Scale By The Bay 2020:Matei Zaharia,Scaling Databricks to Run Data & AI Workloads on Millions of VMs
[Music] okay great so so yeah so i'm going to give you know a different kind of talk here than maybe you've seen me give at uh most conferences i'm not going to talk about apache spark or new open source projects or anything uh like that that we're developing uh to help uh people do data processing i'm actually going to talk about building the databricks our service itself and some of the lessons we learned from that um and today uh database you know because it runs uh large-scale workloads on behalf of uh enterprise customers it's actually operating at a very high scale i wouldn't be surprised actually if we manage more vms uh than any other private company at the moment because all the workloads that people on on databricks are large scale and uh you know we we launch and manage these vms on uh public clouds on behalf of our customers so we had to learn a lot of things to do this and uh you know carefully design everything that we're building to support these kinds of workloads and i think there are you know maybe interesting uh lessons for everyone looking uh into scaling services and in general into designing their systems uh so that they scale well and on top of that a lot of what we do is using scholar as well so you can see some uh some cool ways that we're using it um so this is a talk about you know with really work that's happening across our whole engineering team these are some of the folks that i actually work on or at least lead a lot of the platform teams that i'm going to talk about but it's really a major part of what the whole company does so i'm going to talk a little bit about the database sort of service itself what it provides to give you a sense of what it is and maybe what's uh challenging about supporting that um next i'm going to talk about uh the specific challenges in developing a large-scale cloud service like this for you know enterprise b2b customers to use and finally i'm going to drill into four lessons from our experience uh which go you know from how to build the service to how to test it to actually things that we are doing uh in spark and big data processing technology to make it work better in the cloud based on you know lessons we learned from what what uh issues customers ran into okay so i'll start with the first one so databricks itself is uh you know as a startup company founded in 2013 by the apache spark research group at uc berkeley so the group that you know initially created the project and uh you know as the open source project was taking off we really wanted to continue building it and we also thought you know you need more than just this open source engine to actually uh enable lots of companies and lots of users to do things with our data and we decided that a cloud service was actually the best bet uh to really make this accessible to the broadest um audience possible so we started building uh you know a cloud cloud-based data and machine learning platform on top of public clouds and today it's grown to about 7 000 customers across aws and azure and also pretty large scale of the workloads we're running millions of vms as i said processing uh exabytes of data each day and there are also hundreds of thousands of users on the platform which are the data scientists and engineers and so on and our company is going as well but it's actually you know it's not that large given the skill of what we're doing um we have about um 1500 employees 300 engineers um and uh we just announced um that last quarter we we cost 350 million in ar so if you look at it you know for each engineer there are basically uh tens of thousands of um vms running workloads in there that have to keep working well or else the engineer will get paged okay so what what is the product itself what do customers do and what do they expect out of it um so basically our users are professionals working with data it could be data scientists data engineers or business users who are using things like dashboards or bi tools and we have a cloud hosted you know data platform the kind of data platform you might you know consider building yourself in your company um and it's it's it includes a bunch of services so it includes a data science workspace for interactive queries with uh notebooks and it's all that's all a collaborative service where many users can work together uh it includes uh sql analytics this is actually a new uh product that we just launched today which is basically a sql focused uh workspace based on the open source uh redash project uh and it's designed for data analysts to use it also lets you connect with uh tools like tableau and click and and so on um it also includes a job scheduler this is a huge driver of the workload because every time you build some useful analysis you probably want to turn it into a production job and then you expect that to work reliably and it also includes other things like machine learning platform features data catalog security and so on so the key thing though is that all of the workloads running on this are usually working with large amounts of data so they're each parallel um and uh so so we have to scale in that direction you have to make sure that we provide a sql service or data science uh you know notebook experience or whatever that that works with large scale and also a huge amount of the workloads are production workloads you know if you do something successful like you design a machine learning model or you design uh you know a data pipeline that's going to load and process your data you want that to keep running and so reliability is is extremely important uh even as the data sizes go over time and so on uh ideally you know these users would be able to set up a pipeline and then forget about it and you know never have to go back and deal with errors in that or anything like that so that's kind of those are the services that the users see now under the hood where i do all the workloads on so actually we set things up so that the workloads aren't in the customer's own cloud account so basically the customer just gives us an iam role for launching uh instances in the account and accessing you know parts of their storage bucket uh and then we just we deploy um clusters in there we have our software on it called on the clusters called databricks runtime and it directly accesses their storage so the the nice thing about this is that customers own all the storage and also if the customer has you know a special virtual network they want things to run on or um you know they have a special discount or reserved instances or whatever with the cloud provider um they can benefit from all those you know as as we on the workload over there um and in terms of what we actually run most of the uh stuff uh we we run on the clusters is based on open source um so of course apache spark lots of other open source data and ml uh products and in fact since we started the company we actually launched two new uh widely used open source projects uh delta lake for storage management and ml flow for a machine learning platform so you know even though there's all this talk about like well the cloud providers are are they just going to to take you know to to use your open source project um and compete with you um uh we actually thought that it's it's very important to uh launch some of these things so that they can be standard in the community uh and uh and really uh have a you know basically um benefit from the openness uh that you get using that and then the other thing i wanted to mention is uh in both our services and the the runtime we use uh scholar uh you know pretty heavily actually uh we only just recently launched uh basically our first uh non-scholar service uh uh tehran re-dash because v-dash is a python application but pretty much everything else is scholar so we have a lot of interesting experience about just managing you know large scholar projects and testing and upgrading and all that stuff as well if you uh you know if you want to hear about that okay uh how how do people use it so these are just some examples you know kind of the sales slide with the wall of uh logos who are using um uh the the product and uh you know the the key point though is that companies in all industries are actually working with uh very large data sets these days for for pretty important internal use cases um so it's not it's definitely not just tech companies who need these anymore um and just to give you a few representative examples uh first of all health and life science is one of the biggest areas uh where people are collecting and working with large data sets um you know one one use case we we have that regeneron for example um is correlating uh patient records from um half a million patients with their dna so the dna is you know many uh you know potentially gigabytes of data per patient and they want to correlate these together to design therapies that are tailored to your genome um and you know maybe a drug that works on some people uh and doesn't work on others maybe you can figure out when that will happen based on the genome um so this in itself is a is a you know very large use case um another one is large industrial companies so companies like shell and others have instrumented you know pretty much every uh you know every uh machine in a in a in a factory or in a chemical plant or whatever uh to produce you know really detailed data about how it's operating uh and what kind of um inventory is in there and so on so they're using day one large scale simulations machine learning and just basically large-scale analytics to optimize that whole process and then another cool use case i wanted to call out is finra which is in the us is the financial industry regulatory authority so basically their job is to look at all the trades that happen on stock market securities markets in the u.s and find um find basically illegal trading and so they have 30 petabytes of data and growing uh uh with all those trades and they're looking uh in that for various patterns uh using machine learning to you know to find people who are uh you know doing illegal things like insider trading so these are just some you know example of you know what what people will do with big data outside of uh kind of a tech company okay and because of uh you know the the usage um because of the scale of these and because of the fact that each application also tends to grow over time as you collect you know more data at higher fidelity uh we also have to deal with very large usage of our platform so this block here is showing the number of virtual machines were running each day over the past um uh three years and you can see it's been growing uh at about a factor of three per year actually in fact even uh this year with covet and all that um and um uh you know this is a combination of of course uh bringing in new customers but also existing customers and existing workloads going over time so we have to make sure that everything we build works with these and uh you know these days even new customers you bring on will already have maybe petabytes of data somewhere and will want to run on that so they'll bring in a large uh chunk of workload um uh just just by coming onto the platform so what's challenging about all this about building this so there are basically two challenges and we have to think about both of them uh because they come together in our products so the first one is all the challenges of building business-to-business cloud services when you build something that another company is going to rely on uh it needs to be highly available uh it needs to be secure um uh it's uh if you have any component that's shared uh that you know has information about multiple customers that needs to isolate them from each other so it needs to be multi-tenant um and also you need to be able to update it and push changes and do maintenance without breaking the workloads it's especially important because as i said a lot of our workloads are these production pipelines where you know as the user you hope that you set it up once and then you never have to worry about it and it'll keep working and producing that report or that uh model or whatever it is uh that uh you know that your uh uh business uh depends on so that is is challenging you know in itself whether you're uh building uh data breaks or you're just uh building you know like an email or spreadsheet or whatever issue tracker for uh you know other businesses to use uh but then of course on our side um we also have the challenges of large-scale data processing all these workloads are a large scale and if you think of all you know the the work you have to do maybe to keep a data pipeline and the software for it running you know inside just one company wherever you work uh you know imagine having to do that for seven thousand companies um so obviously everything uh has to be um you know as resilient as possible without human intervention and when there is human intervention it has to be very easy to fix uh and to understand what's going on um so that's that's kind of the second challenge that we have to deal with so i'll talk about both of them uh later on to show you ways that we've uh you know we've dealt with some of them so let me start let me explain these in a little bit more detail i'll start with just the cloud one to explain what's hard there so i think everyone you know maybe probably knows this but um basically developing software and selling software and the cloud is quite different from the traditional model traditionally you know before the cloud software was was sold you know using these packaged releases and basically you would work a team would work on something like let's say the next generation of oracle database or whatever windows or something like that um and you you would you know you would make releases anywhere from every six months to every few years and so they would work on it test that package it into a release and then the customers would be responsible for operating it so they just get a bunch of bits on it on a cd-rom maybe they get a support contract but they uh they have to install it and manage it and operate it um with the cloud software uh the vendor does the operations for them that's basically the difference it means you can release updates a lot faster you usually you know you'll release every one or two weeks uh sometimes maybe even every day depending how you how you set up your infrastructure and uh it also means that your customers need uh fewer uh basically uh less operations work to support the same number of users so they can bring more users on the platform um and or you know they can spend their time doing more interesting things but it also means that you take on this responsibility so from their customers perspective if everything is working well they expect to get a number of benefits from the cloud software uh most importantly they expect uh you you the cloud vendor to do all the management and this is something that's really valuable this is one of the reason that cloud services uh uh you know will you know may cost more uh then and they may command higher revenue and higher margins and so on uh uh then um you know than just selling someone some some bits uh because basically uh a lot of the value is the security higher availability and so on out of the management um they might also they generally will expect elasticity as well everything should be pay as you go they should be able to bring in workloads just on demand and uh you know when you bring in a new workload it should it should always work um and the other thing that they usually expect because um it's in the cloud and you can release updates anytime you uh you will get new features released faster you can give a team feedback and get a bug fixed or get a you know a new feature launched within uh a few weeks maybe um because you know they can update your software anytime so these are these are what the customers um expect uh and obviously they're valuable and it's great to build a product that has these properties um but building them and actually delivering on those uh is also hard um there are you know many issues you have to deal with and uh that you wouldn't have in traditional package software um and so for example um one one of the top ones is is building a multi-tenant service how do you design you know the the control plane on the cloud provider's side to scale well with lots of customers to provide strong security by construction so nothing can go wrong and you know show information to the wrong person and also to i isolate the customers from each other in terms of performance and faults and so on for an on-premise software or even an open source project like apache spark uh you don't have to worry about this because you just provide one instance of the software and your user manages it but for here you have to worry about not not how how do i provide like you know one instance of apache spark or of a database but how do i let anyone spin one up and make sure they don't interfere with each other in any way second challenge is of course operating the service um security availability monitoring and so on so you have to take that on as a cloud software provider uh although arguably you know it's uh the outcome is hopefully better than if you asked each customer to do it on their own because you'd be better at it from doing a lot of it um and then the third one is uh is is upgrading without regression so um you know unlike uh something where you you tell your customer you know please update to the next version of this and they can all let out over you know a couple of years they can even keep critical workloads back uh in the cloud you're trying to have one instance of the software that you're continuously updating um and um uh it's critical for this to work to always work and to not regress um because otherwise users are not going to trust your cloud and so this includes everything from the api to the uh the semantics of different operations the performance regressions and so on uh whenever you make a release if any of these things are you know getting worse you have to be able to undo the release and and fix these issues um and it's it's a lot more work but of course that leads to a product that evolves faster and you know gives the users better features faster so it's really important to figure out how to do this well if you want to compete with other cloud services for example okay and of course on top of the challenges of cloud uh the the challenges of large scale which you know people are probably all familiar with here so uh each individual app needs to deal with failures and stragglers and all that stuff uh uh also uh an interesting one uh you know that we've seen is each customer app you know if it starts making a lot of calls into the control plane it could certainly overload the services that we're providing so you need to guard against that you know imagine a spark job that launches you know 10 000 tasks that are each trying to you know to change permissions on something in in the workspace or whatever it is they're trying to do um and the other one uh is of course the weird failure modes that you only see you know happens one in a million times but it means you're gonna see it when you have you know a few million vms okay so that's uh kind of just an overview of the the kind of things uh that are challenging i hope it's given you a sense of what the you know what the customers expect and what you know we hope to deliver and why it's hard um so next i want to dive more deeply into into four you know interesting things we learned uh you know based on our experience uh and some of the things that we're doing in response to them and you know some hopefully you know a bunch of these uh will apply or at least they'll be insightful for uh things you might be working on as well whether you are working on large-scale data processing or building a cloud service or something else so i'll talk i'll talk about what actually goes on you know in cloud services in our operational experience uh how to test for scalability and stability the stuff that i i mentioned is um can be quite hard to guarantee um how to develop control planes so this is our service framework and infrastructure for that and finally a bit about evolving big data systems themselves to work well in the cloud and to work well you know when you have to manage uh 7000 or you know maybe even more uh instances of them for each of the workloads so this includes some open source uh work that we're doing in the delta lake project so let me start with this one so a lot of things you know you might imagine could go wrong uh in building this kind of application and we're always keeping track of like what if you know if anything is causing problems um so uh but but in our experience there are actually some uh very common themes that come up um so to give you a sense this is a plot of some of the uh some of the issues that cause significant outages uh for us at databricks over uh a three-month period last year uh basically and by significant outages i mean something that took down uh production workloads for customers so this would be things like scheduled jobs and so on um because that's the thing you know if those go down of course they'll be hate right and so on later but it it means that they might not deliver their results on time and it you know it costs money and so on um so that's what i'm i'm counting here so you can see there are a number of causes and a lot of them have to do with scaling the uh you know the top ones were scaling problem in some of our services where they couldn't cope with some dimension of the load coming in or scaling problems interestingly in the underlying cloud provider services that we rely on which are things like the uh cluster manager uh that gives you vms when you request a vm from the cloud provider or the virtual network or sometimes storage uh so interestingly some of those can have issues too um another common one was insufficient user isolation so you what i mentioned before you've got one user that's making an unusual kind of pattern of calls and that causes something to tip over and it affects other users and then deployment misconfiguration and other a lot of the ones that other have to do with regressions that we couldn't find during testing but there's other stuff in there as well um so you can see that at least 70 percent of them are scale related if you were providing just some kind of package software that someone installs and operates on their own uh they wouldn't really have any of these issues uh but of course you know we want to provide something better where you know we will manage it and give you you know best in class uh availability for for those workloads so here are just some uh concrete examples of the issues um and i'll also give uh you know a more detailed one later to show how different things can interact so cloud networks are actually um you know a common um uh source of issues depending on how you're configuring things uh because these are you know even though these are virtual networks there are a whole bunch of limits in there for example how many entries can be in a routing table you know how how uh how quickly can a machine talk to thousands of other machines and so on uh and also some and some of the the failure behaviors that you get there can be quite weird where for example packets are flowing one way but not another way um automated apps creating large load i mentioned this you know you let people call something uh you know on thousands of machines and place a load against your system it you better be able to handle that um very large requests and results and so on we've seen everything from you know the sql query that's uh you know that's that's uh you know many like uh hundreds of thousands of characters long to the single individual record in your a table that's you know gigabytes in size so all all these things you know all the assumptions you might have about sizes of things will probably be blown away um and slow vm launches and shutdowns and lack of vm capacity um from you know as i said this cluster manager inside each cloud provider uh and even things like data corruption writing uh the cloud storage so all these kind of things can happen oh i see some questions in the chat um uh can you talk about the other 20 outages i will talk about some more of them um later i think maybe a question i did mention a few things uh and then eric asked his databricks aws golf cloud certified uh yes it is available on golf cloud and um that's actually another thing i didn't go into too much in this talk but i'll go to it later which is how do you design you know the same infrastructure to work well on many clouds and across sort of constrained environments like golf clouds that's a that's an interesting one as well yeah so these are you know so these are just some examples of the issues um i'll give you one concrete example of an outage because it it it shows a few of the things that can happen um and why it's kind of hard to to do some of these things uh so i'll start by so so this is this was a problem involving our job service in one region so the job service is something that launches and tracks jobs and clusters on a schedule so you say for example on the spark job you know every two hours to ingest some data and you know convert it to a different format or compute a report or whatever it is um so the drop services is sitting in our control plane let's say an r aws account or azure account or whatever and then it's um uh you know it's uh it's it's launching these clusters uh and it's communicating to them to see how the job is going you know is it is it still alive what's it doing and so on and these are in the cloud in the customers network okay so it turned out in this region there was one customer that decided to run many drops per second on the same cluster actually it's kind of a funny story so this this customer set up essentially a lambda function and they set it up so that whenever they upload a file in a in a specific bucket they are going to send the command to our job service and on a job to process that file so that's actually awesome you know we we love that people are automating the workloads using lambda and it's easy to do that like that's great um but um you know no one had done it at quite this speed before and actually what discussed what this you know user did is um they set up that lambda function and then they uploaded you know a million files into that bucket um all at once and so you had this lambda function that was submitting lots of jobs per second many of the jobs are super short and suddenly this one instance of the service had to deal with you know very high load coming from this one customer okay so now you know the high load like you know and maybe it's okay to deal with that you can have a cure request you know there's all kinds of stuff you can do um but it turns out that actually one of the limits we hit was a little bit earlier than that um so the job service is sitting there uh trying to launch jobs um but it turns out that actually the the network the virtual network and this cloud provider had a limit of a thousand connections per virtual machine uh between different virtual networks so between our control plane and the customers vpc basically and and that's because you know it is a virtual network and there are elements in there that have little limited memory and so on and after this new limit if you try to send if you try to open a tcp connection to you know to something in a different network those packets are just silently dropped so you don't get a reset pack at the back or anything like that you send the tcp syn packets to open the connection and you never hear back you don't even hear like hey this this connection is blocked by by the network in some way um so these are going to hang so so why is that upon so so as the job service is sending commands to to kick off these jobs it opened you know many connections to you know within that network to run these things um and eventually this vm the job service vm had too many connections it reached this limit over a thousand connections going on to this customer network so now the the the problem is now normally when you open you know you try to open a tcp connection to somewhere and you can't connect um uh it will fail quickly basically um uh but in this case you know we we just sent the synth packet and we we never heard anything back so there's a pretty large timeout uh in linux until we decide to give up on the connection so now all these connections passed a thousand that were uh you know that you know that were being created on this job service for each connection there was also memory pressure because we had some information in memory we thought we're ready to start that job so we had some state about it you know which commands should be executed and so on and these were all in memory in this job service and this was causing memory pressure and garbage collection and slowing things down even more and the final thing that happened is um so okay these new jobs are not running that's you know that's not the end of the world because we didn't tell the customer that they run and you know it's expected that it would be a little slower but the interesting thing that happened is the the memory pressure and gc uh actually caused uh other jobs it caused the health checks to some other jobs uh in this instance the time out um so that means so you know as we're monitoring the jobs from other customers that should be running we occasionally ping them and we ask hey you know are you still alive what's going on and they they reply back and if we don't hear back you know within some basically like you know tens of minutes then we're going to assume that they've uh they've hung and we actually abort them and shut down that cluster because you don't want them to waste time having these vms running that are doing nothing and so in this case because the service itself was doing gc some of those health jerks timed out only just because the job service was uh doing extensive garbage collection and so we aborted some perfectly fine jobs that were running for other customers that you know were working just fine but our servers thought that they were hanging so and and so we we when the service woke up it sent them a message saying you know please shut down and they actually shut down so it's a pretty you know interesting um uh kind of situation because you you can see okay there are scaling things involved but it's not something as simple as like you know i was doing a join in a database and it became slow when i had too much data um and in fact the resources that ran out were all a little bit strange there was this this resource here uh with their connections um in the virtual network and then that caused some other resource the memory pressure to build up and you know the gc was actually freeing memory and stuff like we didn't completely run out of memory uh but it caused uh some assumptions about how long things will take to to go wrong uh and we we ended up cancelling someone's uh someone's jobs basically so that's you know it's an example of what happens um so i'll talk a lot about how to test for and prevent uh issues uh like this by design ideally um there are also some surprisingly rare issues that happened where you know when we started i would have thought these will happen more but they uh didn't um so you know um one thing uh there's only one instance i think since the company existed that a cloud provider said uh we're going to restart all your vms to patch a hypervisor this was patched to zen on aws a few years ago and i think now pretty much all of them can do this without shutting down customer vms at all so that's pretty cool we had one instance uh where there was a misreported security scan on a customer vm so customers you know run various software to look for potential problems and their software said you have a virus or you have malware or something on this databricks vm and of course we're uh going to be very concerned about that because we certainly hope that there wasn't anything on our side that that caused that to happen um turned out to be misreported so that was good but that's a scary event where you have to uh you know look at a lot of logs and and see what happened and and understand the situation um there was one really big as the outage that caused uh the thing that i would consider a significant outage where jobs are failing you know there are smaller ones that degrade performance but nothing uh only one really large one and i think there have been only kind of two linux kernel bugs that were uh big issues and actually both of them were related to tcp they led to hung connections in situations when they shouldn't have hung like the one i described before uh cool i see one question about this uh um so about from nicolas about data corruption uh we've been using git and lfs to ensure a storage system um and give high confidence against corruption because of git is this a wide use of why uses of git plus lfs i do think yeah i do think that git helps uh because it's doing uh additional hashing uh of the data and it should be easy to tell when something is wrong so that helps uh yeah we definitely use a lot of checksums and uh on just under the customer's data to deal with uh these issues and sometimes there are issues in the in the cloud storage it's not super common but again with a large data set it it it could happen eventually um yeah that's a good question yeah cool okay so yeah this hopefully this gives you a sense of the kind of things that can go on so basically one takeaway here is that um you know the load in your cloud service is going to vary on many dimensions even weird dimensions like how many uh network connections can you have to somewhere uh and also like you know how how many people are submitting very tiny jobs you know each second and so on um and also these services rely on other services with different limits and failure modes um and these problems i think will uh if you're not careful they'll just get worse as you build you know more and more services that rely on each other uh especially across vendors but even within one company because uh you know all the things you rely on can have these uh these kind of issues so for example with stuff like now you know each time we we build something we we think about you know each api that we call what if that just hangs out forever that's one of the dimensions that we'll uh test against because we don't uh we don't want to end up consuming resources because another service is being slow so these are the the kind of things that that would go wrong okay so i told you what can go wrong but how do you fix it so next thing i want to talk about is uh testing uh and then engineering to handle these things so the i'll start with the testing method that we developed this is actually something i worked on early on at databricks that like really helped to improve things it's not the only way to do uh testing but it's uh i thought it's a pretty cool way to think about it um so if you go back to first principle like let's say you want to test the system to prevent outages like the one i talked about um so testing for scalability and stability is quite different from just testing for correctness of like of your code usually if you think about correctness it's a boolean property does your software give the right output on this input or not it's it's easy to check for you know as long as you can think of all the inputs you want to test you can just check if it's right in contrast scalability and stability are both matters of degree for any software i built there will be some load where it's going to fail you know if it has limited resources so it is going to fail at some point and the question is just at what level and is that level acceptable for you um and then the other thing is okay when it stops being able to take on new load what failure behavior does it have there are some bad ones and there are some good ones a bad one maybe the worst one would be to crash everyone who's using it so far like just you know give up and shut down that would be really bad um another example would be to to drop some of the users now that is you know not uh ideal that's for example something that happened in the jobs outage i mentioned a few jobs were canceled but it's also not as bad as the first one uh and then maybe the ideal one is to stop accepting a new load and give a nice error message uh and um you know but but still continue running all the stuff from uh from other uh customers who are not uh placing very high load okay so uh just as an example of the kind of issues that can happen you know these scalability problems can happen anywhere in your software stack so um so one of the earliest things we launched was our data science you know notebooks and um you you think it's not you know there's not much that can go wrong there you just take something like jupiter you know host it in a in a container and run a spark cluster on the other end and let people send the commands too um so it seems pretty easy but actually it turns out that all these components on the path from the user to the browser to our servers uh and finally to the app that actually launches the spark commands could have problems uh so one example problem is if you have a large result set you know your spark job is collecting data but it decided to collect uh you know like hundreds of gigabytes of data into the driver program um or uh you know maybe the data got collected into spark but now you're sending it to the web browser all of these components along the way could crash in that case and of course users on our platform do want to work with large data sets they do want to send back you know as much data as possible to their browser to visualize it and they also want to maybe download the rest as a big file or something um so all these components would crash because of that another interesting dimension was a large record in a file so uh inside spark you know do we assume that each record uh fits uh for example in a byte array in java well a byte or a is only uh two billion bytes two gigabytes some some uh customers had you know individual records that were larger than that um so or do you assume that you can fit you know at least whatever like 100 records in memory at a time sometimes you can't um another interesting one is large number of tasks if you run a job on you know millions of files and it launches a task for each file then the scheduler and the the ui that shows you the progress of your job and all these things could have issues visualizing that um and finally if someone sends a command that freezes one of the workers uh how did the other components respond do they realize that it's gone and and not do stuff on there and of course there are also other users using the same cluster uh and using the same notebook and so on so uh so all these will affect other users as well okay so now i i maybe i've scared you like a lot of different things can go wrong how how in the world are you going to test for them and certainly when we started offering this product to real customers very quickly you know we would have new issues each week that we'd never seen before and it was kind of a game of whack-a-mole where you fix one and then other things uh will start breaking um and you know you just have to deal with these these kind of scalability limits all the time um so we actually designed this stress test infrastructure to handle this and it's got a few um it's got a very simple process so first of all you identify dimensions that you want to scale the system in like number of users number of output rows in a query size of each row latency of an rpc to one of your dependent services and so on so you just list them out what are all the dimensions that might go easy to do that and then we just go the load in each dimension until a failure occurs and it turns out that actually testing each dimension by itself is off is enough to to find the bottlenecks and the bad failure behaviors and in most cases you don't need to think about combinations that often because even one dimension will push on that particular load so that is one of the kind of surprising maybe insights for us but that actually makes it tractable to test you don't have to think about each combination or even about what's the right mix that exactly matches a customer workload if you just stress each one by itself you are going to find most problems um next thing we do is once you've pushed each dimension and got the thing to break record the failure type and the impact was there a good error message was there a timeout was it the wrong result uh are there clients affected and did the system recover automatically eventually or not and then finally you just compare this over time and whenever you make a change to your software so a very simple approach but it actually caught you know a huge amount of the bad behaviors and once we did this you know we went from many new issues each week to pretty much no issues about stability in uh you know in this part of the product um and so we we just ran this automatically and we had this uh you know this uh this spreadsheet it's more sophisticated now but basically uh that showed you for each test we had all these suites we had these individual tests that we're trying to to do and for each one we recorded the maximum value that is worked on and the state after and we could compare these over time and if you see in the states the failed state is actually green failed means it gave a good error message and it said i can't do this and that's actually good we want that is going to happen at some point but states like timed out for example are bad uh if if something times out that's a bad experience for the user and some of these flags here also showed the ones with red flags mean that it also affected other users on the cluster or even if you know it broke the web application in front of it or something like that so that's um so that's what it looks like so easy and uh it's quite straightforward to test the class versions using this approach um the next thing i'll talk about briefly is designing control planes um so there are a lot of talks about this um i'm not gonna say too much but um uh you know obviously like uh even within our servers we have a lot of uh smaller services that interact with each other and there's a lot of debate i think there's a panel at this conference about the right programming model the scale of the services how they should talk and so on um so we built an internal service framework that handles a lot of these and i just wanted to show some of the technologies we're using on top of the service framework we also use databricks itself extensively for monitoring and data analytics on what's happening but basically using the stuff in here and in particular using a lot of scala and kubernetes we can do cross-cloud deployments special environments like golf cloud and we also manage storage in a consistent way everywhere security tokens roles monitoring api limits and routing and very importantly feature flagging so we can all out changes to just a few customers and hold them back if we uh if we need to um so just some best practices we learned just kind of quickly so the first one so as we build a new service you know what do we think about the first one is to really isolate the state and make sure you have a great handle on where that is because that's the thing that you can't mess up and actually for us we found that just a relational database was usually enough for our control plane state with maybe some sharding for organization and it kind of makes sense because you know we we don't have like big metadata basically about um you know like how many notebooks you have and stuff like that so for for our case it makes sense of course if you're building something with billions of users you you have to do something different there um second interesting one is to isolate components that scale differently i'll show a picture of this but one issue with the jobs outage i mentioned before is that there was a single um service that was responsible for monitoring the already running jobs and for um you know scheduling and launching new jobs and when there was high load on that second part the scheduling and launching then the first part suffered so one way to prevent that structurally is to actually have different services that do these different parts and then even if you stop accepting jobs for a while you'll still be able to monitor existing ones manage changes through feature flags we tried a lot of ways and there's no such thing as a change that is always safe and never needs to be rolled back uh you want uh a feature flag for everything so that's that's what we're doing and you you may have to turn that off uh in the in the future um uh watch key metrics so uh another interesting thing we found is most outages could be predicted using just very simple machine level metrics like cpu load memory load thread pull exhaustion and load cpu load on your database so you don't necessarily need super complicated metrics like number of jobs per second or like average amount of tasks in a job or stuff like that that are tailored to your domain you actually can get pretty far with these and so that's that's kind of the first line of defense and in many services to to catch issues early on and hold them back uh and then finally test pyramid uh very standard um software engineering stuff uh even for distributed systems you want a lot of uh quick to run tests like unit tests so i just saw a picture of um you know the the the second one the isolating thing so i talked about it in the job service another place we've done it is in the cluster manager so this cluster manager that um that talks to the cloud vm api and launches clusters and each customer was also often um you know a source of failures and in particular one interesting reason was because these calls to the cloud api could sometimes be very slow and but also you know it could be that doing stuff on each cluster was slow and so the the you know second generation of this that we launched separates out the control logic things like usage and billing and so on that it has to do from the talking to the cloud provider and talking to the vms and that stuff is now sharded across these delegates and so this stuff your usage billing or ability to launch new clusters never goes down any failures are isolated in one of these delegates okay and then the final thing i wanted to just briefly talk about is evolving big data systems themselves for the cloud so it turns out i mean how you design your application has a huge impact on how easy it is to operate and to create a multi-tenant version of it and and run 7000 instances of it and so on so it turns out that actually big data systems themselves including apache spark were mostly designed when the world was on premise and uh you need to do a bunch of work to like really make them on while in the cloud and really also benefit from the underlying platform for example from elasticity and so on uh so we've done this in a few places i'll talk about the delta lake project as an example um which is a a way to manage storage uh that replaces things like apache hive but is quite a bit easier to operate at scale and but we also did a lot of stuff in cloudifying apache spark itself like making the scheduler know about elastic scaling and implement auto scaling and so on so motivation for delta lake so everyone wants to store data in cloud object stores because they're the largest and uh most highly available and often most cost effective storage systems out there so that makes sense but the open source big data stack was designed for a team that would install and operate you know one instance of it in one data center and it's got a few issues a few aspects that are different it's got this richer api file system because it assumes that you can implement consistency within that one data center um it's it assumes that you're running a a relational database for metadata so like a hive metastore because of course if you're going to operate hdfs certainly you have someone on your team who knows how to operate you know my sequel or something um and also it assumes you're going to deploy other distributed systems like zookeeper so when you're looking at offering this in the cloud you know these post problems you don't have the strong consistency there and also we don't want to run you know many instances of zookeeper or five meta story if we can help it uh and keep those available because um uh you know these are you know much harder to operate than something like sd and sc you just put blobs in and then there and that's it um so it's it's you know lots of management burden if you want to run the old stack in the cloud um so a big question is how can big data systems fully leverage cloud object stores hopefully in a simpler way than than they did before um so it turns out that the hardest problem to solve in there is actually doing atomic updates so it's the consistency problem um without having to rely on these third-party things like zookeeper that you have to operate um so when you run spark on hdfs so just as an example of an atomic update let's imagine you you just have a job that's writing a table it's easy to make the table appear atomically on hdfs because you just write stuff into a temporary directory and then you rename it and you've got you've got all the data appear at once in the cloud if you're using something like s3 there are no multi-object operations that are atomic so you can't rename lots of files at once or make them all appear it's a key value store and you can only do stuff on each key value pair so you could write a bunch of output parts as different uh objects in there that's fine um but but you can't rename them um after that certainly not all at once and then there are hacks like for example adding a special file that says when your output is done or something but a lot of problems can happen like for example if your job fails part way through you might be missing a file and also because of eventual consistency some clients might not see some of the files even though they see the dots so it's uh so it's actually pretty hard to just naively use these things and things like appending to a table or editing some of the files uh atomically uh are even harder to do so in delta lake we had this very simple idea we're going to track uh the metadata that says which objects are part of a data set so the key value store can't you know it it's eventually consistent but we'll have a different story that tells us exactly which files are in it and are not um and that we can update atomically and then this metadata that we're tracking we're actually going to store it in the cloud object store itself as a riderhead log and we're going to even compress it using park parquet to make it very fast to query the metadata for for uh huge tables so that's how it works so we have your job runs it creates output partitions and then there's a special location called the delta log and we have a commit protocol that for each cloud even though we can only do operations on one object at once you can have a reliable right ahead log with serializable isolation between the commits and then once you do this you can uh immediately you can easily work with your data you can do all the stuff that zookeeper and the hive metastar were doing before and it only relies on s3 so it's as highly available as st and there's nothing extra to operate and to get paged about in the middle of the night unless sv goes down um and it's actually even faster than hive so it made a huge impact on customers basically before this about half our support tickets were about you know consistency issues and errors with cloud storage and after that you had fewer issues and increased performance so very cool to to get that um this delta lake is now an open source project and it also because it provides transactions it lets us build quite a bit more so for example things like time travel or caching on top or indexers that are consistent with your data and or you know our background optimization of the data and it's also changing the way that customers are building data architectures because now they have something in the cloud that looks like uh you know like a database with acid properties and it can be used uh you know you don't have to to kind of step very carefully around your cloud data you can use that uh you know directly as uh for that purpose um so um uh i'll skip some of these things but i did want to mention that just today we actually announced this sql analytics product that is a native sql engine specifically for delta lake and uh provides square performance that's very competitive with cloud data warehouses and provides a sql interface including access control with grant statements and uh access from b icons and stuff like that so it's it's really uh basically it enables you to do all this all these workloads directly against the cloud storage um so that's pretty much what i wanted to talk about uh so you know in in some way the cloud is uh is definitely uh providing better software products but building reliable and large-scale products there is hard and you really have to rethink how you do engineering and there are a lot of kind of interesting ideas from new testing methods to designs like delta lake that actually minimize the amount of maintenance you have to do by design so you rely on one super available service instead of having to you know to manage many instances of it and uh if you're interested in these problems uh we are also hiring engineers especially on the cloud side of the house uh we'd love to talk to you if you want to do that and you can also learn more about some of the stuff we're doing for users at our conference next week so that's it thanks and i'm going to head into the qna zoom to answer questions there [Music] you