SF Scala: Exabytes Delivered Each Day -- Some Lessons Building Cloud Software at Databricks
hey welcome tea hey thanks a lot Alexia and as Alexi said I can see all the questions in the chat while I'm presenting so I'll be able to see and answer them let me just make sure I pop that open here yep so yeah I see someone asked about the website for reactive foundation I guess Alexi can send that out okay so you can see the screen right all good okay great yes so yes this is a talk I've given a few times before but mostly at universities and at academic conferences so it's about building you know large scale cloud software data books and what we've been doing it there and I think there are some some pretty interesting aspects here for maybe any software engineer that has to deal with really large scale systems and you know maybe also people working on enterprise products for for other companies to consume and so on so I hope you find it interesting and a database all of our controls plane services are written in scala obviously apache spark itself which is what we use to run a lot of our workloads is written in scala too and we have you know we have a bunch of lessons on that and you know we actually have a few other talks we've given on the scholar specific things we've done as well so I'll talk a little bit about how we built our software stack so let me yeah let me just start with the setup and what why it's challenging to build cloud software especially when owning these large-scale workloads and some of the kind of issues we are into and then I'll talk about how we tackle them so binary this talk is has contributions from a whole engineering team really this is the majority of the work we do is making sure that these applications can keep earning reliably and securely and these are six people who contributed you know especially significantly to the work here so these are people who are in some of the some of the software teams managed our platform aleeah's our CEO and was also our head of engineering early on and came up with a lot of the designs that you know we're still using for various things okay so let's start about why cloud software to begin with which is kind of this interesting trend in the industry and then I'll talk about what we're doing data books and some of the challenges and solutions we've run into also some questions for things that I think we don't nobody really knows how to do well yet okay so so why why the cloud so if you look at traditional ways of delivering software to customers the traditional way has been packaged software where you know your development team creates these releases of this after you know like Oracle 11 or whatever Windows XP or whatever it is and then you know II to release takes some time to make you build it you test that you package it up it might take six to twelve months and then you ship these to the customers and then the customers have to install it so they have maybe different installations and each customer has to have some some users but also some operations people who manage that installation so this also adds to the latency of getting the software deployed because you know if you're a customer you may not want to immediately set this thing up it takes time to migrate to it so it can also take six to twelve months for them to do it now the difference with cloud software is that of course the vendor takes takes on the ops all as well so they develop and operate it for you and that means you can push new versions out quite quickly we do every two weeks for most things and data books and other things are even faster they can be like every few days if needed and so they manage that for you it's a it's more work for the vendor but it also means that the users have to do less stuff in particular you can have more users and few more operations people and feet less overhead because a lot of the work to keep a turnin is done by the vendor you still need some ops of course there to configure every and make sure that you steer policies and so on okay so why would people want to use cloud softer so actually in most day as a verified software this is really the most popular way to buy stuff now and basically if you're creating any kind of product that's not cloud software your main competitor is usually some cloud thing and your product can better be a lot better to beat out those those self-managing features of the cloud product so there's a few reasons so first one is that the management is built-in which is just more valuable than just a bunch of bits you know if I give you a CD with like even the very best database software in the world your experience with that might still be kind of crappy because you don't know how to operate it super well and you have to do a whole lot of work to do whereas here that that operation you know aspect of it is bundled into the product another important advantage is elasticity you can pay-as-you-go you don't have to buy like a huge chunk of it upfront and then not use it so that's kind of nice and then finally it's the fast release cycle I just imagine I told you that you could get you know a product from two years into the future today that could be pretty nice and that's what happens because of this faster release cycle essentially but there are also some differences in building it and some significant challenges so pretty much you know the main benefit of building stuff this way is the fast release cycle and fast feedback it's also kind of nice that you'd have to maintain fewer software versions you don't have people who are running these legacy you know five and ten year old versions that you still need to support but one of the main challenge is that you really you know you don't have to worry about nearly to the same extent with packaged software is how to keep upgrading without regressions and it's really it's really hard to do this because you know you're changing stuff every few weeks and pushing that out but it's critical for customers to really trust your cloud system and this can include any kind of things that might change or request including TI's semantics performance or questions all that stuff right you know when you're using manage software like let's say Gmail you never worry about you know which version of Gmail am i on today for says yesterday but it means that they better make sure that all the new versions keep working and if your software's not just a consumer thing but something with an API and something with you know performance characteristics that are important to keep scaling then you know it gets even worse the other challenges are to do with operations so building a multi-tenant service how do you make sure that you can have maybe thousands or you know maybe even millions of concurrent customers that share some infrastructure in a way that's still secure and that isolates them from each other in terms of performance that's something you don't need to do for him stuff because of your own installs their own copy and then operating the surface monitoring and so on you know it's obviously more work that you have to do although you could argue that you know someone would have to do that anyway perhaps tough and finally you do get the advantage type to the fast feedback loop of getting real-time usage data so we use that a lot to see what's happening and catch issues early and fix them and also to guide our product development ok so about data books itself so we are you know cloud platform company we are founded in 2013 by the team that originally created Apache spark at Berkeley and we we just provide this this platform on top of Amazon and Azure that you can use to set up you know data and machine learning applications in your company so that's the idea so and it's only in the public cloud we have over 5,000 customers on those now and we actually operate at really large scale so we have millions of VMs launch for a day in both in each of these public clouds they're processing exabytes of data per day we have over a hundred thousand users now and the scale of our service because all the jobs are skill is actually larger than many of the cloud providers you know built-in machine learning and big data services so it's definitely getting up there on Azure well so we've also partnered with Amazon so that they basically sell our service as an azure product so so definitely like pretty large scale and we have about two hundred and forty engineers so we actually launch and manage more than ten thousand VMs for engineer each day so definitely a lot of stuff has to be automated this is a plot that shows how kind of the scale of our platform has been growing over time so this is showing the number of VMs who are managing each day and you know you can see it's it's going up pretty quickly it's continuing to go up this year as well so you know it might look like we weren't doing anything and 2017 and before that but actually we were working hard of course to get the first few use cases at our customers and then once someone is set up with the platform they can naturally increase the workload on it you know migrate more stuff create new use cases and so on so the past ones keep going and of course new ones keep being added so we were really excited that we goo usage close to a factor of three last year actually and 2019 and it's already going you know it's already like almost doubled this year if you look at this plot so you know this is this is one of the exciting things about the elastic pay-as-you-go model is if people like the product or if their obligations that they build are successful they can very easily add more stuff on it of course the problem for us is how do you deal with this this fast growth where you know while you're trying to build your new features and so on something that you built might be starting to tip over or break in some way because the the scale of it is going so quickly so just to give you a sense of like what kind of companies use data but this is just like a marketing slide with some of the customers but the main thing is there are you know I think everyone on this call if they're from Silicon Valley is probably very with kind of ads tech and media companies you know the kind of things you'd see in Silicon Valley but also like really serious you know enterprise and kind of you know government organization type use cases that are using large volumes of data and technologies like like Apache spark today so you know just as examples a lot of biotech companies are using these things to develop new medicines or to process medical records and kind of see what's happening there's actually like some of our customers have written you know new research papers on on these things there's also really interesting use cases and finance or insecurity or an internet of things so people have these like you know all these devices out there where they're trying to collect information about how these are being used so these are just some of them yeah okay so I see a couple questions actually so one question is we're assuming that customers bring their data to the cloud what happens if they're too distributed so yeah we and they can't even upload the data to the cloud so there's definitely a potential issue we've seen that you know especially because cloud providers give you many geographic regions a lot of these use cases set up basically if you are an enterprise and like you're setting something up like suddenly you're you have smart meters for power you have like you know factories all over the world you're trying to put in sensors to do predictive maintenance the public clouds are a pretty good way to get compute infrastructure in each geographic region with the same API so that you you can have you know somewhat local data processing if you're not able to upload even to a cloud provider in the same region then you have to run some stuff on the edge and we see some people doing this we don't cover this use case as much but you know we have some examples who are like we we let people on spark cluster on the edge and and it summarizes data and upload something so it's not gonna cover everything but there are quite a few things where you know if you're trying to set up a new kind of data collection infrastructure and especially if it's worldwide one of the cloud providers is a pretty good bet to just quickly have an environment like a navvy country without having to stand up like you know 30 different data centers and whatever and there's also a question about private cloud customers so almost none basically like all our revenue is on public cloud is on Azure in AWS in some cases we have partnerships with you know or special set ups with some companies so they can also hunt things on-premise but these are a small fraction of revenue and we actually don't we try not to do a lot of them because it's a lot of cost of engineering for something that you know it's it's gonna be very unique to that organization so you know as a as a start-up we have to focus on certain things and we're focusing on what will reach the most user is especially in the future a few years from now where do we think that most of them will be okay cool so so that's a bit of like what we do so you can imagine you're like all the work to run a data pipeline in one company we have to do it for like maybe up to 5,000 companies that and maybe for some of them it's also applicated in several regions and we basically have to automate all that and in terms of what's involved in the product this is just a very high-level overview so there are user is there different interfaces to it including like data science notebooks you know scheduled jobs and just sequel for for visualization and analytics and these all around in our control plane so this is some machines that we own and then there's also what we call the data plane which does the computation and we've set this up right now so that you bring your own Amazon or account and you give us an I am hold that lets us launch cluster is in it and access some of the storage and we own the computations in your cloud accounts so it's kind of an interesting set up the banner of this is that if you are some large customer of AWS or a juror maybe you've already negotiated a discount with them maybe you have some way of managing your VMs like something that oversees all of them you know maybe you've got reserved instances and you can just use all of those for your database workloads because we are on your account of course an alternative is for a store on all the compute we haven't done that but you know for some customers that's also that's also an interesting approach because maybe they don't have the expertise to manage this so it's something that we may want to do in the future as well and within this so we just installed what we call database on time to execute the clusters and then the stuff that we actually are we try to make all of the api's that you use to write your applications open source so that you can easily find a lot of users to know how to do them and you can also hunt them in other environments that are not database and you also not locked in to us so again if you go back to the value of the cloud service there's a huge part of the value is the automatic management so it actually kind of makes sense to say look we'll we'll focus our engineering on this these automatic management things that are really hard and that provide differentiated value and then in terms of api's we'll try to to work as much with the community as possible we're not going to create like a competitor to tensorflow or something we'll just make it easy to run tensorflow on this platform and benefit from all the other stuff happening in that ecosystem so that's that's kind of been our approach in terms of the workload okay so that's a little bit about us but going back to the challenges you know it's we've got all the usual challenges of software as-a-service higher availability security and so on plus the workloads themselves being large scale which means that you can easily imagine it's happened before that someone runs like one spark job on our platform that talks to our control plane and that becomes I could denial of service attack because it's a very large job that's making API calls and we also get to see all these like weird edge cases and very well failures that happen with cloud providers with these millions of VM so anything from like you know network packets being co-opted by bit flips to like you know very issues with the Linux kernel and stuff like that we've we've gone into quite a few of those and so we you know we obviously have to make it easy to catch those and hopefully recover them recover from them automatically without even paging us or even having the customer notice that something went wrong in those in some of their VMs okay so I'll talk in terms of more detail about four more four specific things that we learned so what are the causes of errors especially the ones that are hardest to catch how to test for scalability and stability are very important to prevent these building these these control planes that have to deal with potentially very bursty workloads and then evolving big data systems for this cloud environment making them easier to manage and sort of more servers if possible yeah I also saw a question from Sam about are we looking at Federation or not if you mean across clouds and so on we're not doing it on there but we do have we try to provide api's that make it easy to deploy the same pipeline in many places similar to something like cloud formation so that helps with that aspect yeah well okay all right so I'll talk about what what goes on in these kind of systems at this large scale so again like if you if you look at say academic research on this or like read books on other wonderful you know there are there are many kinds of failures that people worry about and some of them are really well studied so things like software bugs you know how can you check them can you statically check yourself network configuration crash failures these all happen and they are well studied and you know there are solutions for them but if we look at like which issues have the largest impact interestingly in this kind of cloud and you know business-to-business like server setting we see a lot of problems that are as carefully studied and where it's not as clear what to do about them so in particular scaling and resource limits workload isolations and updates and regressions and this last one especially I think it's kinda new to the SAS and and kind of API delivery model just because people weren't delivering software that way until pretty recently you know basically until you know the last five to ten years when it became very popular okay so just to give you a sense this is a overview from I put this together second half of last year on some of the what caused significant outages for data fix by significant I mean that this is something that took down production workloads such as scheduled jobs or streaming jobs for multiple customers so that's like a large outrage and you know there are smaller things like you know this button on this dialog box wasn't working or whatever but those are usually easier to you know define and to fix but these are ones where someone has to redo work essentially to handle with them so you can see so there are quite a few different causes but the main ones have to do with with scaling so scaling problem in our services where something changed in the workload or maybe we put out a new version of the service that didn't scale as well as the previous one that's 30 percent interestingly some of them are from underlying cloud services where something there was not working I'll give a few examples insufficient user isolation this is where some user is placing high load or doing something unusual and because these are multi-tenant services it brings down others as well and this can be very hard to fix because someone might have a reasonable looking let's say API request rate but then their specific requests are causing you know lots of memory pressure or they're causing runaway queries in the database or something crazy like that and so you kind of want to isolate them throughout the stack so this is I think this is still kind of an open question how to do this super well and then we also had deployment misconfigurations or another one that I think is more common than people know about so anyway if you just look at it at a very high level about 70% of them are only related to scale these wouldn't happen if you only have like one user or one customer out of time so these are actually some of the hardest things to face so here are just some examples of the issue as we've seen so an example of like cloud provider infrastructure is cloud networks these virtual networks are not the same as real networks they often have limits on let's say how many you know pairs of VMs can connect or how many connections each VM can have two different V pcs or things like that and when they fail they can fail in very weird ways where some of your packets are going through the connection and some of them are not which is different from our physical network or like if the cable breaks you know none of the packets are going through so so that's an that's an interesting cause automated apps creating large load very large requests slow PM launches and shutdowns this is another cloud provider thing where you ask it to launch a VM and then it takes 30 minutes and if you're trying to keep state and memory and stuff like that it can easily you know explode while you're waiting for at least to come back and then data corruption writing to cloud storage so I'll just give you an example of one one outage that highlights some of these and and I'll get back to some questions because I saw them after so this is an outage where we observe that at some point you know some customers like production jobs were aborted the system decided to cancel their job and you know why did that happen alright that's kind of bad the jobs are talking along fine and we just decided to shut them down so just to help you understand what's going on here we basically have this job service it's a multi-tenant service that looks at the schedule of what we need to 1 and then it launches jobs on these customer clusters and because these are and the customers resources you have to go through the cloud network to talk to their virtual private cloud or whatever it is that they're on again ok so what happened is that one customer submitting lots of jobs per second on the same cluster interestingly this customer had set up basically a lambda function so that each time they upload a file into a bucket they on a job on X to parse it and you know do something with it as a spark job and that's like really cool right that's exactly what you want with service computing it's awesome you put in your file and immediately like some work happens downstream it's great but then they also put in like a million files into this bucket at the same time so now there were these kind of millions of API calls that were queuing jobs and we absorbed all the jobs and we were happily you know trying to launch them on this person's cluster so that's what was going on in there so interestingly though in this particular cloud there was a limit of a thousand connections per VM between any VMs in in different virtual networks basically so so there was this limit on how many connections we could have between our servers and the cluster where we were running and monitoring all these jobs and these were tiny jobs so we were trying to launch a lot of them at the same time and once you hit this limit unlike a real Network this virtual network decided to just drop packets but for any new connection that it sees like any new you know pair of ports that you're connecting between and it didn't even send back a reset packet to tell you your connection was shut down and just it just dropped all the packets so from the Linux kernels perspective these things were connections where it sent the syn packet and it never heard back so it was waiting a few minutes to hear back from that syn packet before you know deciding that the connection has died okay so this so that means that at least and here's some threads or some some connections where you know we're waiting to hear back from these same paths that they sent so now you know each of these connections where you're launching for a job we assumed we assumed that that would be pretty fast so we had some state in memory like for example the details of what to execute in the job and so on that we that we'd keep in memory for a very short time because we're just going to send that to them once we connect and then you know and then like delete that stuff but now because these connections were hanging for like you know minutes or like tens of minutes each we were starting to accumulate a lot of this stuff in memory and this cause memory pressure and garbage collection in this in this job service it wasn't out of memory but things were just getting slower and slower and because of that some of the health checks we had two existing jobs where we actually had an active connection we were able to send packets back and forth timed out so we said you know we try to contact this job and we couldn't hear back for like five minutes even though actually we were just doing garbage collection for five minutes and then when this service woke up again it said oops that thing timed out let me abort it and of course that packet got through to the existing cluster and aborted it so I think it's a it's an interesting example of user isolation you know resources and some of the issues that can happen here because you know if you think about it it does have to do with many users being together but the resources that got exhausted where it is like pretty weird resources first of all the number of connections per VM something you know we had until you are in two before but that happened in this one weird case out of thousands of customers this one you know weird group decided to do this thing and so that was kind of a strange resource and then there was also this you know memory pressure were like we didn't crash actually we had everything we need to keep existing jobs running and monitor them but because of this GC pressure and the JVM queue we decided that a few of them timed out and while you're designing a system you know it's not as simple as having great limiting and things like that to handle all of these kinds of issues so are some questions what's the average heap size on the job service and our costs two nodes what's the default GC I I don't know exactly but I think the heap size on this was was definitely like in the tens of gigabytes and we chose one of the GCS that minimizes late and but when you've got large objects that are sitting in there for a long time you'll still have you know even these GCS will slow down so so eventually like I mean the other thing that could have happened is that this thing with crash which I guess would have been a better outcome but on the way there most things do slow down for a while we don't have anything to shut you down if there's a lot of GC or anything like that there's also a question about priority customers can you do you like isolate medium customers do you give them SLA you can do that to some extent and we actually have some of that we have some ways that you can deploy just by yourself but there are also disadvantages because if you isolate all of them now you have to manage you know like hundreds of copies of the control plane and it's actually more likely that something goes wrong in one of them and it's harder to catch with monitoring so in many cases we think it's better to have a single like highly replicated highly available control plane as opposed to something for customer but of course you know you do have a long tail and some customers are placing large loads so if you can do something special you know for some of them it makes sense yeah would the resource limits be solved by having a separate job scheduler pertinent account definitely would it's just there are so many tenants and as I start managing like thousands of copies of the control plane is not easy either but it's it's an interesting question right maybe if someone can make that easy it would be could be great to do stuff that way we actually have other solutions to this that I'll talk about that are more about reducing the blast radius of failures like this even when you have a single instance of the service and I'll talk about how that helps but but that kind of helps you have them the management you know the ease of management of a single instance of the service without these like weird things bringing everything down but there are other ways to do this too yeah okay cool so that's that's just an example of you know these kind of issues there were also some things that happen we're very rare so for example across like you know six or seven years of operating the service we only had one time when a cloud provider said we have to restart all your VMs which was when Amazon had this I think now they can do it without shutting down VMs they can just upgrade the hypervisor you know we had one instance where someone had a security scan software that told them you've got malware on these VMs that are running database and then they thought shoot is there like a problem with database did it get malware it on my VMS turns out it was a false alert for them but you know that's a pretty scary thing to have to debug it's you know if you think maybe someone you know hacked your customer through through your service that's not good we had one biggest fear did that created these these big enough issues to shut down jobs and we had two coronal bugs that were actually very related to the kind of issues before they both resulted in hung connections and then they broke a bunch of assumptions or like you know we for example we could check if a node is up by talking to it on one TCP connection and then we sent it's tough to do on a different connection and the second one broke and then we were you know confused about why I never did the stuff that we asked it to do even though it it seemed to be alive so these are these are the ones that actually caused problems for us okay so basically the takeaway is that especially when you've got something with an API and something that runs programs and when those programs are trying to do large-scale workloads your Lord has got a value on a lot of interesting dimensions and especially if you're going to rely on other services you know there are more things that can break including with internal and external services so it's it's really important to think about how these will happen how you're going to react like some something underneath you you know not like hitting it's it's SLA limit and so on okay so this leads me to the second part which is testing for scalability and stability how you know now that we've seen some of these issues how can we prevent them they're obviously a lot of other things you can do to prevent as well such as monitoring but certainly it's nice to be able to test for them so the way we've thought about these and I experienced with these is that when you test for scalability and stability it's actually quite different from just testing for software correctness so jessica recognises usually a boolean property you hold a function you know if you give it a certain input does it give the right output or not so that's that's what it means on the other hand these issues scalability and stability are a matter of degree every system with limited resources is gonna run out of resources at some point and the question there's really like kind of two questions what is the point where it runs out of stuff what is that level of load and then what is the failure of behavior is graceful behavior like you know K like dropping some of the new requests or rate limiting people or stuff like that and then there's ungraceful behavior like you know bringing down the jobs that were already running on the platform or crashing everyone and so when your mom when you're testing and monitoring and you know doing you know upgrades for a system that needs to worry about scalability and stability these are the two things that you want to measure so it helps to explicitly measure these two okay so just some examples you know to give you another example of scalability problems this is just for a basic like this is our very first version of our product which was just interactive notebooks doctor Han you know that connected to an instance of spark and then Han and some co-workers even in this example there were a lot of different dimensions of the workload and of the system that could cause things to fail so for example you know one of the first things people try to do is visualize large results like large query results it day in their data and if you're not careful with this the large results could crash spark itself they could crash this driver app that phase there is also back to us they could crash the service that's not serving the web we even had things that pass through all this but then crash the users browser and so you know these are examples of some some you know scaling issues in each component same thing if you have a you know a file which has a large record or millions of columns in it you know large numbers of deaths in your spark job you're trying to display them in the UI and it can't display them any kind of code that freezes a worker so actually in the fourth version of database we kept getting you know like every day we as we added users we kept getting these kind of issues and we also you know once we fixed an issue it would come back a few weeks later this was before we built our scale test infrastructure for these kind of issues so it seemed very daunting at the time to even get this basic thing to run to run really well as people were throwing more and more weird stuff on it and to make things worse all of these are happening on shared clusters and of course the shared instance of the service so these can also affect other users okay so how did we handle this so we actually designed this fastest infrastructure that's based around those two questions I mentioned at the beginning this is something I was involved in designing actually back in 2014 so it was pretty fun to work on and it's actually very simple but very effective so it does kind of four steps so first of all you identify the dimensions that you want the system to scale for example number of users number of output those in your quiz I saw vitro or for things like talking to the the cloud provider to ask it to set up VMs the latency of that service all right what if it goes from taking 30 seconds to taking 30 minutes so you just identify all these dimensions you can list them out you can even add more later as you go along and discover things so it's pretty easy for a team to just list out all the things where the you know where the workload might go then in our framework we just go load in each dimension independently until a failure occurs we would take we would create a simple workload that you can parametrize and then just take one dimension at a time and increase the load you know by a factor of 10 each time so like you know one user ten users a hundred users or like one column in your table ten columns a hundred columns and so on so just go it exponentially and we found that one dimension at a time actually was sufficient to catch pretty much all the issues because if you exhaust the resource it's very likely that you can exhaust it with just one dimension so we didn't have to weigh so much about the combinatorial explosion of how do you check across like you know two or three dimensions at once that was an interesting insight we had fun from this work and then once you do that eventually the things going to tip over in most cases so you record the failure type and then the impact on the system so was there an error message did it just hang in timeout that you got the wrong results that's really bad that other clients get affected like if someone else was running a query on that cluster it did their thing break or could it still go through and also did the system Auto recover and how fast ideally it would auto recover and it to just tell you you know something went wrong but you know here's we're back in business and you can continue earning stuff and then we were just compare this over time and our changes so this is the kind of infrastructure we hold out and within you know like just within months of Alling this out we were able to go for a mistake or like a new thing was failing you know every week and each change you know cause new questions to a state were like all these things remain stable even though the number of users was continuing to go very quickly so really made a difference just to run this and just push the system in each dimension and record these behaviors if you want to see an example output this is like a really old example where we were tracking these results in a Google spreadsheet basically and this is comparing two runs of a test like it's literally copy and pasting some TSV files in today so you can see there are these different Suites and they each have tests like for example what if I have a big broadcast variable inspark what if I try to cache lots of objects in memory and large what if there really tiny so that places the GC pressure you know what if I try to display larger hoes like all these different things that we thought could create problems and for you trouble them we track the max value that the things scale to so you know we were able to vodka stuff to whatever this is like 1 billion bytes or whatever it is and then we track the state what happened to the system so failed here is actually good it means that it gave you an error message and said I can't do that and you know nothing else that's awesome timed out is kind of bad because it just hung we would prefer to avoid that succeded is in the rare cases where it actually can handle that because this cache for example will just not cache stuff once you fill up the memory so that's fine and you know there were a few other states as well and then these flags here are other things that happen like for example the CB means that the whole cluster was broken and now no one can run commands on it so that's something you want to avoid whereas other things like this SB is just that your shell-like your session your notebook session got broken but other notebooks are working fine so that's okay and then this is like the current value this is the old value and then you can see you know some things improved some things got worse between these two tests so you can quickly highlight okay you know this thing is scaling to one step less this thing actually got better and so on so very simple framework but again the really cool thing is how it went from kind of this game of whack-a-mole with like weird things that would break every week to suddenly everything's kind of working okay so pretty cool outcome of what happened there and the other thing I'll mention about this is you know this is an example of an integration test that actually ran everything but we also have in many cases were able to reproduce these things as unit so it's even faster the one this is this was actually a little bit too heavyweight so we moved a lot of stuff into unit tests over time for these kind of issues cool so hopefully that gives you a you know interesting example of how we do that umm I'll also briefly talk about developing control planes as we as we hold out these new services and a little bit about revolving big data systems okay so developing control plane so of course I think everyone knows this we've got lots of interacting independently updated services so how do we actually make these easy to build so like many other companies we have a team that builds a service framework and that everyone else can can build on and that can handle a lot of these issues so so our server stack handles deploying the same system into into multiple cloud such as AWS and adder and also into you know potentially special in environments like customers to have restricted internet access to their VPC and stuff like that so we want to abstract that away from my engineers it handles storage databases team updates security tokens and holes and managing credentials so these things have the minimal access to other services and the systems monitoring routing and feature flagging and we use a bunch of things we use scholar very heavily we actually use basil as our build system for that and I think we've contributed a bunch of things to make basil work better with Scala and we also use envoy prometheus JSON it and and of course we use data breaks itself to monitor how the product is doing in many cases and actually to implement some of our features where we give people analytics on their workload what are some of the best practices we learned so these are just things that worked for us so in terms of state we actually have we don't have that much state really given the size of the workload so we we actually use my sequel basically for all the state and we have four customer sharding and it's usually pretty good so that's kept things really simple so have the new service we begin by designing the schema for what it's gonna store and you know that actually works pretty well for that what a more interesting thing is isolating components that skill differently so I'll give an example for that but in that job so dish and the part of the issue is that we had this one service that was meant to both launch new jobs and do monitoring and then when there was an issue with someone launching jobs it messed up you know the part of the system that was doing monitoring and that course thinks the crash so one of the lessons there is just isolate those and maybe let them scale independently or at least like reduce the blast radius if something's going on with launching jobs you know don't mess up to think that monitor is existing jobs which are happily talking away on on the data managers through feature flags we tried a lot of ways every change everything you put out you have to be able to turn it off and all that back ones once it comes out it doesn't matter how trivial it is and so we use feature flex for everything and you know any other way that which I've had had problems watching key metrics so this is a pretty interesting one a lot of people and they talk about metrics they think about very custom metrics like you know number of VMs launched per second or like whatever like command sent to the cost or stuff like that but actually most outages when something was crashing due to load we could have detected that from these very basic system metrics which are basically CPU load memory load database CPU load and thread pool exhaustion so these will usually catch most things and in fact actually if you just watch these in like less in in like smaller regions basically with fewer customers you can often catch a lot of regressions in advance and not on into them when you deploy to the biggest region so that was an interesting learning that you know you don't have to get every team to like log you know hundreds of metrics to get progress here you can get pretty far with just ease and then the other one is test pyramids so have most of the stuff in unit tests including things to do a scale testing turns out a lot of them can become a unit test because if you don't do that you just slow down the dev cycle because you have to run integration tests every time so these are some of the things we we ran into and they helped improve the system quite a bit well just to give an example of this the second point with isolating components this is similar to the job service this is a system called the cluster manager which actually launches and monitors the clusters that we're gonna execute SPARC on in each account so this started out as a monolithic service where it served API calls like you know please launch a cluster for me and it talks to the cloud provider and an asset to set up VMs and then it also talked to the VMS to see what they're doing so this had similar outages to what I mentioned with the jobs on where for example when this thing slows down or if there's an issue in the network talking to these this whole thing the whole service would run into issue is memory pressure you know whatever like so that pool exhaust you know all kinds of issues as the load increases and so the idea here was to separate it into a master that handles the really important kind of stateful stuff about you know which clusters am I supposed to be running and how am i billing people you know how we are counting for how long they executed and all that stuff and this thing gets relatively few requests or operations per second and then we have these delegates that are basically stateless workers that do the VM launch setup and monitoring and they talk to this and now if one of these is facing problems you know most of your clusters keep working and you know basically the blast radius of of that thing going wrong is it's quite a bit smaller so this is an example of of this second point and it took a while to migrated but it's you know it definitely improved reliability quite a bit okay let me answer a couple of questions about this how do you guard against the service framework or API gateway becoming a New Age monolith you know I think there is a risk of that so far we haven't run into it and it might just be because you know we're pretty small overall so we don't have a team there that's just like you know look that has nothing to do and it's trying to add lots of things into different work they actually always have like pretty high priority things that they're doing usually with one of the users so it's um so so so basically we don't see a lot of like extra stuff going and that that they will maintain and then the individual service teams are we design our organization so that they kind of own their service and - and they don't have like separate front-end and back-end for example so they're pretty focused are like anything they do should should actually help their users and they'll often kind of take matters into their own hands if something's not available in their service framework they'll build it themselves and then later they'll talk to that team about how can we share this across the company so so it's been okay there's definitely a move like as we move we used to have some services like for example the thing that serves the web application that were a bit of a monolith and we've been factoring those out but that's usually been the biggest problem another question have you considered something like shuffle shining for workload isolation I think this is about the SPARC data plane like how it does shuffles if I if I understand correctly and in there we there isn't there aren't as bad actually as the ones across tenants but there are definitely users saying a short a spark cluster and we have tried to isolate some aspects of that okay it looks like there's a library to do this okay I'll take a look at that I don't think I've seen that before but it does sound yeah it should be pretty interesting to look at I don't know if I can easily see what this is without messing up yeah I'll look at it after they do after the dark well thanks for the point I do not don't okay great cool so next all right so so final thing I wanted to just talk about briefly is how do you involve big data systems themselves to run in the cloud environment so as I said this environments are pretty interesting because you have this massive scale and elasticity but a lot of the big data systems including spark really were primarily designed for an on-premise setup so I think there's quite a bit you can do to really make them embrace this environment and that's some of the stuff that we're working on so if you look at the current you know big data stack it was really designed for these on-prem data centers where you have a team that sets up the data so knowmaybe sets up lots of other things like zookeeper and my sequel and then you get to run against that but in the cloud you want to get to somewhere where anyone can provision them they don't have to set up like three or four auxiliary services and so on and where these things are actually elastic so I'll just give a couple of examples of things we've done Delta Lake which is an open source way to manage tables and have transactions on data Lake storage and then some of the things we're doing with running Apache spark in a more serverless a kind of cloud native fashion so Delta Lake motivation here is you know we saw tons of users as soon as you're using the cloud you want to use the object stores like s3 and Azure blob storage and so on because they're just kind of the lowest cost and most highly available systems for that kind of storage out there they also have massive parallel i/o bandwidth so if you are going to store a large amount of data it makes a lot of sense to just put it in cloud storage instead of standing up let's say a separate data warehouse or something like that and having to manage that and things like Hadoop and spark could work against them but not very well and to get really good behavior on these like for example to get transactions or to be able to go back in time and edit your data and your table the wild jobs are running against it and so on you would have to use something like hive acid and that was designed for this on-premise world way you're also running a database and zookeeper and stuff like that so that was a problem for you know for getting this to work well in the cloud so these aren't promised things because they're smaller scale to have stronger consistency and they also you know they assume you have a team to manage them so the question was can we good how can we fully leverage these cloud object stories and give you very easy to manage storage systems without asking you to be an expert and adding these things and without having to run all these things ourselves which can introduce more failures into the systems so in just a very simple example let's imagine you're trying to write some output into a table or into a data set using spark let's say you're doing the source HDFS so one of the problems is you know you're running a distributed job here so as you're running this job you know different tasks are going to write different files for your output and some of the tasks might fail some of them might be slow so you launch a backup task that's also going to try to write an output so for you for your own kind of sanity as you have other applications consuming this you really want all the outputs to appear atomically when the job is done and you're sure that everything has succeeded you don't want to see partial output as the job is going and then have clients start to read some of that and then have it all go away cuz this job crashed okay so how do you do that with spark and HDFS it's actually pretty easy basically your job creates a temp directory it writes all the partitions in it and then HDFS is is strongly consistent and has this atomic rename operation where you can rename throughout this directory to the final path so clients that are looking at this final path will only see it when you're actually done hiding it so kind of nice very simple you can also use this to append to a table atomically and stuff like that now think about doing this on epstein so the naive way to do it is you launch your job and you tell it to write you know each task is going to write a different blob in SD but the problem is that s3 is eventually consistent and it doesn't have atomic renames the key it's a key value store so if you--if you want to change the key of a particular object you basically have to rewrite it and copy it somewhere else and that happens one by one there's no operation that lets you act on multiple objects at once because it's like you know globally scalable like geo distributed key-value store so you could try just having them write to these there's no cheap rename and maybe you could try a hack like for example you tell clients you know don't ignore everything in here on this day senior file that says that it's done and then maybe that's going to tell them that okay now it's safe to read all the stuff in this directory but even that approach goes as follows for example you know if one of the tasks fails and it hasn't written this you you might you know basically you need to get another test to I to the same object and if in fact it didn't fail it's actually just in the background trying to hide it then they'll clash about who hides it there's also a problem with eventual consistency for example someone might see they've done object but they might not see one of these and then they'll think there are only three files in the directory and there were actually four so it becomes pretty messy to manage and real child like real operations are even harder like if you go back and you try to insert data into multiple partitions of a hive table atomically that's like writing into multiple directories at the same time okay so what did we do about this so we wanted a super simple easy to operate approach and a key idea was to rely only on the key value store and to basically use it to implement the transaction log so we're gonna track we're going to track basically a big list that says which objects are part of a table at a given time like which file names I'd like part 1 part 2 and part 4 whatever and then we're going to store this in the cloud object store itself as a writer head log and we're actually going to compress the log periodically into park' so now it becomes very fast to search across your entire tables - you don't have to do these super slow list operations to list millions of objects you can just square your park a file and quickly see you know which files are in there may be a matter specific column or whatever so and we figured out because it's a writer head log we figured out a way to do it is atomically on both espionage err so that we don't need to hang zookeeper or my sequel or anything else so it's made basically the only thing that has to be atomic is appending to the log and everything else you know we'll use the log as the source of juice to deal with the eventual consistency so the way this works is simply when you run your job you just create these out with objects with different or random names doesn't matter what their name is because clients are gonna ignore them until they incorporate it in the log and then you pass it through this different committee API for each service that opens up into this Delta log and there's a way to do this atomically on each of the popular you know cloud object stories and then actually coining this turned out to be about 10 times faster even than querying the hive matter store in a lot of cases so we actually gained because now we can query the log in parallel just by you know reading this massive park a file and SV it's actually a lot faster than playing hive or my sequel or anything like that so this is an example of a kind of a cloud native storage system we take the key value store as a given this thing is massively scalable it's like more highly available than you know anything you could really try to hire yourself and so on and we figure out how to make it responsible for all the hard pieces so that we get transactional storage system that's as highly available as the key value store itself just to give you a sense of the impact you know before this about half the support tickets you are seeing we're about cloud storage issues like my data set is corrupted because my job crashed or the output is wrong because of eventual consistency or whatever and once we rolled this out all of these went away so it's like it's a you know pretty significant impact of just this one like you know concept really simple idea of relying more on the object story to do the difficult storage things and we've also used this to build quite a few other features once you have this log you basically can build a data warehouse directly on these objects so you can support things like search and deletion caching indexing where you maintain an index transactionally with the data and you update the index whenever you add data in a safe manner audit logging and stuff like that so you can read more about this on the Delta Lake website and it means that for a lot of users they're able to you know they don't have to worry about how to stream data between different systems they can just get this familiar interface of transactions but using the same low cost as the cloud object store I'm going to skip some of this but there are some really cool use cases just at the spark a I summit for example we had you know one of the first largest cases of this is for network security and intrusion detection our Apple so they're basically like hundreds of terabytes of logs from you know commands happening on all of Apple servers you know network connections you know employees logging into VP and all that kind of stuff and they have a team of hundreds of people that write rules to analyze these logs and to detect and diagnose you know potential security breaches or go back in time and explore petabytes of data to see like if an attacker broke into Apple what systems and they touch so this is just one example use case of like where this can be used yeah so that's one example there's quite a bit to do one the current execution and a computation plane as well that were gradually doing things like auto scaling of spark lustres and storage volumes or kind of a server less user isolated experience for spark clusters themselves so these are some pretty tricky problems but if you can provide this it obviously makes it easier for people to use cool and then that's pretty much what I wanted to cover I want to make sure I have some time for questions still the main takeaway here is I think if you look at it you know business to business software cloud is is really here to stay and it just enables better products because they're self managing but building them is pretty challenging and you know we hope to give you some light on how we're doing that and we think there's also opportunities to do new things here so I love to see you know if people have new lessons or new service framework so anything else that that makes it easier to build these cool and yeah I'm happy to answer a few questions yeah let's please ask questions now guys is the usual and logical goes oh yeah maybe I'll stop the screen show just if people can see each other in the cool picture panel view I see yeah yeah I see one question if I'm Alexi was about delta engine so yeah so Delta engine is an example of how we're trying to create a more kind of server less high concurrency environment for you know a big data computation so it's it's it's a new project we just launched it at spark a I summit it's basically a native execution engine for spark sequel that can on spark sequel and data frame workloads and we're rolling it out it's currently in private beta with users but it's it by passes some of the challenges with doing this in the JVM with isolating users and also with getting really high performance on modern CPUs so we'll see how that goes but that's that's like one thing we've done as we see these high concurrency workloads we think that underneath the spark sequel API you can't implement you know there's different engine design that that can do them better oh thank you yeah all the questions I also see does client implementation language play all and implementing a large-scale system so we found working on the JVM was was awesome because there are so many development and debugging tools for the JVM ecosystem and of course we use Scala for that so I think I do think like writing services and Python or Ruby would be quite a bit harder and you'd find fewer people who know how to operate those and who know how to squeeze out performance so and you know I think there are definitely people who had success with go as well but we just had this JVM background and it's such a mature ecosystem that you know it just kind of made sense to use it yeah I wish I don't know it's really tight now so when you use color do you use anything like z/os ecology cat or something some kind of like if you library and if not why we haven't been using those we do have some custom libraries that we have built for specific things that we have to deal with like accessing the database or handling feature flags or even users logging logging metrics like you can for example for feature flags it's easy to create a context where you know you check which flags are active and do something in it I think really we just we don't have that much to do that involves like heavy duty processing of collections so things like scholars you or maybe not as useful for that but you know it we'll take a look as these things come up you know we might we might use them yeah I'm trying to think so so so Lee how he works on our like developer tools and infrastructure team and and he's built a few things they're using like interesting features of scholar but I think like some of the most you know impactful ones were things like JSON it and a configuration tools and stuff like that at least so far these were the biggest pain points let's see okay comments on ml workloads and streaming workloads so yeah four ml workloads we we do see people doing checkpointing rollback and distributed deep learning we have quite a bit of usage of tensor flow and and pi torch and and scikit-learn actually on our product and we actually have like these integrations in the UI and in their api is for those does yeah so i would say i don't think it's like that different in some ways from data engineering in the sense that if you have a production application you want it to keep working without bothering you so so things like fault tolerance i can can you recover from a failed node and can you load from a check point and so on are equally important in both of them the the most unique thing there is I think is deciding when to use GPUs and figuring out how to use them efficiently so there's some work that we've done in an Apache spark and also in our platform to make sure that it's easy to launch GPUs and stream data into them from you know from data sources and spark yeah of a streaming in real time and Apache beam so you can definitely run Apache beam because it just compiles down this part we have a lot of streaming jobs we also have people who are you streaming with this basically they have they trigger at a job to only run for a few hours each night but they rely on the streaming engine to maintain state so we have this like on now mode for spawn structured streaming that allows you to just on it one batch at a time so I think when I last looked at it we had something like a hundred thousand you know concurrent streaming jobs on the platform many of them are small some of them are very large and the main advantage of this is that you don't have to do the bookkeeping manually of like last time I ran my job which data did I load versus not and how do I need to upload downstream stuff because the streaming engine does that for you the real time is less in some cases it's less important and than this reliability aspect like how can I launch at once and then not have it page me for like months because I have other stuff to do with my time so that's that's the main reason why people are excited but of course some of them are also doing pretty real-time things where they want to update something within seconds of a user doing stuff cool let's see what else - his are we planning to use fibers haven't looked into that too much yet it might make sense it's just when you think about these are its contention as I said there are so many other things that can be contended for such as your database your memory all these things so I think it's instead of like focusing on just one mechanism at a time like let's say to use threads more efficiently I think it's even better to look at an architecture where you know something going wrong it doesn't mess up everything else what else is going on let's see Comcast use ml flow + scoop flow if you if you think I'm also doing you need to come flow so so these this is a pretty specific ml question they do different things so coop flow is about how to stand up on ml stack easily on kubernetes which is useful if you're earning that but that's the main focus whereas ml flow is more about you know regardless of where you're earning on how do you have kind of this centralized hub to manage ml development that's that's another interesting thing with ml I could have said is that the there's software development and maintenance life cycles pretty different so we're trying to figure out what to do there cool we shouldn't buy pirat I know if you ask it yet he asked whether client implementation language plays a role in developing large-scale system so I guess being a subscale I wonder if you see differences in systems where clients actually are written in Scala versus clients right with Python yeah yeah that's how am I saying most of our services are written in Scala and that includes like how they talk to things and we like the JVM and Scala both of them for you know for different reasons but it's just a big ecosystem with really good monitoring and debugging tools so it's like awesome to have that in terms of to end clients we don't see as many issue with with those really because because the place where we're using it it's all in the JVM so I can I can't really say about that we do have a lot of users using Python as well but these are more like the workloads that are in the data plane they're not doing you know huge amounts of requests per second - or 30s api's yeah and we use G RPC for like to actually like you know package api's and stuff so you can easily create alright and if as its allowed to ask questions really yeah first of all thanks a lot for this presentation and thanks motif the delivery angelic c-44 engine it was really insightful so my question is mostly related to the foundation you selected for Delta light project spark was like the place was park is the patches of their foundation and if it's possible you're sharing your thoughts why you decided to fight fire find a new home for your next project it's not a specific reason I think they're both great foundations and they're you know they have like really interesting projects and they have you know like very solid communities behind them so I think with with Delta lake and an ml flow which are in Linux Foundation we also wanted to be close to the - a lot of the open source deep learning and and cloud computing software so kubernetes is there onyx is there horrified feather storm like some of these these deep learning software's are in Linux Foundation so we thought that you know it would encourage more integration with those but it's not it's not like a huge difference either way like we you know and and there are other things like we also contribute for example - to either Apache projects such as Apache Parque and you know we'll keep working with that so well and of course we contribute a lot to Apache spark we continue to do that yeah so and you know we also like we hadn't really worked as much with the next foundation before so we also wanted to be more visible there's a lot more of those folks interact with us and you know we make sure that these things integrate well with spark as well be so instead of a completely separate stack happening over there yeah I guess the other thing I should mention in the talk as we are if you're interested in these problems just like you know shameless speech but we are hiring especially senior engineers on these cloud platform teams so definitely you can reach out to me or whoever if you if you if you are interested in some of these and you know we'd love to talk with you yeah well other questions I see one on where is spark moving in terms of deep learning support and GPUs yeah so we have in spark sleep oh I know the scheduler has now aware of accelerators so that means you can on libraries that one or acquire GPUs and do stuff with them we don't have too much in spark itself yet but there are pretty good integrations with existing deep learning frameworks so we're hoping that those are going to you know those are going to kind of explore like what's the best way of using these together and then maybe over time we'll add GPU support and some of the things in in spark itself so that's kind of the way we've gone with that yeah and we have a lot of use of Apache aho for streaming data in and out of the engine it's used for Python and R and it also makes it easy to scream stuff into the GPU actually screaming based on event time rather than processing time yeah so structured screaming has support for that that's like the newer streaming API is based on sequel so start if you look for like structured streaming an event time you should be able to find something some examples basically you can say which of your columns is the event time and do stuff with adapter is one of the key reasons that we develop structure is do you mean I'm just googling it to see if it's easy to find yeah there's actually it's in the programming guide but yeah maybe it's not super obvious from from the websites that you know that's the thing to look at the older API the low level sparks do you mean one doesn't have that yeah the spark have inbuilt support for event sourcing there are a lot of people using it with change data capture to do a event sourcing but there isn't like a specific API people are just building up themselves on top I'm going back to the to the streaming event time you mentioned that seco framework is that only available in C code can it be used for simply transforming from one stream yeah you can use it to have custom transformations too but it's based on data frames and the sequel data model underneath it so so you can combine sequel operations and non-sequel and and other stuff yeah let me send a link to that actually yeah that would be great and our in our use case we there are more attributes than the scholar limitation of 22 fields so I was looking yeah yeah this would be a good fit for that cuz because you can use it you can have a table with many attributes and then you can pull out just some columns into scholar objects if you want yeah and it's the same as the data set and data frame API so you can easily mix like custom logic with you know with sequel operator if you and you know when you see co-operators it's likely to be more efficient but yeah thank you okay maybe I can take a couple more questions I know it's basically lunchtime for people in today or at least yes and unless you guys have any other questions I think we should thank ma te for the great talk and we'll publish this video on factual TV which is our usual youtube channel it will be linked from the meetup itself and I you guys are always welcome to comment and some questions I just wanted to to note that it's very interesting to me Matane what you presented is basically a reactive system as defined in the reactive manifesto she's the resilient elastic message driven so this is very interested to me you know kind of the launch and reactive foundation that we are kind of grappling with the question how can we enable all developers to write this kind of applications right so I think it's very it's very instructive to learn from this lessons and the question I guess for for us is can we package these findings into easier IP is for developers because developers should not be really thinking Jeep about you know scaling the backend right like this should be just using little magical scales so so this is yeah yeah yeah that this is something we're looking at as well like whether we can adopt something like this or more more reactive sort of technology so it's definitely on there's just it's it's kind of hard to duties as the the thing is growing and changing underneath all right right yeah it's it's definitely it's like one of the leading ways to handle these problems yeah we have you know that's it's not yet official but we have our two DB C which is the reactive DBC driver from being where and multiple contributors kind of MariaDB and oracle and others soon inducted into the rock the foundation and I think the interesting question is if you really go reactive you have to go into end because if you just do a piece of reactive you will have to have a legacy systems and will not see big payoff right so that's I think an interesting question if you can go reactive m2n but up with with streaming which is just made a feat for it and now our JDBC for reactive I think we can just a question we would like to see Delta like supporting reactive approach that would be really really cool there that would be a very good time to try to do here thanks guys thanks again for joining us thanks mati and we'll see you guys at the next Meetup thanks a lot yeah thank you everybody