How Many Spark Applications Can Your Etcd Really Handle? - João Soares & João Azevedo, Feedzai

In this talk, João Azevedo and João Soares from Feedzai present their experiences handling Spark applications within a Kubernetes infrastructure, focusing on the performance limitations related to etcd storage. They describe the challenges they faced with operational efficiency and resource management as they transitioned to a cloud-native model, emphasizing the need for a multi-tenant architecture that supports workload isolation. The use of Kubernetes for orchestration and the incorporation of innovative scheduling through Volcano are highlighted as key strategies to improve application deployment. They also delve into the implications of storing job metadata and historical application versions in etcd, noting the impact of Kubernetes' multi-version concurrency control on system storage and performance. Finally, the speakers share their solution of delegating resource management to a custom API service with an in-memory store, allowing for more efficient handling of Spark applications without overwhelming etcd.

So, hello everyone. Well, I know this is the last day of CubeCon. Everyone should be tired, but we hope this is a fun one. We'll be presenting how many Spark applications can an etcd really handle. So, I'm João Azevedo. I started my career doing database research back in the university. I now work at the data reliability team in Feedzai. We basically manage most of the data infrastructure

as a service in a Kubernetes ecosystem. So, I'm João Soares. I work as a platform engineer across multiple domains, mainly compute and network. Uh I also come from a research background on high-performance and distributed systems and work as an assistant professor at the University of Coimbra. So, our use case. At Feedzai, one of our products is a real-time uh fraud detection platform for banks and payment processors.

Uh during the last year, in numbers, we processed around 90 billion of events, which totaled uh 8 trillion 8 trillion dollars in payments. So, fresh data is essential for accurate decisions. Why do we use Spark? So, this is the backbone for our batch processing, feature engineering and model training. Delaying execution of Spark applications can result in degraded transaction scoring and possibly undetected fraud. Regarding scale, we have

over 150 production Yarn clusters, isolated per client and use case. 4,000 Spark applications are launched each day, which totals three per minute across several regions. This is a growing number since Spark as a service on our platform, we've enabled a plethora of new integrations. as you can see, our previous setup was not the best. We had lack of resource efficiency. Basically, we had to manage independent clusters

per client and use case. We didn't have a multi-tenancy concept. And with all these independent clusters, we had sparse configuration over where everywhere, and it added operational burden in terms of managing all this configuration and also coordinating all these upgrades. But most importantly, we had to involve SREs in day-to-day operations. We had to page teams to actually help with simple operations like checking the logs of a

Spark submission or even more complex stuff like replaying a session of a job that has already executed in history server, for example. Well, to give you some context, in Vitsœy, we are undergoing a strategic shift, so we are basically moving towards Kubernetes consolidation with a single operational model and trying to adopt the more cloud-native approach to our infrastructure that is more cost-efficient as well. And although it

looks like moving our services to Kubernetes will make our lives easier, we will show it throughout this talk that we faced a couple of challenges that we hope would serve as learning lessons for for everyone. Well, Spark as a service was born in Kubernetes with three main requirements. The first one is we want users to have the serverless feeling. We want users to have a single entry

point to manage their submissions, and that would be an API and a UI. Um although Kubernetes simplifies a lot of our lives, it adds a lot of layers of complexity that we don't want users to be impacted with. Um at the end of the day, they may not even know this is actually running on Kubernetes, and this can actually enable integration with outside of Kubernetes systems, for

example. Well, second, we want we don't want platform team involvement in day-to-day operations. We want users to be self-serviceable with a single entry point that I just mentioned. And third, well, because we are trying to manage a multi-tenant system, we want to ensure proper workload isolation boundaries, and that not only includes security boundaries but also workload guarantees and limits quota systems all of that. we believe from

our past experience with Kubernetes that it provides already the right primitives to build this kind of systems. We just needed the right abstractions on top. Well, let's start decomposing our system. So, we will start with the user facing layer and this is a composition of an API and a UI. The UI would allow you to, for example, check the job tracking last past submissions to go to

a job that has already finished and for example replay and proxy to an history server pod to evaluate the performance of a previous execution of Spark. But things get a bit more interesting when we move to the orchestration layer. And we can actually use this layer to tell the full story. Well, the user is first exposed to a concept that we call a Spark cluster. A Spark

cluster is basically a virtual abstraction to the capacity that the user can use. It goes to our API. It sends a JSON payload with the job submission and references this Spark cluster. Uh we eventually accept the job in our system. Our operators will make sure all the underlying, let's say, Kubernetes resources will get created for the job to run and execute. At the end of this pipeline

of resources, we create the core resource, which is actually the Kubeflow Spark application object. And here a note on Kubeflow. We changed a bit of the submission works. We are not doing the the the job submission through the operator pod itself, which is the current state of the operator. We actually changed to do it in a separate Kubernetes job. We found it would be better for us

in terms of scalability, mainly memory usage, and also it would present a better security posture for our for the architecture we were trying to achieve. So, you have the submission job, then the submission job would eventually create the driver, and the driver creates the executors. This is a Spark workload architecture. But, drivers uh well, the job only makes sense if we could schedule the drivers plus the

executors as a unit. Let's say we have a short of resources, and we receive an urge of thousands of Spark workloads. We don't want to end up only scheduling or finding space for drivers. Applications only make sense if they are scheduled as a gang as a unit, let's say. So, we use a concept from Volcano, which is actually a pod group, to ensure this gang scheduling. Well,

eventually the driver is running, the driver is writing his event timeline to the local storage, and we are constantly exporting this event timeline to object storage through a fluent deconector, and this would enable an extra feature. So, if the users go to the UI, they can go to a job that has already finished, and we can on demand spin up an history server pod that streamlines all

these events from the object storage to be able to do this replay. Well, we go now to the core layer of the system that actually enables the multi-tenancy, which is the the scheduling layer. And this is where we chose Volcano as our batch scheduling system for Kubernetes. It not only provides the gang scheduling um feature that I just mentioned, it also provides the default Kubernetes primitives uh

that includes priority scheduling, preemption, all of that. But, one of the features we benefit the most from Volcano, it's actually the hierarchical queues feature. We previously have independent clusters for clients and use cases and all of that. With hierarchical queues in YARN, we now are able to migrate to Kubernetes because of this feature from Volcano and connect all these clients, all these use cases in the same

Kubernetes cluster using this same concept. So, on the right side, you can actually see uh the different perspectives, what the user sees versus what the cluster actually has deployed. So, the user is exposed to two different concepts. The user is exposed to uh the first concept is the Spark cluster resource, and this is what I just told you. So, this is the virtual capacity for users' workloads.

And this Spark cluster resource, by the moment they create this, they are telling us one thing. They are telling us you I want to use your service. So, by the moment they do this, we create a respective cluster-wide queue for that exists in Volcano for this specific cluster. And we actually And we also create another cluster-wide queue that serves as the quota for this specific tenant. So,

they can create multiple clusters under their tenant, let's say, namespace, and we can restrict all of them to the same quota that can be reviewed later, let's say, in a different process. And also, the user, to have self-serviceability over their queue strategy, we have developed a namespace resource that we call, let's say, a namespace queue that mimics a cluster-wide queue from Volcano as well. So, by the

moment they create these resources, we make sure we connect all the dots up to the the cluster root queue, let's say. let's try to see how our our service actually looks like at scale. In this slide, you can see the overall picture of what the Spark workload actually means. And um you can quickly see that for a single submission, we have about 15 Kubernetes objects being created.

This involves objects related with workload isolation, gang scheduling, platform features, and a lot of them related with configuration. And I can give you a practical example. So, we have users in our system that typically submit us a payload a JSON payload to our API that's worth 150 kilo kibibytes worth of data. So, this includes uh environment variables, uh secrets, labels, Spark configurations, all of that bundled together

in a JSON. When we receive this uh payload, we create 15 Kubernetes objects that if we were to sum all the manifest size from all this data logically, we would uh get 700 kibibytes worth of manifest size. This is about 4 and 1/2 times more uh Kubernetes objects than the JSON payload that was submitted. And for us to keep this job running, we actually need to keep

all these objects alive, but when the job terminates, we may only keep the core object, which is the Spark application uh resource. We will now see the consequences of accumulating these objects um over time. So, what if we took a look at this from a logical perspective? Today, there's multiple offerings of etcd and managed services which with different sizes. Different sizes imply costs or other constraints might

affect scale. So, doing a few calculations, 1,000 concurrent apps are roughly 15,000 objects. This puts us at the uh near the safety threshold for the recommended etcd size of 2 GB, and we don't only run Spark on our clusters, we have a platter of other workloads running at the same time. For this example, we've considered a total usage of 40% to be a a safe threshold. So,

if we look at 10,000 concurrent applications, this would not be safe even if we account for an etcd with 16 of storage. Regarding terminating jobs, remember that we only need to keep around Spark applications after they end. So, if we have one job submitted every minute with the retention time of 14 minutes, this ends up with 40% usage of an etcd with a total of 8 GB.

Scaling this to five jobs a minute would add up without storage pretty fast. And as an example, some cloud providers gives us managed Kubernetes clusters with only 250 MB of etcd. Doing some benchmarks, using the same 150 KiB job by loader as a reference and trying to submit 3,000 we got after a run of 15 minutes, which includes starting, load injecting, and doing a cleanup after around

30,000 events, which 800 Spark applications at 700 driver pods, thousands of other objects such as secrets, roles, role bindings, config maps, etc. So, in total, we were at the 50,000 55,000 mark on objects. Do note that we started with our etcd here with around 270 MB as a baseline. And our load campaign added roughly 700 MB logically in objects. So, considering the object raw size times uh

the count. But given this starting baseline, we would expect around 1 GB worth of storage after ending our tests. We ended up with almost five times that storage. This is roughly 4.8 GiB of reported storage. This is already over 50% of the total storage we had available for a cluster. Which was 8 GB of capacity. Other notable side effects of this is increased latencies in the API

server. So, mainly list and post calls. But this increased API latency caused abort aborted requests and timeouts which impact other services running in the cluster. Controllers or operators that watch these resources or even webhooks that interact with the calls. X example, Cilium, Kyverno and kube-proxy are affected on this scale. So, a reasonable ask would be where did this storage came from? another interesting fact is after we

ended our tests and start actually cleaning as you can see here, deleting storage went up instead of going down. So, we went from 3.5 GB to 4.8 GB. Which is counterintuitive, right? Even when deleting something, it does not get removed, it's still be there and new objects will even be created in response to us deleting past objects. So, we'll be looking at possible sources. Some are on

the boxes below to what this comes from. One of the main culprits are Kubernetes events. Usually, from our observations, one GB These are just yet another Kubernetes resource, like a pod or config map. They usually live for 1 hour and then are deleted. And not not only the fact that we delete objects and does not free up space, it commonly will cause controllers watching these resources being

deleted to create events in response. So, as you can see, we started our cleanup phase, there was as surge of that then got deleted after an hour. But there's yet to be explained why we have five times increase happened. So, when backing our etcd and those data, it uses a non-locking concurrency control method, the multi-version concurrency control, for short term VCC. This is not exclusive to etcd.

It's also available in PostgreSQL, MySQL, and gives a snapshot isolation. This means that concurrent readers for the same data actually see a snapshot. Instead of having object locks to ensure consistency, readers see different versions in time. So, as everything in life, it comes with trade-offs. One of them is every updated object will actually be a full copy of So, if you consider do a Kubernetes object during

its life cycle, it goes through multiple stages. Update status field from multiple phases. So, deleting an object can create yet another version of it, a tombstone. As an example, on our left, if we consider Spark application with 150 kibibytes, it goes through five phases. We have a create and five updates on its life cycle. So, in total, we have six different versions of this object. So, 150

kibibytes of an object quickly became 1 mebibyte of storage on our etcd. And not only Spark applications are being updated, right? You have submission pods, driver and executor pods, pod groups, etc. So, you can imagine the amount of data that is being generated. We need to keep in mind that these versions will not be kept forever. Uh on Kubernetes, there's usually a process that runs every 5

minutes called compaction, which will remove these outdated revisions also have our Removing a revision does not actually mean that we'll get the space back on the host file system. It only means that logically our etcd can reuse that freed up space to fit the new object. But here comes a new effect, fragmentation. And there's yet another operation that we can use, which is defragmentation, that actually will

reorganize the data on our storage. So we're going to see an amplified example of this. this is an issue that does not affect only etcd, but also other databases as well, file systems, and disks. Here we start in moment one with the very good usage, no fragmentation. Then we remove A1, A2, and C2. Also run compaction to free up the space. logically we came with chunks of

space available, but it's scattered around, right? If we go to moment three, we add a new object, B2. If we sum the space that became free in moment two, we could fit the new object, but the space is not contiguous. So this means that the object is written in the end, in the next available free chunk, and as such the physical size actually grew even more. And

puts us in this case in the dangerous zone. So here only defragging the etcd would help us. It would reorganize the physical space so we could fit B2 in the space that became available previously. From our observations on some cloud providers, this usually happens in the cadence of every two days, or after a certain high threshold between the logical and physical space has been reached. So I

how can we mitigate this, right? we could think about having a data store with hundreds of gigabytes of work worth of that CD, which becomes expensive at scale, or might not even be feasible, right? Well, the truth is that Kubernetes already has something we can use for this. So, let's try to think a little bit beyond etcd. You can think of the API server as like a

composition of layers when it is handling your requests. So, when you do a request to a core object, let's say a pod or a deployment, uh you can think of it as it is being handled by the first layer. But, when you do a request to a CRD, this is being handled by the let's say the extensions layer. The The way you have to extend the API

server, and it gives you for free all the schema management and storage for for these But, if you do a request to an object that is not core and it's not a CRD, then it could be handled by the aggregation layer. And the aggregation layer is a layer where Kubernetes would actually act as a proxy to a custom server that you control in your namespace. And as

such, you receive the requests, you can handle the data related with the request whatever you want. Um but, how can you do this delegation? How can you say that a certain object is actually handled by a custom server that you control? Well, you can use yet another Kubernetes object, which is called the API service. And you actually have uh a lot of API services already in your

cluster. You have API services for core objects, you have API services for your CRDs. And you can create your own API services that point to uh a service, and a namespace, and a port. Well, um actually, a notable example of this is the kube-metrics server. I don't know if you uh probably you already use kube-metrics server in your cluster. And kube-metrics server works like this. So, there's

an API service that controls a certain API that the the Cube Metric Server controls, which the which has different objects under its scope, namely the Pod Metrics object and Node Metrics. So, Cube Metric Server periodically queries Kubelet to get information about pod container memory usage and CPU usage. It bundles this information in a pod metrics object. And these objects never go into it at CD. They actually

get inside an in-memory store of Cube Metric Server. So, let's try to apply this same concept with the Spark operator API group. We'll be doing a quick demo where we will show that we can delegate all the objects from the Spark operator inside a custom server that we control uh which has an in-memory store. So, let me see if I can quickly mirror the screen to show

you the demo. Oh. Yeah. Oh, okay. I don't know if it's easy to see. Um can anyone give me a Okay, thank you. Uh so, we have here a um local cluster and we will be applying a couple manifests in in this cluster. We have an API service a deployment. after we apply these these manifests, we can actually inspect the contents. Let's see the contents of the

API service that we created. And we can see that it points to a service that lives in a name space that we control. Um we can actually get the logs of the service and see that there already some requests coming in uh and we can do a couple requests. Let's list all Spark applications for And we can see that the request reaches the server, the list operation.

Let's now look at an example application. This is a Spark application from Spark operator. You see the drivers configuration, executor configuration. And we can create a couple hundred of After we start this load injection, you'll see that the requests for the create operations show in the logs, the metrics start to grow until we reach 300 applications. And we can now do an interesting thing. Let's since we

have a very reliable system, we can actually restart the our custom server and we will not lose any data, right? So, let's do a count over the Spark applications after the restart. And after a while, you'll see that the we will drop to zero applications. So, this is actually what you get when you rely on an in-memory store. So, yeah. Let's go back to the presentation. Let

me here open So, let's run the same benchmark again as we did previously. 3,000 jobs, 15 minutes. What we got is an increase in 30% on objects created. Well, this shows that we are able to react to more things in the same time frame. Logical size also increased by 20%. This is a consequence of creating 30% more objects, but on the etcd physical size, we actually had

a 45% decrease in etcd usage. And this is the important point that even though we were able to create 30% more objects, 20% bigger in logical the ability to delegate this to our API server made it that we needed 2 GB less of physical storage to handle more All the MVCC updates related to Spark applications were offloaded to our API generating less bloat and fragmentation within that

CD. An important point is to note that we are using an in-memory data store for this purpose. This brought faster data lookups, improving performance of crude requests, but it may differ when using a different data store, which is more realistic scenario for some production cases. But the physical storage improvement will remain. what did we learn, right? CRDs are not free. Size has a cost. So, large objects,

bigger than 10 KB, will multiply over the MVCC revisions that we'll create. Churn. If we do frequent status updates and create and trigger a lot of revisions, this will compound. And retention costs. Historical objects that will stay in So, as design principles, we should keep custom resources small, minimize status updates, and design for deletion either through TTL controllers or finalizers with timeouts. So, when should you use

aggregation layer? So, for cases like in this talk and what we've encountered, this was used to move heavy, high-churn objects out of our etcd. But other purposes are multi-cluster state synchronization. So, if you have an ADL or a blueprint, and remember, retention belongs in a long-term storage, not in an SSD. Thank you.