Scaling Distributed Compute Engines on Kubernetes and YARN

Modern data engineering involves a constant struggle between resource availability and infrastructure cost. In a static allocation model, a cluster is provisioned with a fixed number of executors regardless of the actual computational demand at any given moment. This leads to a scenario where resources sit idle during low-traffic periods yet become a bottleneck during peak processing windows.

Dynamic resource allocation solves this by allowing the processing framework to request and release resources based on the workload requirements of the current stage. If a job is performing a simple map operation that requires minimal overhead, the system can scale down to a few executors. Conversely, during a massive join or shuffle operation, the system can aggressively request more containers from the cluster manager.

The primary motivation for this elasticity is not just cost savings but also multi-tenant stability. When dozens of different jobs share the same physical cluster, one inefficiently sized job can starve the entire ecosystem of resources. By implementing dynamic scaling, we ensure that every application consumes only what it needs to make progress.

1// Define the configuration for a production Spark session
2val sparkConf = new SparkConf()
3  .setAppName("ElasticDataProcessor")
4  // Enable the dynamic allocation feature
5  .set("spark.dynamicAllocation.enabled", "true")
6  // Set the minimum and maximum number of executors
7  .set("spark.dynamicAllocation.minExecutors", "2")
8  .set("spark.dynamicAllocation.maxExecutors", "100")
9  // Require an external shuffle service to prevent data loss
10  .set("spark.shuffle.service.enabled", "true")
11  // Adjust how quickly the system requests new executors
12  .set("spark.dynamicAllocation.schedulerBacklogTimeout", "1s")

The transition to dynamic scaling requires a shift in how we think about job reliability. Because executors can appear and disappear at any time, the data lineage must be strictly managed to ensure that a lost executor does not result in a complete job failure. This is where the external shuffle service or a remote shuffle storage layer becomes an essential architectural component.

Selecting the right size for an executor is one of the most impactful decisions in distributed system design. An executor is essentially a single process running on a worker node that manages multiple threads or cores. If you create executors that are too small, you increase the management overhead and lose the benefits of shared memory within the process.

If you create executors that are too large, such as assigning 32 cores and 128 gigabytes of memory to a single process, you often run into severe garbage collection issues. Large memory heaps lead to longer pause times when the system tries to reclaim memory, which can cause the driver to think the executor has become unresponsive. This leads to unnecessary task retries and instability.

Research and practical experience suggest that the sweet spot for most workloads is around five cores per executor. This provides enough parallelism to share resources effectively while keeping the memory heap small enough for efficient garbage collection. It also aligns well with the throughput capabilities of modern network interfaces and storage systems.

• Thin Executors: One core per executor. Provides isolation but creates massive overhead for metadata and shared variables.
• Fat Executors: Many cores per executor. Shares memory efficiently but suffers from long garbage collection pauses and potential I/O bottlenecks.
• Balanced Executors: Approximately five cores per executor. Maximizes HDFS throughput and keeps garbage collection manageable for most datasets.

In a Kubernetes environment, executors are deployed as individual pods that are managed by a custom operator or the processing framework itself. This allows for excellent resource isolation and the ability to leverage existing DevOps tools for monitoring and logging. However, the ephemeral nature of pods creates a significant challenge for the shuffle phase of data processing.

During a shuffle, executors write intermediate data to their local disks so that other executors can fetch it later. If the dynamic allocator decides to kill an executor to save resources, the intermediate shuffle data stored on that pod is lost. This forces the framework to recompute all the tasks that produced that data, which can drastically increase job duration.

The greatest hidden cost in elastic distributed systems is the recomputation of lost shuffle files. Without a persistent or external shuffle storage mechanism, dynamic allocation can often hurt performance more than it helps.

To mitigate this, many modern architectures use a remote shuffle service or persistent volume claims that exist independently of the executor lifecycle. This allows the system to scale down the compute power while keeping the intermediate data accessible to the remaining nodes. It effectively decouples storage from compute, which is a hallmark of scalable cloud-native design.

Data skew is the primary enemy of efficient resource allocation. If one partition of data is ten times larger than the others, a single executor will be stuck processing it while all other executors finish their tasks and go idle. In this scenario, dynamic allocation might scale down the cluster, but the job still takes a long time to complete.

Adaptive query execution can help solve this by monitoring task statistics at runtime. If the system detects a skewed partition, it can automatically split that partition into smaller chunks and distribute them across more executors. This works in tandem with dynamic allocation to ensure that the increased resource count is actually utilized to solve the bottleneck.

Another critical strategy is the use of speculative execution. If a few tasks are running significantly slower than the average due to faulty hardware or network congestion, the driver can launch duplicate versions of those tasks on different executors. The system then accepts the result from whichever task finishes first and kills the other one.

1import requests
2import time
3
4def check_executor_health(driver_url):
5    # Query the internal status API for executor metrics
6    response = requests.get(f"{driver_url}/api/v1/applications/app_id/executors")
7    executors = response.json()
8    
9    for executor in executors:
10        # Identify executors with high garbage collection time
11        gc_time = executor.get("totalGCTime", 0)
12        task_time = executor.get("totalDuration", 1)
13        
14        if (gc_time / task_time) > 0.1:
15            print(f"Warning: Executor {executor['id']} is spending 10% of time in GC")
16
17# Run the health check loop
18while True:
19    check_executor_health("http://spark-driver:4040")
20    time.sleep(60)

Finally, we must consider the grace period for executor decommissioning. Instead of killing an executor immediately, the system can mark it as decommissioned. This allows the executor to finish its current tasks and migrate its shuffle data to another node before it is completely shut down. This minimizes the risk of job failures and ensures a smooth transition during scaling events.