How Does Apache Spark Manage Executor Memory?

Memory management is a critical aspect of Apache Spark, as it directly impacts performance, job execution, and troubleshooting. Understanding the various types of memory in Spark and how they are managed allows for better job tuning, optimization, and issue resolution.

In this article, we will dive into Spark Executor memory, exploring its components and management through a practical example.

Spark Executor Memory

Before we dive in, let’s quickly recap Spark’s architecture. Spark operates with two main types of nodes:

Master (Driver): The central coordinator responsible for task scheduling and cluster management.
Worker Nodes: Machines that execute tasks, each containing one or more executors that process data.

In this article, we will focus on Executor Memory. Spark Executors primarily manage two key types of memory:

Off-heap memory

It was introduced in Spark version 1.6. In this mode, memory is not allocated within the Java Virtual Machine (JVM) heap; instead, it uses Java’s unsafe API to directly request memory from the operating system. This allows Spark to access off-heap memory directly, reducing unnecessary memory overhead, minimizing frequent garbage collection scans and collections, and ultimately improving processing performance.

To utilize this memory, it should first be enabled by setting the parameter spark.memory.offHeap.enabled to true and then providing its size using the parameter spark.memory.offHeap.size. While off-heap memory offers advantages in terms of performance and management, it requires developers to handle memory allocation and release logic explicitly, as opposed to the automatic management provided by the JVM heap.

On-heap memory

It refers to the portion of memory allocated within the Java Virtual Machine (JVM) heap space. The JVM heap is the memory area where Java objects are created and managed during program execution. On-heap memory in Spark is used to store various data structures, objects, and temporary data generated during the execution of tasks. The parameter spark.executor.memory specify its size.

The on-heap memory is managed by the Java garbage collector, which automatically reclaims memory occupied by objects that are no longer in use.

Executor JVM Heap Memory (On-Heap)

The on-heap memory is divided into three memories as follows:

Reserved Memory: It’s used to store Spark internal objects the size is hardcoded and it’s equal to 300 MB.
User Memory: Used to store data required for RDD transformation operations, such as RDD dependencies.
Unified Memory: It includes two types of memory:

Execution Memory: It stores temporary data during calculations like Shuffle, Join, Sort, and Aggregation.
Storage Memory: Mainly stores Spark cache data such as RDD caching, unroll data, and broadcast data.

Executor JVM Heap Memory. Source: Author

How Spark Compute Memories?

To demonstrate how Spark allocates the memories described above, let’s consider an executor with a heap size of 10 GB, which means spark.executor.memory=10GB (or –executor-memory 10GB).

Step 1: Set Reserved Memory

The first step is to allocate Reserved Memory, and only once this memory is allocated does Spark start allocating others. Let’s refer to the remaining size as Usable Memory, and the formula used to compute it is as follows:

UsableMemroy = (Heap size - ReservedMemory)
           = 10 GB - 300 MB
           = 9.71 GB

Step 2: Allocate User Memory

To allocate User Memory, Spark considers the parameter spark.memory.fraction, with the default value of 0.6. The formula used to compute this memory is:

UserMemory = UsableMemory * ( 1 - spark.memory.fraction)
           = 9.71 * (1 - 0.6)
           = 3.85 GB

The User Memory that we’ll be allocated is 3.85 GB which is 40% of the Usable Memory.

Step 3: Allocate Execution and Storage Memories

To allocate these memories, in addition to the parameter spark.memory.fraction, Spark will also consider the value of spark.memory.storageFraction, which by default is equal to 0.5. The formula that will be used to compute the memory, keeping the default values of both parameters (0.6 and 0.5), is:

ExecutionMemory = UsableMemory * spark.memory.fraction * ( 1 - spark.memory.storageFraction)
           = 9.71 * 0.6 (1 - 0.5)
           = 2.913 GB

The same formula is applied to compute the Storage Memory. So the value of both will be equal to 2.913 GB, around 30% of the Usable Memory.

Note: Eviction or overlapping is possible between Execution Memory and Storage Memory. This means that when no Execution Memory is used, Storage can acquire all the available memory, and vice versa. Execution may evict Storage if necessary, but only until the total Storage memory usage falls under a certain threshold.

References: