Cloud Native Computing Foundation Kubernetes Batch + HPC Day NA 2022, 1 Nov 2022

Previous Meeting

Next Meeting

⏯

youtube image

►

From YouTube: Beyond Experimental: Spark on Kubernetes - Weiwei Yang, Apple

Description

Don’t miss out! Join us at our upcoming event: KubeCon + CloudNativeCon Europe 2023 in Amsterdam, The Netherlands from April 17-21. Learn more at https://kubecon.io. The conference features presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects.

Beyond Experimental: Spark on Kubernetes - Weiwei Yang, Apple

Apache Spark on Kubernetes takes advantage of containers and the large, rapidly growing Kubernetes ecosystem to maximize the data processing capability on the cloud. However, running a large-scale production environment is not an effortless combination. Challenges at scale, dev-ops complexity, multi-cluster management, job scheduling, and autoscaling are all roadblocks that could quickly fail the mission. In this session, Bowen Li and Weiwei Yang will share their insights on leveraging open source technology such as Apache YuniKorn, Spark K8s operator, and Cloud primitives to evolve ML data infrastructure in the cloud, including considerations for multi-tenancy, observability, scalability, and cost-effectiveness.

A

um Hello, everybody, my name is weiwei and I'm from Apple. uh I am a data infra team and today, as we're supposed to be me and Bowen, to give this presentation, but Bowen couldn't make the trip here so I will be presenting this along and the session is beyond experimental spark on kubernetes. uh If you are running smart, no matters on kubernetes or not uh this talk. This is the right room for you and, if you're not not running, inspired you're running some batch jobs, I think this is also very relevant.

A

um So today's agenda is we um the first part I will be talking about how to build the cloud native data platform, so everybody is talking about Cloud native, and how does it mean for a data processing platform uh where the backend is sparked and the second part I'll be talking about. So basically we build the platform.

A

We actually made a lot of challenges, I I believe most people do, and so we basically talk about how we Leverage The Open Source software, to tackle the challenges, and then uh the third part is when we go to scale and we are seeing a lot of more problems.

A

And now we are talking about how to addressing these issues when we scale up- and the last part would be talking about the future work- and first thing is that um what we're doing here is to want to build really build a cloud-native batch processing and they're interactive, analytics platform on top of kubernetes and also powered by Apache Spark, and these were handling large data processing, machine learning, workloads and for different users, all sorts of jobs.

A

So, basically, the platform where supports like scheduled, workflows, ad-hoc queries, also have a batch jobs and interactive sessions and for job types or help some of the jobs I'm really Mission, critical and the they don't they want. The job can be done as soon as possible, and also there are some other POC jobs, user, randomly submits sandbox jobs and from the jobs execution. uh Some of jobs may just take a few minutes or even seconds, but some of them are taking days.

A

So, given all these things together- and we are essentially building a pretty complex system where we want to support all those workloads and also do not give user complexity right, users just need to submit their jobs and everything will just work. This is this is the user expectation and our work is just to make it simple based on the cloud native software and the ecosystem, to make it happen.

A

um First, let's talk about the architecture Evolution, so I'm from the old Hadoop word, so I I spent a lot of years working on Hadoop, and so in the past days that we are running like spark on on thousands of nodes. Cluster at that time, we won't really want to build a really giant clusters. The reason for that is basically uh at this time, so data and the compute are sharing the hardware.

A

So when you want to store more data, you need to scale up the cluster when you want to run more computes, and you also need to scale up the cluster and that's why we end up keep adding nodes to the cluster and essentially that the nodes, the Clusters just got bigger and bigger. So uh when we want to replicate everything on on on Cloud on on kubernetes, the first thing came to our mind is like we can't do the same.

A

We can build a giant compute cluster and the beauty of kubernetes like kubernetes were on cloud is like we can disaggregate disaggregate, compute and storage. So we can, we can run operate our own compute clusters. So why don't we just build really large compute clusters, um but the first attempt definitely not the right solution, and when we, when we evaluate these parts, we realize that there are two major issues.

A

One is it's really hard to scale kubernetes with thousands of notes, so we have tried, but uh after 1 000 nodes, um we are sending a lot of problems and we don't want to deal with that, and the Second Challenge is that sharing the resources between batch jobs is really hard.

A

um So when we put all those user jobs on the cluster and basically users are constantly competing resources. Teams are complaining about why I'm not getting resources. My why my job is Stark. So how we solve these challenges is a major problem and that's the that's why we go to the next the Lexus stage, and then we want to involve our architecture to be more Cloud native and also a leverage.

A

What we call this Dynamic compute boost, so Dynamic compute pools, is really nothing but just a lot of more just split, the giant compute cluster to a lot of more smaller clusters, the Clusters where we are comfortable to manage and we are comfortable comfortable. They are be running just fine and give us lights overhead and so to give users that we actually had a layer here, which is the number one um coded here, is the batch processing Gateway. The reason to introduce another layer is to hide the complexity of how many clusters we have.

A

We don't want users to know that user doesn't need to know where are their jobs will be running on which cluster? This is not something user really concerns about. So we introduced a near doing all the this is the basically job API for all our under users. The user just needs to interactive with this Gateway, but underneath will be multiple compute clusters and each cluster will be looking like just around the right side and we are leveraging coupon spark kubernetes operator to handle the spark job submission on the kubernetes cluster.

A

That is easy to use, and we love that. But then remember the SEC. The Second Challenge is the resource, the scheduling on the each of the individual kubernetes clusters, and that's why we have a kubernetes batch scheduler instead of using the default scheduler and with that put them all together. We have architecture like this. So on the left side, this is what we exposed to our un users.

A

So first user will need to talk to the batch processing away, which is a bunch of apis res API command line tools, so they can submit and monitor their jobs. Spark jobs and second part, is the tuning and the monitoring which is collecting the present in the resource.

A

Ui also spark history server, where user can cause their historic jobs and the profiler which user can use that to to tune their jobs, job UI, where user can track their jobs log in the metrics, how those those things are exposed to users, and uh essentially these things are the only things that users would want to deal with, but on the right side, that's the compute pose and each of the computers.

A

We are managing data, separate kubernetes cluster, and we use the Unicorn scheduler to replace the default schedule to the scheduling and for each of the Clusters.

A

We do the skill up and down uh by leveraging the cluster Auto scaler and also use the smart operators that meet and manage the spark jobs this way, and also for the computers where we are able to scale our skill in which means we can spin up new clusters on demand when it is necessary when our ecision clusters has not enough capacity and also we can destroy those clusters when when we don't need them. So, with all these things put together, we really come up with a cloud related platform where we can serve our purpose.

A

The recipe here so basically uh remember the challenges we mentioned so there we need a scheduler that works for batch jobs. In our case, it's the spark jobs and uh that's why we choose our parts unicorn and uh we'll talk about more about unicorn later, and the second part is the service Gateway. That really gave this user very, very simple interface. They don't need to need to deal with all those infra complications.

A

So that's the batch processing, giveaway and the third part is really empowered by kubernetes ecosystem, because within the kubernetes ecosystem there are so many components, great components. You can choose to do the networking to do the Locking Matrix.

A

So all those things we can we can leverage and in order to build our cluster to our our platform better and better.

A

A typical workflow is like, when users submit their jobs, they are interacting with our Gateway service, so the Gateway service actually exposed the rice, API and command line, so user can simply submit their jobs and once the job is submitted, Gateway the Gateway actually will send me the job to your queue. This queue is a very Central concept. We have our system. uh This is actually the virtual cues.

A

When we see there are virtual cues, because overall, the compute clusters, we just gave user a central virtual queue to manage the resources um and this queues resources are actually coming from different one or more kubernetes clusters and then, after the job is submitted to the queue the Gateway will dispatch the job to a dedicated cluster, to run the run the job and in our design, the job will only be able to run on one cluster. It's not crossing the cluster.

A

uh We are avoiding a lot of complicated things when you want to deal with one job to multiple clusters and when the job is dispatched onto a direct cluster and the spark operator will interpret the jobs back and launch the job on that cluster. After then, once the job is launched on the cluster, that means the job will create a bunch of PODS and unicorn. Scheduler will reward at the pending part and then schedule all the parts on on this cluster, and also it will repack the queue settings. So everything will be coming up together.

A

The batch processing Gateway we are talking here, is a central, stateless job, API server over kubernetes cluster for spark and the the key idea is to we want to hide the kubernetes Clusters behind behind the scenes, and we really want users just to use the API to submit and monitor their jobs instead of caring about which cluster my job will be running. Do I need to log on to the cluster to see what's happening, so we want to add all those complexity, and it's also manage the virtual cues.

A

So, basically, the virtual cues we're actually mapping to the Unicorn actual cues on those compute clusters and we aggregate the metrics, and then we have a virtual queue present to other users. That's all the things they need to care about and also the Gateway will handle the load balancing and to avoid like creating super um hotspot clusters and also it will give user the essential apis to track their jobs status and also retrieve logs.

A

uh This is basically what we are thinking to bring this platform to. Our users is what we really want to make it very easy to use from the user perspective they can. um They can access the gateway to to operate their jobs. They also have great observability, based on the uis and the logs that we we provided and also the monitoring Leverage The. The system will have built. They have a great monitoring metrics over the jobs and on the right side. You can see how much complexity we're hiding from this from users.

A

Cluster provisioning user, doesn't need to care about that cluster upgrade they're, we're doing that quarterly for kubernetes versions, scale up and down is automatic on demand. Ip rebalancing basically their IP limits. We want to avoid instance, capacity planning, node, balancing workflow schedule, finial recovery, all those things are hided from the user and also more important job queuing, pinpacking resource fairness. Those those key features are provided as well log rotation archive.

A

So all these complexities are not seen by the user, so user would be just happy run their jobs on our platform and they will be handling all the great parts.

A

um We keep talking about the virtual resource cues and really want to reinforce that. The key idea here is: we have some virtual cues and this. This is where our teams and users plan plan their their budget for the year and they basically tell us how much resource they want, but their resources for each of the cube may come from one cluster.

A

We may come from 10 clusters, so how we provide a resource for the virtual cues is totally depends on how we plan the compute pools and also once the user's jobs are submitted, the queues, actually, the jobs will be queuing and also will be scheduled, based on number of factors such as priority submission time and sometimes resource usage based on what user needs, and also all the Gateway provides the tracking tracking capability for user to be able to view their jobs retrieve the logs blah blah blah.

A

So the virtual cues actually give the user a really good way to track planned resources and also hide all the complexity behind this. Behind of all the architectures that we have so many clusters, so batch processing giveaway will be open source as well. That's why I encourage anyone interested to search this project on GitHub.

A

Then the second challenge is the job scheduling which we we need to talk about more. This is uh basically what unicorn fits in the picture. uh Unicorn is the open source batch scheduler for kubernetes and it over the default schedule actually provides um some key features that are needed by the batch system. The first one is the resource code, management and I know that user folks are using namespace Code as bad namespace.

A

Resource code has not really designed for batch systems and resource code management, job scheduling we need to queue the jobs and also we need to really schedule jobs, not as just schedule the parts and the jobs has their priority. There has some different factors we need to consider during the scheduling, so that is also considered um and beyond that we have advanced scheduling, features such as gun scheduling and also there's most recently. The committee has a scheduler plugin deployment for some of the use cases, and one last thing is the circle.

A

So really cares about throughput and we really want the schedule to be first as fast as enough to catch up with the user requests, and so, basically, with the Unicorn that we are. We are dealing these challenges, we're able to solve the problems for queuing jobs and also, and also for the resource code and management and for the performance. You can see that from the our Benchmark, it's a list twice better than the default scheduler on the same environment and for research.

A

The Q use case um unicorn actually provides this cute concept for the users and for them to be able to manage plan their resource and also sharing between teams and users. Searches, guaranteed resources is the minimum resources that you can get from the queue and based on that, we are when the Q utilization is below that. Basically, we know that you are, you are starving you, you need more resources, so your weight will be higher than other cues.

A

Your game resources faster than other cues Max resource is the hard limit, but the casing here is that even we are users submitting jobs to to the same queue and it won't go beyond the hand limit, but the jobs will not be killed. The parts will not be killed. Even you submit more and more to the same queue. So that's the case encourages basically that simplified the user client side. They don't need to handle a lot of fingers because of a quota exceed.

A

Then there are some use cases that we have the hierarchive cues and in the old young world we have a lot of use cases using hierarchical, cues to for teams and the users under the teams, and we can use the same concept here in unicorn beam packing. So that's another feature built in with unicorn scheduler, basically squeeze the ports until minimum number of nodes. This is actually works very well with auto scaling, and but that also has some set effects such as creating some hotspots and sometimes creating dicks pressure.

A

So there are some optimizations to distribute jobs based on the jobs and type like what is CPU intensive or memory intensive. So all those things are job level metrics and that can be leveraged by the scheduler and when we scale up right. So the challenges we have- and here are the tips.

A

The first thing is to be disclaimed, because we won't really don't want to skill. You know to thousands of single cluster um we've seen a lot of problems. We have to do really really hard clean up in order to get a cluster uh back um so in our project. Maybe this is not a not a rule of sample, but in our practice we just keep the cluster size, 200 000 1000 nodes, which is running pretty well very well in most of the cases.

A

The second thing is the agenity, so basically we really want to leverage Cloud. You know um how easy to build clusters, how easy to destroy them. So we really want to leverage that, and the third thing is elasticity and all the nodes, if, if possible, put them onto Auto scaling group and make sure they can allow skill on demand and also provided good observability. So you can detect issues and also diversity is the key. Where you are running. You know um workloads on cloud.

A

You need to leverage as many instance types you want, and automation is also very, very important um because we are having those complexities and those complexities are actually coming to the infra layer and they want to have the good tools to to automate them. So this is a like the pattern we see. Does it already doesn't have any pattern, but basically it's on demand, so user can submit anything to the cluster and it can scale up and scale down, and the number of clusters can also scale up and down.

A

So this is a casing that we are running the instance that will really based on our demands, we're not going to run preserve resources for, for any reason and uh um further research. We are looking at several directions. One is the remote shuffling. A lot of people know that spark uh has those shovel data on local disk, which is actually creating some problems when we are running on on cloud and so remote shuffling is moving the Java data to remote storage.

A

That's definitely a way to look at how we can uh to make spark more Cloud native and also there are some efforts in the community on the new, auto scaler to replace the cluster, such as the Carpenter and also ocean from spot. So.

B

A

Think those projects are actually very good idea to to to towards the no group in these cluster setup. Actually that will save a lot of Dev efforts, I'm running a lot of time, so that that's all about my presentation, questions.

B

C

C

How are you constructing the virtual cues because it looks like with the scheduler? That's actually a cluster local concept.

A

So the question is how to how to construct the virtual cues and the version Q. So for each of the physical cluster. We have a unicorn instance running and the Unicorn instance will provide the actual cues on that cluster, with virtual cues. What we mean is that for for one queue in the virtual queue that can actually go to multiple kubernetes class, one or more kubernetes clusters for the same Q resources and for each of the compute cluster queues.

A

You still have the mean Max Capacity setting and when you sum them up, that's the Max Capacity in the virtual queue. That's basically how it works today,.

D

Hey uh really great talk um a lot of it's remarkably similar to what we're doing with Armada. Actually in the setup, I was wondering um your batch API said it's stateless. Have you achieved that because does it not need some kind of state to know after you've submitted things where to find out information about jobs, for example?.

A

um uh Matt Matt ask you to repeat your question.

D

Yeah um the batch API, the batch IP API I, think I think you called it the batch or Gateway sorry I'm, like yeah I, forgot Gateway.

A

Okay, wait: okay,.

D

The Gateway API, um that's for how is that stateless? Oh.

B

D

Not need to save some state to find jobs or workloads.

A

That's right, that's right, okay, so the question is about how the batch processing Gateway is stainless. Yeah, um so it it is, um it is I I, don't think it is um typical. You know it's dailies. It still has still high State. We have a database, and so it writes the so because it's basically tracking all the jobs. So actually we are writing those data into database that we are able to run different instances that might be into the same same database. So still we can easily turn down our instance and skipping up another.

A

So we run several instances together to do load balancing um that's what we we want it to be. So it's not a no state. It still has State makes.

D

Sense thanks foreign.

E

So for this Gateway, what exactly the users submit to this Gateway.

A

um So this the Gateway that the question is what users submit to this Gateway is that right? So basically, that's.

A

um For example, if you send me jobs to uh kubernetes classes using Spar Summit, you have to inspire some a blah blah a bunch of parameters, and what we do is that user will not use parsa major use the Gateway, comma line or rest API to submit their jobs, and then they still need to give a bunch of parameters like how much um how much, how many, how many quarts, how many executors you want to run and which queue you want to run and what is your class main class?

A

So all those things are provided to the other job submission API and that will be submitted through a right service to the Gateway, and we know that then we dispatch a job.

E

All right, it's essentially something like a kubernetes resource, but you put some extra metadata in there that you need.

A

The Gateway is more like API server for spark jobs. It's not doesn't actually doesn't, um doesn't necessarily to do any resourcing stuff.

F

Hi, so for log and Metric aggregation. Is there like one large instance of Prometheus or elasticsearch that users are accessing logs through or is there some sort of segregation between users accessing X, Y and Z, or does each queue associate with an instance of Prometheus and Etc.

A

You are asking about logging.

F

Yeah, specifically logging and metrics that you, you said that users have access to right.

A

Right, so logging is basically I, think we we have done nothing different than others and we just leverage our uh the like fluently decline logs and uh put it on on the storage, and uh we just in our service Gateway we open some. We create some API for user to retrieve the logs from the remote storage and I think it might be. Most of people are doing today, I.

F

See so there's the API annotation in between yeah.

B

Yeah I have a lot of questions, but uh hopefully we can talk later, but I wanted to focus on now on the cluster autoscaler, you say that uh unicorn supports clusterity scaling yeah, but um so I asked because as far as I know, the cluster Auto scattered at least they open source, one just embeds the cube scalware code. To do the simulations so I wonder how does it work with uh yeah.

A

Actually, um actually for the cluster orders here and so right now, what do we have done? Unicorn it works with classical perfectly, uh but when we want to use different um strategy on the schedule like we want to um in some cases we might want to distribute the positive differently on on the nodes.

A

Sometimes there will be conflict, that's why we are not doing that today and to be able to fully compatible with uh with the cluster Auto scaler and, in the last part of my session, I mentioned about some other autoscaler project and uh and I. Think right now, the auto scaler is is doing part of the sketching work, because you need to do the computation and see how many instances needed, and so sometimes that creates some challenges.

A

I would say, and one of the directions we're looking at how to you know, really make uh unicorn schedule, works better with cluster Auto skill and also other Auto scalers. So that is a little bit unknown for me at least for now. Definitely some something we can talk about more.

C

I think we are almost okay, one last question and then we have to.

G

uh So like from in Jamie's talk, there seems to be a lot of interest in like multi-cluster and general, like workloads, I was curious from your from this work. Are you able to submit uh like MPI operator, jobs or, uh like you know, or other types of crds into this, or is this kind of more geared towards spark yeah.

A

This is right now purely for spark. um So all our computers are running on sparse. So there's no general. You know crd's work.

C

All right, thank you so much. Thank you.