Why Spark Executors Leave Cores Idle on YARN: A Packing Model for Executor Sizing

The Problem

A batch Spark job on YARN can show memory nearly full and CPU half idle at the same time, and you pay for both. Memory looks full because YARN reports the memory allocated to containers, not the memory in use. CPU looks idle because the executors that fit on each node don't add up to the node's cores. The cluster is billed for every vCPU and uses a fraction of them.

Background: How Spark Sizes Itself on YARN

Spark requests one YARN container per executor [3]. The container size is the executor heap (spark.executor.memory) plus off-heap overhead (spark.executor.memoryOverhead), which defaults to 10% of executor memory with a 384 MiB minimum [1]. The heap divides into a small reserved area, a unified region shared by execution and storage (60% of the usable heap by default, set by spark.memory.fraction), and the remainder for user data structures [2]. YARN also rounds each request up to its allocation increment, so real fits can differ slightly from paper math [4].

Two limits decide how many executors land on a node: the vCPUs and the memory that the NodeManager advertises. Whichever runs out first sets the count.

executors_per_node = min(
    floor(node_vcores / cores_per_executor),
    floor(node_memory_mb / ((executor_memory_gb + overhead_gb) * 1024))
)

The Packing Model in Practice

Take a worker that exposes 16 vCPUs and 112 GB (114,688 MB) to YARN, running executors with 5 cores each. By cores alone, the node fits three executors. Memory decides whether it actually does.

The first two rows are the trap. A large executor looks generous, yet it strands cores that you still pay for. The third row is the sweet spot: the core limit and the memory limit meet. The last row shows that shrinking memory further gains nothing, because the core limit now binds.

The overhead setting changes the packing too. With the default 10% factor, a 48 GB executor requests 4.8 GB of overhead. Setting it explicitly to a smaller value can raise the number of executors per node, and setting it too low can get containers killed for exceeding their memory limit.

The misleading dashboard follows from the same model. YARN's memory-used metric counts allocated container memory. A node full of reservations looks 100% used even when the executors touch only part of their heap. High memory use plus low CPU use is therefore a signature of oversized executors, not a contradiction.

The Approach

The fix is to shrink the executor until the core limit binds instead of the memory limit. Keep the cores per executor fixed so that memory is the only variable. Five cores is the common starting point, a heuristic that grew out of HDFS client throughput [7][8]. Leave one or two vCPUs per node free for the NodeManager and the operating system.

Then set the shuffle partition count. The default of 200 leaves a large cluster mostly idle during shuffles. A common rule is 2 to 3 times the total core count, checked against a partition size near 128 MB. When adaptive query execution is on, which is the default from Spark 3.2, Spark can coalesce small shuffle partitions at runtime, and the configured number sets the ceiling [6].

spark.executor.cores                5
spark.executor.memory               32g
spark.executor.memoryOverhead       1g
spark.sql.shuffle.partitions        4800     # about 3x total cores (1,600 in this example)
spark.dynamicAllocation.enabled     true
spark.shuffle.service.enabled       true
spark.speculation                   true
spark.speculation.multiplier        3
spark.speculation.quantile          0.9
spark.executor.extraJavaOptions     -XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35

A few of these deserve a comment. Executor autoscaling on YARN is usually paired with the external shuffle service so that shuffle files outlive their executors. Speculation relaunches tasks that run more than three times slower than the median once 90% of a stage's tasks have finished, which trims straggler tails at the price of some duplicate work. The G1 occupancy threshold of 35 starts concurrent marking earlier than the default of 45, trading some CPU for a lower risk of full garbage collections on large heaps.

The Failure Mode: Local Disk

Fixing CPU utilization can expose a limit you never reached before. More active cores per node means more concurrent shuffle writers on the same local disks. When a disk passes the NodeManager health checker's utilization threshold, which defaults to 90% and applies to both the local and log directories, YARN marks the node unhealthy and stops scheduling on it [4][5]. In a bad case, a large share of the cluster drops out mid-job, and the run has to be killed.

The remedies are mechanical. Give workers larger local disks, size shuffle partitions so each one spills less, and watch the unhealthy-node count during any resizing experiment. Run the experiment at the quietest and busiest hours of the day, because the busy hour is where disk pressure shows up.

Query-Level Tuning Still Matters

Cluster sizing only goes so far. Find data skew first, since it is the most common cause of stuck stages. Prefer broadcast joins when one side is small enough to fit in executor memory, use equi-joins where possible, and always filter on the partition key. Look for joins between two very large datasets that a pre-aggregation could replace. Removing one such join can shrink the cluster a job needs far more than any configuration change.

Trade-Offs

Smaller executors fit more tasks per node, but each task gets less memory. Skewed tasks are more likely to hit out-of-memory errors, and broadcast variables are copied once per executor.

Larger executors share broadcast data across more cores and carry less per-JVM overhead, but they can strand cores and tend to have longer garbage collection pauses.

More active cores per node means more concurrent shuffle writes, so size local disks for the peak and not the average.

The five-core rule is a heuristic. Test it against your own storage layer, whether that is HDFS or an object store.

A Sizing Routine You Can Reuse

Read the NodeManager memory and vCore limits from the ResourceManager UI. Fix the cores per executor. For each candidate memory size, compute executors per node with the function below. Pick the largest memory size that still reaches the core limit. Set shuffle partitions to 2 to 3 times the total cores, then check the partition size. Test at a quiet hour and at the peak, and watch NodeManager disk health.

def executors_per_node(node_vcores, node_mem_mb, cores_per_exec, exec_mem_gb, overhead_gb):
    by_cpu = node_vcores // cores_per_exec
    by_mem = node_mem_mb // int((exec_mem_gb + overhead_gb) * 1024)
    return min(by_cpu, by_mem)

for mem, ovh in [(64, 2), (48, 2), (32, 1), (24, 1)]:
    n = executors_per_node(16, 114688, 5, mem, ovh)
    print(mem, ovh, n, n * 5)   # executors per node, busy vCPUs

The Result

Applied to a production batch job, this model explained a large gap between the cores the cluster paid for and the cores it used. Resizing the executors moved the busy cores close to the node's usable limit and shortened the run. The same job later needed fewer workers. The practical rule is simple: count the containers that fit on a node before changing any setting.

Vikrant Bachhav is a Staff Software Engineer specializing in large-scale data infrastructure, real-time and batch ML feature pipelines, and cloud cost optimization.

References

[1] Apache Spark, Configuration

[2] Apache Spark, Tuning Spark

[3] Apache Spark, Running Spark on YARN

[4] Apache Hadoop, yarn-default.xml

[5] Amazon EMR, Cluster terminates with NO_SLAVE_LEFT and core nodes FAILED_BY_MASTER

[6] Databricks, Adaptive Query Execution: Speeding Up Spark SQL at Runtime

[7] Distribution of Executors, Cores and Memory for a Spark Application

[8] PayPal Tech Blog, Tuning Spark Applications to Efficiently Utilize Dataproc Cluster