Right Sizing Spark Executors

This is WIP

“Right-sizing Spark executor memory” details LinkedIn’s large-scale use of Apache Spark and the challenges faced in efficiently tuning Spark executor memory for analytical workloads.

Executor Memory was the single biggest source of configuration changes (52%) and significantly impacted resource efficiency (memory utilization was only 50%) and user productivity.

LinkedIn’s Solution

LinkedIn developed an automated Spark right-sizing system, focusing first on executor memory.

The system:

• Uses a continuous feedback loop: nearline (Kafka → Samza → MySQL DB) for historic memory metrics and real-time signals for immediate OOM failure response.

• Policy-based adjustment: OOM errors trigger a 50% memory increase (Policy P0: Executor OOM Scale Up); otherwise, a heuristic uses a 30-day memory usage history to set memory with conservative buffers for jobs with limited historical data (Policy P1: Heuristic-Based Executor Memory Scale Up/Down).

• Guardrails cap how much memory can scale up or down to maintain stability.

My Thoughts

This is an excellent solution that, with some effort, can significantly improve resource utilization for jobs in a multi-tenant Spark cluster.

Furthermore, once implemented, it can lead to substantial cost savings for projects running Spark on serverless platforms such as EMR Serverless, AWS Glue, Databricks Serverless, and others.

The cost models for these serverless solutions are primarily based on the amount of resources used and the duration of their usage. By right-sizing executors, it’s possible to reduce resource consumption, resulting in notable cost savings.

Although the LinkedIn blog does not provide detailed implementation steps, it does offer lots of details that can be used by a seasoned Spark Developer.

Implementation Idea

Here’s how I am thinking to implement it

The general idea is to get these details for every executor for a Spark Application:

• PeakJVMMemory details, i.e., MAX_HEAP_MEMORY
• ProcessTreeMemory details, i.e., MAX_TOTAL_MEMORY
• Calculate MAX_OVERHEAD_MEMORY

$OverheadMemor{y}_{max}=TotalMemor{y}_{max}-HeapMemor{y}_{max}$

How to get memory details?

All these executor-related details can be fetched from Spark History Server REST APIs (SHS) All the executors details for an application ID can be retrieved from: https://<history-server-baseUrl>/api/v1/applications/<appId>/<attemptId>/allexecutors

This returns a list of all the executors. To retrieve the memory details, we need all of these details present in peakMemoryMetrics.*

Enabling Process Map Metrics

To enable Process Map Metrics to be calculated during batch run for each executor for a Spark Application, it’s required to set these parameters at job level:

• spark.executor.processTreeMetrics.enabled = true

Additionally, these metrics are only available for /proc file systems, i.e., any Linux/Unix-based systems.

Calculating Memory Details from Executor Metrics

Once we get all the peakMemoryMetrics from SHS for EACH executor, memory details can be calculated using

$$\begin{array}{c}HeapMemor{y}_{max}=JVMHeapMemory+JVMOffHeapMemory\\ \\ TotalMemor{y}_{max}=ProcessTreeJVMRSSMemory\end{array}$$

ProcessTree Memory Metrics

ProcessTree Memory RSS Metrics has 3 types of metrics being reported:

• ProcessTreeJVMRSSMemory
• ProcessTreePythonRSSMemory
• ProcessTreeOtherRSSMemory

Initial thinking was: TOTAL_MAX_MEMORY will be sum of these, but looking at the metrics for a particular job, it looks like ProcessTreeJVMRSSMemory include ProcessTreePythonRSSMemory also when there is any Python UDF being used in the Spark Application.

How did I conclude this? If I add both ProcessTreeJVMRSSMemory and ProcessTreePythonRSSMemory this is exceeding the entire Executor Container Memory. Also, ProcessTreeJVMRSSMemory actually includes the entire process and the child process memory. During Python UDF execution, a worker process is initialized by the executor as a child process.

Relation between Heap and RSS Memory in Spark explains more details on how these are related.

Once you have all these details, LinkedIn Blog mentions to calculate the suggested HEAP and OVERHEAD memory using:

$$\begin{array}{c}HeapMemory=HeapMemor{y}_{max}+BUFFER\times \left(\frac{HeapMemor{y}_{max}}{TotalMemor{y}_{max}}\right)\\ \\ OverheadMemory=OverheadMemor{y}_{max}+BUFFER\times \left(\frac{OverheadMemor{y}_{max}}{TotalMemor{y}_{max}}\right)\end{array}$$

Looking at this formula directly, it appears to use raw maxima to suggest updated values for HeapMemory and OverheadMemory. However, applying this approach might lead to slight over-provisioning of executor memory, especially in cases of data skew, where a single executor may experience significantly higher memory usage.

Deciding BUFFER

A good point to start buffer is with 25-30%, when there are no historical runs present for an application. This is to avoid any aggressive downscaling and causing OOM or significantly under provisioning.

Here’s what I have decided to do:

• runs < 8, use 25% buffer
• runs > 8, use 15% buffer

Further in the post there will be a section that explains how to decide on a Good Value after N number of runs for a Spark Application.

Additional Ideas - Discarded for simplicity

After running multiple tests, it’s better to keep it simple. Right sizing calculation will still be close to the accurate calculation.

Previous idea, is discarded for now.

I believe that using percentiles (such as p50, p90, or p95) of maximum memory usage is a more effective method for determining appropriate Heap and Overhead memory. This approach accounts for skewness without resulting in unnecessary over-provisioning of resources.

Updated Calculation based on percentile

• Calculate p50 of Heap and Total memory to calculate Overhead Memory.
• Calculate p90 of Heap and Total memory to calculate Overhead Memory.
• If p90 - p50 gap is small (e.g., < 10-20% of p50 ⇒ (p90 - p50)/p50 ), p90 is a good point to do the further calculation
• if p90 - p50 gap is large (e.g., >20-30% of p50), flag/highlight:
  • Investigate further: Is the skew legitimate (e.g., data skew, uneven partitions)? Can the job be optimized to balance usage?
  • Optionally set a “middle ground”:
  • Take a weighted mean between p50 and p90 (e.g., 70% p50 + 30% p90).
  • Allow the user to select/confirm whether to provision for p90 (safe but costly) or p50 (more efficient, might risk OOM for outliers).
  • Can be configurable.

Calculating Number of Executors

To extend this idea a bit more, identifying correct number of executors is also required, especially in case of dynamic resource allocation for max executors, i.e. spark.dynamicAllocation.maxExecutors

One of the good strategy is to identify how long the executor was idle during it’s entire active duration. This can simply be calculated for each executor using:

$$\begin{array}{c}Executo{r}_{uptime}=Executo{r}_{removeTime}-Executo{r}_{addTime}\\ \\ Executo{r}_{idletime}=(Executo{r}_{uptime}\times Executo{r}_{cores})-totalDuration\\ \\ Executo{r}_{idlepercentage}=\left(\frac{Executo{r}_{idletime}}{Executo{r}_{uptime}}\right)\times 100\end{array}$$

Executor Metrics

• addTime: Timestamp when the executors was added.
• isActive: Boolean value. true if executor was active during application’s lifetime. In this case, removeTime will not be available. ApplicationEndTime
• removeTime: Timestamp when the executor was removed (if removed).
• totalDuration: Elapsed time the JVM spent executing tasks in this executor (in ms).
• totalCores: Number of cores available in this executor.

Once idle percentage of all the executors are calculated, we can:

• Calculate average idle percentage, avg_idle_pct.
• Define a threshold/expected percentage, from my experience 15-20% is a good point to start, target_idle_pct.
• avg_idle_pct > targe_idle_pct, adjust the max executors by calculating adjustment_factor and multiplying with max executors

Sample code snippet:

import numpy as np
 
target_idle_pct = 0.15
avg_idle_pct = np.mean([
		executor_metric["idle_percentage"] 
		for executor_metric in metrics
	])
 
if avg_idle_pct < target_idle_pct:
	adjustment_factor = 1 - (avg_idle_pct / 100.0)
	recommended_max = max(int(max_executors * adjustment_factor), 1)

GuardRails

• Extreme downscaling - percentage cap of downscale not more than 30%
• Recommended Heap and Overhead memory shouldn’t be 0.
  • OnHeapMemory recommendation coming as 0, represents that the executors that were allocated are not used at all.
  • Probably an application that doesn’t really require Spark and can be rewritten without it.
• Start with an initial configuration
• in case of increment because of OOM, make sure an executor doesn’t exceed the Nodes Task Memory defined by YARN

Deciding Last Known Good (LKG) Value

Factors to consider:

• time taken for completion is not increased.
• No spills are introduced.