Scale By The Bay 2020: Jean-Ives Stephan, Cloud-Native Apache Spark: why & how to migrate your Spark
[Music] today we're gonna be talking about cloud native apache spark why and how to migrate your spark pipeline to kubernetes um thanks a lot for having us by the way at scale by the bay uh you already gave the personal introduction i used to be a software engineer on databricks now for the past two years i've been working on data mechanics and when i'm not working i like to do some trail running like with the photo on the right um so before i jump into the topic really like deploying spark on kubernetes why to do it how to do it i just want to give a few words about data mechanics we have a bit of background um you know how come i'm knowledgeable about spark on kubernetes so our goal at data mechanics is to make spark easy to use and cost effective with a focus on data engineering workloads easy to use through the dockerization we'll be talking about this later but obviously that's one of the benefits of running on kubernetes we also have some intuitive user interfaces so that you can track your logs track their metrics and we have some integrations with the tools you already use like jupiter or airflow how we make spark cost effective it's a long topic but one of the main things we bring to the table is automated tuning um it means that our platform is going to automatically tune infrastructure parameters and spark configurations like the amount of memory and cpu to allocate onto your spark executors like the type of disks like some spark configs around parallelism around shuffle and we're going to be managing the node poles of kubernetes we're going to automatically scale them up and down based on load we're going to use a mix of spot in on domain instances if you want to to to again reduce your cloud costs and this slide is maybe the most important because it's most relevant obviously for our talk our platform is deployed on a kubernetes cluster inside our customers cloud account so in fact if the customers uses say amazon we would be deployed on the eks cluster and in fact we create and manage the communities cluster for them now on this cluster there is one instance running all the time that we call the gateway and the gateway is the entry point for launching spark applications um whether there's our interactive notebooks or whether these are script um or you know airflow applications or coming from any any other scheduler once the request hits the gateway we're going to actually launch the spark pods configure them scale them and give you access to a monitoring ui so as you can see our entire product our entire startup is built on top of spark on kubernetes and so this is why i want to talk a bit about it today um so first we'll go over the the core concepts of spark kubernetes then we'll be talking about really the reasons why more and more companies choose kubernetes and then uh we'll we'll go through like concrete tips to help you get started with sparkling communities so a setup checklist some best practice practices around performance around the cost efficiency around monitoring and at the end we'll be talking about the future works that are happening for spark on kubernetes in spark 3.1 and beyond great so first um before i actually even talk about kubernetes um just recap about spark i mean it's a distributed analytics and gene for big data and machine learning the reason why spark is popular first it's fast second it's easy to write a spark application because you can um yeah there are these high level apis in python in scala and sql and more and last uh it's versatile because with spark you can do obviously etl and data engineering that's the most common use case but also interactive data exploration machine learning streaming and so on now where does kubernetes fit within spark well one thing spark doesn't do is um is manage um the the cluster manage the containers manage the executors um it can request executors but it's not going to launch the instances to run them so for that it needs a cluster manager and kubernetes is a new cluster manager for spark the other cluster managers are just standalone which is built in with spark apache missiles hadoop yarn which is by far the most common cluster manager today and kubernetes which is the new kid in the block it's been available since spark version 2.3 in 2018 so what do we mean what happens really when you when you submit a spark application on kubernetes well we say that spark runs natively on kubernetes because there is a direct communication between the spark driver and the community's master so when you when you submit an application it's first understood by the communities api server who is going to then start the spar driver pod so basically the spar driver container and then the driver container and the committee's master are going to be directly talking to each other to request more executor pods and and once once they're ready obviously start running tasks on these executors um and in fact this communication can happen during the lifetime of the spark application if you choose to enable um dynamic allocation um now there are two ways to submit spark application on kubernetes the vanilla way is spark submit so this comes from spark open source main repo app management is a bit more manual but you know but but you know it works there is an open source project that was developed by google and that is called the spark on kubernetes operator and it is a bit more popular because it provides a lot of them well sugar and nice features on top typically it gives you tooling to read logs configure your app kill it restart it schedule it as if it were a kubernetes object so to to show this in practice if you if you use spark submit well basically you're going to provide the url of the committee's master and that's the only difference between reading a spark submit over say yarn and kubernetes but then if you want to interact with the application well you're just going to have to interact with kubernetes pause directly now if you install the spark on kubernetes operator well you start defining a spark application and you can manage it as a community's object so you can read the logs describe the application even schedule it and so on so honestly sparkle companies operator would be my recommendation the only downside is you need to install one long-running system pod to to have this operator running but it's a very small overhand overall okay so now that we have the core concepts of spargan communities let's look at the main reasons why companies choose to run spark on communities over yarn so first that's probably not a surprise but one of the main benefits is just containerization so with kubernetes you're going to be able to package your applications with docker and well that's going to give you a very tight control of your environment it also means that you're going to be able to use the same docker image and run it locally maybe for development for testing on a subset of the data and then run it at scale on the quantities cluster the other benefit is obviously you're going to be able to package all your dependencies this way you can also change the major spark version um and and obviously you're going to have the exact same environment whether you're in dev test and prod and and last but not least the iteration cycle while developing with docker can be really fast it's it's not true to say oh it's going to take forever to build and so on because docker is going to cache the previous layers of you know if you're iterating on your app you probably push the docker image at least once and so um overall you can get uh of an iteration cycle of maybe 20 seconds between the time you you make a change on your laptop to it runs at scale on an application on the communities cluster um well the second benefit is um it's just the benefits of a shared infrastructure on yarn you have little isolation you typically have a global sparked version global scala version some global dependencies and so as a result there can be um yeah there can be they can be i guess impacts from running multiple uh spark workloads on the same yacht cluster and if you want um your application to be really isolated and stable you're gonna you're gonna spin up a yarn cluster for just your single spark application so what we we call transient clusters or single tenant now on kubernetes um you have a full isolation and control that just comes from docker images and so what this means is even though you're going to have multiple spark application on the same communities cluster and so what this means is you get kind of the best of both world isolated applications and the cost efficiency of our shared infrastructure what's more kubernetes has a lower overhead than yard when it comes to you know starting nodes and starting containers on it and so you have a really fast startup time basically if a node is already available it takes maybe 5 to 20 seconds to to start a spark pod and if a node is not available then you need to get it from the cloud provider but it's pretty fast as well maybe one to two minutes um the the last benefit from using kubernetes is is the ecosystem of the kubernetes or docker compatible tools so first you know kubernetes is the cloud provider abstraction so it means if you build spark on kubernetes there will be a very little lock lock-in compared to being a building for a specific platform or specific cloud provider apis and then you get lots of open source projects for free around the mcicd logging and monitoring networking security and so on so i would say yeah these are all the main reasons why companies more and more choose to run spark on kubernetes over yarn um now that we went over the motivations let's talk a bit about how to actually you know get started how to migrate your spark apps from yarn to kubernetes so we have a bit of a checklist here um i won't have the time to go into every uh every you know end-to-end way to get started but the basic setup is obviously you need to create the cluster with proper networking data access and node pools i would then recommend to install the spark operator which i mentioned earlier and also the cluster auto scaler which is an open open source projects which will dynamically add nodes to your node pools when the communities cluster would benefit from it on the monitoring side i'll go over this later but you probably want to have some collection of logs and persist them to a storage and and similarly for system metrics and for spark ui and in terms of um yeah some some important optimization stuff you should use um local ssds to improve the performance of shuffle that's pretty pretty critical and also maybe configure spot now spot nodes and optimize your applications so that typically you're not wasting too much room on your nose and your pod being packed efficiently on the currently's notes so let's dive in into a few of these topics now so first yeah performance best practices this is the number one gotcha when people start working with kubernetes let's say you have nodes that have four cpus and so pretty naturally you're just going to submit an application and say spark spark.executors.course equal four but if you do this you realize that your your executors will be stuck in a pending state why okay because um even though your nodes have four cpus there is a small number of resources that will be reserved for kubernetes and system daemons so the node allocatable isn't quite for cpus it may be just 95 percent of that then if you're running daemon sets this could be another small overhead that goes to your demon sets let's say that at the end you have a 3.6 cpus available for your spark pods well then when you configure spark you should set the second this config spark dot communities.executor.request.course equals 3.6 and then spark executor course can still be four and when you do that you realize that you're actually overprovisioning a little bit the number of of spark executor cores compared to the actual cpu request and in fact the two numbers could be decorated so you could say oh spark actually you you could tell spark oh actually i have eight cores but still have only 3.6 um cpu allocated and that's really called over provisioning in some cases they can this can be beneficial for for performance um if basically your cpu isn't uh use usage isn't that high it's it's something that could uh provide the performance boost all right uh this is the kind of of settings that is automated by the way on our platform so now a critical yeah performance tip around shuffle so we we published a blog post with a benchmark where you could find the results and learn more but just the tldr is that you should use local ssds whenever it's whenever possible so um yeah you should look at your cloud provider to know how that's done and it's not enough to just use them you should also configure spark to be able to to actually use them and so here on the right we're showing you how to do this with the host path and that gives basically um uh the the path to the mounted disk and there can be a performance gains of up to 10x on i o bound workloads so it's really worth it it's worth the additional price of paying local ssds now if for some reason local ssds are not available then the other important parameter is the size of your disks as disk latency and throughput is almost proportional to disk size so here we this is the the plot of um the ratio of the query duration that we got while running the tpcds benchmark once with you know small disks and ones with larger disks and you can see that for shuffle heavy queries on the right there was a ratio of five so you basically um your query would run five times faster if you just increase your disks i forgot by how much we increased them but this will be shown in the blog post uh yeah so that's the second t pretty on shuffle performance now that we covered you know shuffle performance let's go over cost cost optimizations so one thing that we do at data mechanics is we configure the committee's cluster to have multiple node pools and these node pools should be able to automatically scale up and down and also scale down to zero okay not not so that you're not paying anything when um when you're not running any workloads uh so typically we might have um a small on demand node for some system pods like the spark operator we mentioned we might have an m5x large node for or node pool for driver pods and so here it's important to have them on on demand because you don't want to lose the driver if if there's a spot kill because that would kill your application but then spark executors these are fine to put on spot nodes because as you know if you lose a spark executor the application will still continue it will lose some work you'll have to redo some work but it will continue um so here's an example um set up on the amazon eks where um we had to define a labeling skill scheme for our nodes so we have basically two node labels let's say life cycle and instance the life cycle is to decide between spot and on demand and the instance is to use the instance type and then you're going to have to manually create auto scaling groups and assign them corresponding tags and then use those labels as tags to basically inform the cluster auto scaler and this way when you submit a spark application based on certain node labels the cluster auto scaler will place them on the node pool of the right tags and maybe scale the note pool if necessary so let's look at here maybe the example with the spark operator you could say oh i want my driver to be on a preemptable machine well i think this should say on demand and i want my executor to be on spot and so um and so basically that's how you would achieve this um great yeah another thing about cost reduction is dynamic allocation so dynamic allocation is a way for the spa driver to add or remove spark executors on the go dynamic allocation is available on kubernetes since spark 3.0 so it's available through a mechanism called shuffle tracking uh what's shuffle tracking basically um the spark driver will track which spark executor is uh holding an active shuffle file and these executors will not be scaled down so this is the configuration that you need to to enable to to make this happen and it works it works pretty well we'll see that in spark 3.1 there'll be another improvement to dynamic allocation um yeah maybe one last trick around well how to speed up dynamic allocation or how to speed up application startup time in general and this is something that we also enable in data mechanics or make it configurable you could choose to over provision the kubernetes cluster with some low priority pause pods so a passport is basically a pod that does nothing and if let's say an application request a new executor because of dynamic allocation well then since the passport has a low priority it will get evicted by kubernetes and so the spark executor pod will will take its place and then the passport will be rescheduled again and this might trigger a scale up and so you see that pause pods is basically just a mechanism for you to over provision the committee's cluster a little bit pay a little bit of additional cost but then have a lower latency for your applications it's definitely a better option to do a bit of over provisioning than to have a fixed size quantities cluster where you would be wasting much more compute on average right so now monitoring yeah this is important so there are a few different monitoring options available to you when you run on kubernetes one of them is the community's dashboard i've put the link i'll send a slice afterwards by the way but i've put a link with the instructions to install it on the eks and so well it gives you a general purpose monitoring dashboard where you can track metrics for all your parts the only issue is that it's not specific to spark so it's hard to really really reconcile what's happening with your actual jobs and stages and tasks and the other issue is that executors metadata is lost when the spark app is completed the other option you have as you see is the spark ui so for live applications it's pretty easy to access you just need to port forward the spark driver on a certain port 4040 and then the spark ui can just be accessed on your laptop and then if you want to have access to spark ui after an application is completed you need to set up what we call the spark history server so to do it yourself you should write a spark event logs to a persistent storage you can use a config for that and then you can install the spark history server using a helm chart and follow some instructions to give it the permission to read from the persistent storage now as of today actually we have just released a new way to to have access to the spark history server this is a we have an open source spark agent called delight which will stream metrics uh to our back end at data mechanics and then uh you could view the spark ui for your application at this url and this is a free and cross-platform product right now it's just a free hosted spark history server but as of january 2021 we hope to release what you see here on the right which is basically a spark ui replacement with a new metrics and a better ux and so this this free spark history server is just the first milestone and then in january and hopefully we should have something much more exciting to you great um last but not least i see i have another three or four minutes i just want to say a few words about the future works for spark on kubernetes in fact i'll start with just a timeline to to remind you that um yeah the first support of kubernetes came out with spark 2.3 in 2018 then in november spark 2.4 came out with some critical features like support for client mode volume mounts um dependency management improvements and then in june of this year spark 3.0 came out with the dynamic allocation that i already mentioned and um next month or maybe in january we'll see spark 3.1 should be released and this is an exciting day for sparkling communities because uh the experimental label will be removed okay so spark on communities will be declared ga because today if you look at the official documentation of spark it says that it's currently experimental but we're going to actually call it a ga and another feature that i'm really excited about is a better handling for node shutdown so what does that mean this will happen in a few different circumstances typically during dynamic allocation on downscale or if if a node goes down for example because of a spot kill so what will happen kubernetes will warn the node of the imminent shutdown then the spark driver will stop scheduling tasks on the executors okay because we know it's going to go away and very importantly we're going to copy the shuffle and cash data from this executor who's going away to an executor who's going to stay and what this means is that the spark application can basically continue unimpacted so this will make spark a lot more resilient um to spot kills and also it will also make dynamic allocation a little better another feature that i'm looking forward to but more for laser in the future is to use the remote storage to persist shuffle data i'll quickly go over that so what this means is that shuffle file will first be written locally then it will be asynchronously copied to s3 and then on the reduce side when you need to fetch a shuffle file if for some reason the original executor is gone and the fetch fails well then we can just fetch the file from s3 instead so again this will make spark much more resilient to node loss it will completely disaggregate compute from storage which is something that we always try to to achieve and so i'm also really looking forward to that great i see that um i'm at the end of my half hour i hope this was useful and i will now join the other chat for a live q a thank you all very much free to connect with us on our website on linkedin or twitter you