From Idle to Savings: Building a Global Scheduler for Cost‑Efficient Data P... Rainie Li & Ang Zhang

In this talk, Rini, a senior engineering manager at Pinterest, discusses the development of a cost-efficient data processing platform built on Kubernetes. She begins by providing an overview of Pinterest's data infrastructure, highlighting the integration of various technologies such as Apache Kafka, Flink, and Spark for real-time ingestion and batch processing. Rini describes the Mocha platform, an innovative solution that manages workloads using Kubernetes to enhance resource utilization and minimize costs. She details the challenges faced in resource management, including ensuring high availability for critical jobs and making use of reserved capacities during off-peak times. The talk further emphasizes the use of customized Kubernetes schedulers like Unicorn and Volcano to improve scheduling efficiency and job prioritization. Rini concludes by outlining the ongoing efforts to optimize capacity utilization through intelligent scheduling and resource management techniques.

Hello everyone. Welcome to the talk. And today I will talk about how we build cost efficient data processing platform on Kubernetes at Pinterest. Unfortunately, my co-presenter cannot join today, but I will cover his part. I'm Rini. I'm currently a senior engineering manager at Pinterest and managing data processing for team and the big data storage platform team. Here is today's agenda. First, I will talk about Pinterest data

infra high-level overview. Then I will talk about the key challenges of the resource management. And then I will talk about how we utilize Kubernetes to manage in a cost efficient way and how we utilize unused reserved capacity for batch processing applications. I will leave some time at the end for Q&A. Before I dive into the data infra high-level overview, I would like to quickly introduce Pinterest. We

are a visual discovery platform. So user like pinners can select pins and add to their board. Eventually, they can turn these inspirations and ideas to the real life. So we are data-driven company. We have a very large-scale data needs to ingest and process. From this diagram, you can see on the top left side, these are the different data producers. Users can go to the app, browser, and

even third-party data. These data will come into our Pinterest API services. On each host, we have a logging agent. And this logging agent will constantly inject this data into our Pub/Sub service, which we build on top of Apache Kafka and the MemQ. This Pub/Sub service will constantly inject these events into topics. There's a real-time ingestion service written in uh Flink streaming application. It would do real-time ingesting

from this Kafka topic to our data lake. We build our data lake on top of AWS S3. We store all the raw data on S3. We also adopted Apache Iceberg as a metadata layer on top of the original data so that we can provide a standardized table format for the users. It also provides the uh quickly indexing switch between uh different snapshots. So, right now we basically

ingest all the data from users. Let's go to the right side. The bottom part of the right side is our data processing like compute platform. We build everything on top of Kubernetes. We use EKS. We also adopted Apache Iceberg as a remote shuffling to decouple the compute nodes and the storage nodes. On top of that, we have the Flink as a streaming engine and the Spark at

the batch engine. We support different use cases like indexing, recommendations, signals, uh data warehouse. Uh on top right side is our uh like a query book notebook dashboard which interact with um data scientists and uh business intelligence users. They can run these queries uh on top of the query engines. We use Presto and Trino and Spark SQL to query the data um from the Now you have

like a overview of the uh whole data infra. Let me zoom in the uh bottom right part which is a data processing uh area. This is our new generation of data processing platform. We call this Moka. Everything is built on top of Kubernetes. The Moka platform itself uh contains two layers. Uh let me start from the top layer which is our customer teams. They own these data

engines like Spark, Flink, and the ML platform. they own the Ray and PyTorch. And the Moka platform top layer is the actual which is our job submission service on top of uh Kubernetes. It supports the job submission, job routing, uh logging access from the S3 data, and uh fine-grained access control. Along with this, we support the quota enforcement and the resource management. I will cover more later.

And we also support the remote shuffling. On the bottom is the Kubernetes layer internally, but the top layer inside the Kubernetes is also owned by Moka platform. Uh we adopted uh multiple uh customized scheduler like Unicorn, Volcano, and we also support the logging. Uh we use the Fluent Bit. Uh we also adopted uh multiple uh operators like uh Spark, Flink, and the KubeRay. The other layers inside

the Kubernetes internal is owned by uh infra team uh from Pinterest. Uh next, let me briefly introduce our scale. We run thousands of scheduled uh batch workloads in production, and we process uh hundreds of petabytes data daily. There are several key use cases like the AI/ML pipelines. They do content discovery, uh trust and safety, and ads related. And we also support the data warehouse and the uh

business intelligence use case. They generate the most critical metrics like uh MAU, and they run experimentations uh of the data from the data lake. We also support the feature store for the model development so that uh the training team can use those data uh uh Uh next, let me talk about some key challenges to manage this large-scale Spark itself is used about 1/3 of the entire Pinterest

computer resource. So, it it cost a lot. And um the second challenge is we want to make sure the high priority business critical jobs can always meet SLO, but sometimes they cannot get enough resource if the cluster is very busy and we run everything on a shared cluster. The third challenging is during off-peak hours the cluster utilization is not that high. So potentially we wasted some of

the resources. And the last challenging is um uh we have some unused reserved capacity from online services. Uh they are wasted and they are not easy to be used by Spark applications. Next I will talk about how we address these challenges. Here is some uh background of how uh Mocha platform do resource management. From the top left side is uh example of fixed-size cluster. On this shared

environment we run multiple tier And the blue area is a tier one application usage. If it needs more resource, it will be able to preempt a tier two and tier three so that we prioritize and guarantee the most critical applications. If the tier one application usage drops, then the tier two and tier three can take advantage. And the bottom diagram is how we set guarantee value for

each tier queue. Uh we have a data pipeline to constantly query historical usage of each applications like the V core and the memory usage. For tier one, the guarantee value is equal to the historical usage so that we can avoid too many preemptions because preemption happens uh when it cannot get enough resource. But for the tier two, uh the guarantee value we calculate by multiple ratio like

80% of the historical usage. And tier three basically the guarantee value is a best effort of of cluster. Uh the right side is our org-based queue structure. It's a simple um queue structure from one of the cluster. You can see we have three layers. The parent layer is the org layer. We have uh like uh five key orgs at Pinterest. For example, the monetization org. And under

each org, we have multiple projects uh like ads uh ML infra, ads indexing. These projects belong to the And the child queue, which is the last layer, we have like three uh tier queue for each project. And we configure the uh guarantee value under uh each queue. Uh this way we can share the resource within the same cluster, but still create some isolation between each uh project

and each org. And tier one applications will always be submitted to the tier one queue, and they will be able to preempt the tier two and tier three. Uh next I will talk about uh uh the Kubernetes scheduler we use at Pinterest. We adopted a customized scheduler instead of use a default Kubernetes scheduler. The benefits including it provides efficient resource utilization. They can borrow resource between the

queues. And this customized scheduler provide advanced scheduling features like a fair sharing, preemption, gang scheduling. For Spark, we adopted Apache Unicorn, which is a easy transition from YARN scheduler we heavily used on the Hadoop uh environment. And it provides good UI and API. It also provides some uh full queue capacity feature like max application, which we can limit the maximum running application within the each queue to

avoid too many applications compete the resource and eventually cause job failure. Unicorn also provides a app summary feature, which we constantly uh query the historical usage and to generate a guarantee value for each queue in previous slides. Uh we also have a chargeback team, which will use this app summary to uh calculate the resource usage into dollar amount and a charge each org. For PyTorch and training

use cases and Ray, we adopted a Volcano scheduler because it provides like two interesting features. One is a topology awareness scheduling. We use this feature to enable the AZ awareness scheduling because right now GPU resource is very hard to get. So, we spin up a Kubernetes cluster across multiple AZs. And we want to make the scheduler and be AZ awareness when schedule these PyTorch jobs. The second

feature is the co-location feature. Later on, we would like to run uh both training jobs and the serving jobs in a shared Kubernetes cluster. So, the serving jobs always get the highest priority and the guarantee SLO from that setup. Uh next challenge is regarding how we At Pinterest, we reserve capacity from AWS. The There are two benefits. First one is we have the guaranteed capacity for the

most critical online services either in cloud or on prime. And we can also get a good deal from AWS with a long-term commitment. It help us to save a lot of cost. But the problem here is you can see the diagram on the right side. If the online service traffic drops during the off-peak hours, we have a lot of wasted And this is opportunity for Mocha We

can borrow this unused capacity to run some of non-critical Spark applications, which has more flexible schedule. There are two options to implement this Option one is to run everything on the shared community cluster for both online and offline. The good side is it can have the faster preemption and smoother transaction, but the problem here is our Spark application is in very large scale. We have like high

portal numbers. We have a lot of scalability limitations on the EKS control plane and API server side. So, it's very hard to run everything on a shared large cluster. And another problem is with this approach we are lack of the isolation like the network and disk space IO isolation. The second option is how we used at Pinterest is to rebalancing the EC2 capacity from online services to

the offline applications. The good side is everything can work with the current setup and we have a fully isolated environment between online and offline by using two separate clusters. But the downside is if the online services needs a capacity, we have to return back ASAP. So, let me introduce a project ties the traffic utilization. How we implement a customized controller to move capacity from online services during

the off-peak hour to the Spark clusters and we will return back by controlling the auto scaling groups in ties cluster. The diagram on the right bottom side there are two uh graph. The top one is the traffic of the online services. You can see if the traffic drops, the ties cluster will be scaled up and if the traffic increase, the ties cluster will be scaled down. This

is a high-level architecture of the ties system. Let me introduce from bottom up. The ties controller is a customized controller we we wrote to constantly monitor the ODCR capacity metrics and if the online traffic jobs. It will scale up the TiDB cluster. Right now, we have three TiDB cluster in each availability in AWS account. And then if if it sees the online traffic increase, it will scale

down the cluster and it will return back the resource. And the orange box is our global scheduler service. It constantly query a database, which is our job submission service database, to how much jobs uh needs to be scheduled to the TiDB cluster. And it constantly monitors the TiDB cluster capacity. If the cluster is tight, there's not enough capacity, global scheduler will be able to do a global

holding mechanism to hold these And if the TiDB cluster has available capacity, the global scheduler will route those non-critical like tier two and tier three application to the TiDB cluster and route the most critical tier one applications to regular static Mocha There are two small components we implemented dynamic queue template on each TiDB cluster. We also integrated job submission service with a quota in front of it

to make sure we can control the budget for each org. Next, let me talk about how Mocha platform enables TiDB. The left side is our old platform built everything on top of Hadoop. We we store the shuffle data locally. So, if we scale down one of the nodes, job will fail and the shuffle data will be deleted as well. Uh for the Mocha platform, we decouple the

compute and the storage nodes. We have a remote shuffling cluster to store all the shuffle data. So, when TiDB controller scale down the compute nodes, jobs will not fail. They can still read the shuffle data from the remote Other benefits of Mocha that enables tides including we have faster node setup. Compare with the Hadoop, we have to set up the Hadoop demons. And we also provide fine

grain access control to have the unified cluster versus in the old platform we separate the cluster by environment like a PI F guy cluster. However, we have a service called credential vending so that we can have unified cluster setup. The third benefit is um Spark applications are able to use those instance from online services and only by attaching limited EBS volumes because we store all the shuffle

storage in the remote clusters. But there are still several challenges when setting up this tides cluster. First one is how can we guarantee the job success rate because we scale down the cluster and job could potentially The second challenge is how can we balance the disk size and IO throughput because we are reusing the online services instance types. And the third challenging is how can we prioritize

jobs and change the ordering when the capacity become available again. And the fourth one is how can we fully utilize the tides cluster. The last one is how can we better predict the available capacity and do more smart scheduling. Next I will talk about our solution for each challenge one by one. Adding some background information here. Spark driver is like a single point failure for Spark jobs.

If a driver pod got terminated, the entire job will fail. And so, when setting up the Taichi cluster, we want to make sure uh job will still be high available and reliable on the platform. Uh when scale down the capacity and then return back to the online services, the controller will only be scale down the executor pods. Uh so, we set up two separate node groups. Uh

the static node group we set up for driver pods only, and that node group will never be scale uh scale down. And the separate uh the second node group uh we set up for the dynamic changing managed by the controller. Um this way, when we scale down the Taichi cluster, driver pod will always be uh running and job will not And we constantly uh scale up the

uh driver static node groups if we have more jobs onboarding to the Taichi The second challenge is uh disk size and IO throughput. Because the online service instance types, they use uh a block storage, and they use uh a lot of EBS volume. However, the Spark cluster, we use instance storage to handle some the local spill data, because the uh heavy uh uh remote shuffle data uh

handled by the remote shuffling services. And the EBS volume can be very expensive. If we keep uh attaching the EBS volume to this instance from online service, eventually we cannot save the cost. So, our approach here is we dry run some of the Spark applications on these new instance. And to see what is a uh good balance between uh the disk size and IO by attaching the

limited EBS volume, so that jobs can still complete, and the job performance looks good, and we can still uh uh control the cost. Uh the next challenge is uh regarding the job prioritizing and ordering. this is not the uh one cluster scheduling. This is like the global view of all the Uh when tight the capacity becomes very uh we could uh ended up by too many jobs

uh competing the resource. And if each job get only one pod running, they will cause the starvation issue and lead to the job failure. So, our solution here is like I mentioned before, there's a global scheduler service. It can use a global holding mechanism. If it detects tight cluster doesn't have enough capacity, it just hold those jobs. And it can also hold jobs if they are unlikely

to finish before the capacity got scaled down. On the next day, when the tight capacity becomes available, the global scheduler will prioritize these jobs. We have a weight score to calculate how much time each job will wait, and we also will consider the job tier information, SLA information. This way we basically change the job ordering instead of submitting them in the FIFO order. Uh the next challenging

is um how to fully utilize this tight capacity. Initially, we worked with customer teams like the uh users and asked them to opt in. Basically, they provide a flag to the platform that tight equals to true. Then we will route these jobs to the tight cluster. However, this approach didn't work because user doesn't have the strong motivations to onboard to tight cluster. But from the platform's perspective,

we still want to save the infra cost. What we did here is we design and implemented an automatically routing mechanism within the platform. So, it can automatically select jobs and to those non-critical jobs to Taiz cluster based on their priority, uh historical resource usage, and uh waiting time. Uh if the regular static and Moka cluster becomes very busy, uh the global scheduler will also select a certain

like non-critical jobs and route them to Taiz. And if Taiz capacity goes away, uh we can also smart route back those jobs to regular Moka clusters. This way we are able to uh reach Taiz cluster utilization to uh 70% so far. Another challenging here is um which we are still work in progress. Uh right now the um controller in Taiz is um scale up and down the

cluster based on a policy. For example, uh Pinterest online service traffic jobs every day starting from 12:00 a.m. to 6:00 a.m. uh like PST time. So, the Taiz controller uh scale up the cluster uh within this fixed time window. But the challenging here is um there are still some unused capacity, and they can change over time. For example, we have more available capacity during holiday seasons. And

how can we uh even fully utilize all the available capacity? We are thinking about to use ML model to predict what are the coming available capacity uh And for the scheduler side, we can also um calculate smartly by uh by seeing this online uh available capacity and calculate the historical usage and do the best matching for the uh jobs we want to schedule and the available capacity.

Uh this way I think we can uh even push in the utilization. Hopefully, we can reach out to 90% with this approach. Mm, yeah, that's uh all the contents. Uh any questions? Hm.