Back to Apache Spark: Core
Q28Question 28 of 72

Spark's unified memory model

Key Concepts
  • Executor heap is split into a unified region (spark.memory.fraction, ~60%) and user memory (~40%).
  • The unified region is dynamically shared between Execution Memory (shuffles, joins, sorts) and Storage Memory (cached data).
  • spark.memory.storageFraction (~50% of unified) sets the protected baseline below which storage is safe from eviction.
  • Execution can evict borrowed storage blocks when it needs space; storage cannot evict active execution memory.
  • Over-caching steals memory from execution, causing spills; the symptom is slow shuffles, not a cache error.
  • This unified model replaced the old static split where idle memory in one pool could not help the other.

Definition

Execution and Storage share one unified region with a movable boundary; either can borrow the other's idle space, but execution has priority and can evict borrowed cached blocks, while active execution memory is never evicted.
  • Unified region – controlled by spark.memory.fraction (~60% of usable heap), the rest is user memory for non-Spark objects.
  • Execution Memory – holds transient working data for shuffles, joins, sorts, and aggregations.
  • Storage Memory – holds cached and persisted partitions.
  • Dynamic boundary – spark.memory.storageFraction (~50% of unified) sets the protected baseline; above it, storage can be evicted by execution.
  • Priority rule – execution can evict borrowed storage blocks; storage cannot evict active execution memory.

At the National Bank of India imagine one large back room shared by two teams: the processing team (execution) doing the heavy work of sorting and matching slips, and the filing team (storage) keeping frequently-used ledgers handy on shelves. There is a soft line down the middle, but it moves: if the filing team has spare shelf space, the processing team can spread their work onto it, and vice versa. The rule is that processing work has priority: when the processing team suddenly needs the shelves back for an urgent sort, the filing team's borrowed ledgers get cleared off, because losing a cached ledger only means fetching it again.

Understanding the Concepts

What is the unified memory model?

Spark's unified memory model divides the executor heap into regions that cooperate rather than compete. The key insight is that execution and storage share one pool with a movable boundary, so idle memory in one region is available to the other.

PYTHON
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("NBI-Memory-Model") \
    .config("spark.executor.memory", "4g") \
    .config("spark.memory.fraction", "0.6") \
    .config("spark.memory.storageFraction", "0.5") \
    .getOrCreate()

# Memory layout for 4 GB executor:
# Reserved:       ~300 MB (Spark internal objects)
# Usable heap:    ~3.7 GB
# Unified region: 3.7 * 0.6 = ~2.2 GB
#   Execution:    2.2 * 0.5 = ~1.1 GB (baseline)
#   Storage:      2.2 * 0.5 = ~1.1 GB (baseline)
# User memory:    3.7 * 0.4 = ~1.5 GB

print("memory.fraction:", spark.conf.get("spark.memory.fraction"))
print("storageFraction:", spark.conf.get("spark.memory.storageFraction"))
TEXT
memory.fraction: 0.6
storageFraction: 0.5

How is executor memory laid out?

The executor heap has three layers. Reserved memory is for Spark's internal objects. The unified region is managed by Spark for both execution and storage. User memory is for your own data structures and non-Spark objects.

PYTHON
# Memory layout calculation
executor_memory_gb = 4.0
reserved_mb = 300
usable_gb = executor_memory_gb - (reserved_mb / 1024)

memory_fraction = 0.6
storage_fraction = 0.5

unified_gb = usable_gb * memory_fraction
execution_baseline_gb = unified_gb * (1 - storage_fraction)
storage_baseline_gb = unified_gb * storage_fraction
user_memory_gb = usable_gb * (1 - memory_fraction)

print(f"Usable heap:         {usable_gb:.2f} GB")
print(f"Unified region:      {unified_gb:.2f} GB")
print(f"  Execution baseline:{execution_baseline_gb:.2f} GB")
print(f"  Storage baseline:  {storage_baseline_gb:.2f} GB")
print(f"User memory:         {user_memory_gb:.2f} GB")
TEXT
Usable heap:         3.71 GB
Unified region:      2.22 GB
  Execution baseline:1.11 GB
  Storage baseline:  1.11 GB
User memory:         1.48 GB

What is Execution Memory used for?

Execution Memory holds the transient working set of wide operations: the buffers for shuffle writes and reads, the hash tables for hash joins and hash aggregations, and the sort buffers for sort-merge joins and order-by operations. This memory is actively in use during a task and cannot be evicted.

PYTHON
import pyspark.sql.functions as F

txn = spark.read.parquet("hdfs:///bank/transactions/2026-01/")

# These operations use Execution Memory:
# 1. Shuffle for groupBy
totals = txn.groupBy("branch").agg(F.sum("amount"))  # hash aggregate buffers

# 2. Sort for orderBy
sorted_txn = txn.orderBy("amount")  # sort buffers

# 3. Hash table for join
accounts = spark.read.parquet("hdfs:///bank/accounts/")
joined = txn.join(accounts, "account_id")  # hash join buffers

What is Storage Memory used for?

Storage Memory holds cached and persisted partitions. When you call cache() or persist(), each partition becomes a block stored by the BlockManager at the chosen storage level. On re-access, tasks read the cached blocks instead of recomputing.

PYTHON
# Cache uses Storage Memory
txn = spark.read.parquet("hdfs:///bank/transactions/2026-01/")
txn.cache()  # stored in Storage Memory
txn.count()  # materializes the cache

# Verify cache in Storage tab of Spark UI
print("Is cached:", txn.is_cached)

# Unpersist to free Storage Memory
txn.unpersist()
TEXT
Is cached: True

How does the dynamic boundary work?

The boundary between execution and storage is not a hard wall. If one side is idle, the other can borrow its space. The borrowing is mutual but the priority is asymmetric.

  • If storage is idle, execution can use the entire unified region for shuffles and joins.
  • If execution is idle, storage can expand for caching.
  • Execution can evict cached blocks that storage borrowed beyond its protected baseline.
  • Storage cannot evict active execution memory.
PYTHON
# Scenario 1: Heavy ETL with no caching
# Storage sits idle → execution uses the whole unified region
txn = spark.read.parquet("hdfs:///bank/transactions/2026/")
result = txn.groupBy("branch", "account_id") \
    .agg(F.sum("amount"), F.count("*")) \
    .orderBy("branch")
# Execution gets ~2.2 GB (full unified region) for shuffle/sort

# Scenario 2: Heavy caching with light processing
ref = spark.read.parquet("hdfs:///bank/reference/all-tables/")
ref.cache()  # uses most of Storage Memory
ref.count()  # materializes
# If execution then needs memory for a join, it can evict
# cached blocks that borrowed beyond storage's protected baseline

What happens when execution needs more memory?

When execution needs space and storage has borrowed beyond its protected baseline, Spark evicts cached blocks to free room. Eviction is cheap in correctness terms because an evicted cached partition can be recomputed from lineage or re-read. If execution still cannot get enough, it spills to disk.

PYTHON
# Over-caching scenario:
# 1. Cache a large reference table (fills storage + borrows from execution)
ref = spark.read.parquet("hdfs:///bank/reference/large-table/")
ref.cache()
ref.count()

# 2. Run a heavy join that needs execution memory
txn = spark.read.parquet("hdfs:///bank/transactions/2026/")
joined = txn.join(ref, "account_id").groupBy("branch").agg(F.sum("amount"))

# What happens:
# - Execution needs memory for the hash join
# - It evicts cached blocks that storage borrowed beyond baseline
# - If still not enough, execution spills to disk
# - The evicted cached blocks will be recomputed on next access

# Fix: unpersist what you no longer need
ref.unpersist()

Why does over-caching cause slow shuffles?

The non-obvious symptom of over-caching is not a cache error but slow shuffles. When too much memory is taken by cached data, execution memory shrinks, operators spill to disk, and shuffle-heavy stages slow down dramatically.

PYTHON
# Diagnosing memory pressure:
# Spark UI → Executors tab → check GC time
# Spark UI → Stages tab → check spill (memory) and spill (disk)
# Spark UI → Storage tab → check cached data size

# If cached data is large AND spill metrics are high:
# → Over-caching is stealing execution memory
# Fix: unpersist aggressively, avoid caching data used only once

# BAD: cache everything
txn.cache()       # used once
accounts.cache()  # used once
joined = txn.join(accounts, "account_id").agg(F.sum("amount"))

# GOOD: cache only what is reused
txn.cache()  # reused by multiple downstream queries
joined_1 = txn.join(accounts, "account_id").agg(F.sum("amount"))
joined_2 = txn.groupBy("branch").agg(F.avg("amount"))
txn.unpersist()  # done with txn

How did this replace the old static model?

Before Spark 1.6, execution and storage had fixed, non-overlapping pools. If one was starved while the other sat idle, memory was wasted. The unified model lets them share, so idle memory is never wasted.

PYTHON
# Old static model (pre-1.6):
# Execution: fixed 40%  |  Storage: fixed 60%
# If execution needed more, storage's idle space could not help
# If storage needed more, execution's idle space could not help

# New unified model (1.6+):
# Execution + Storage: shared 60% of heap
# Idle space from either side is available to the other
# Dynamic borrowing with execution priority

What should a senior engineer know?

A senior engineer reads the Spark UI storage tab and spill metrics together. They know that the symptom of too much caching is slow shuffles from spilling, not a cache error. They unpersist aggressively and avoid caching data used only once.

PYTHON
# Senior diagnostic pattern:
# 1. Check Storage tab: how much is cached?
# 2. Check Stages tab: is there spill (memory) and spill (disk)?
# 3. If both are high: over-caching is the cause
# 4. Fix: unpersist unused caches, reduce storage fraction if needed

spark.conf.set("spark.memory.storageFraction", "0.3")
# Gives execution 70% of unified region as baseline
# Useful for shuffle-heavy ETL that does little caching

Visual Overview

DIAGRAM
Spark Unified Memory Model (4 GB executor):

┌───────────────────────────────────────────────────┐
│                 Executor Heap (4 GB)               │
├───────────────────────────────────────────────────┤
│  Reserved Memory (~300 MB)                        │
│  (Spark internal objects)                         │
├───────────────────────────────────────────────────┤
│                                                   │
│  UNIFIED REGION (~2.2 GB)  spark.memory.fraction  │
│  ┌─────────────────────┬─────────────────────┐    │
│  │  Execution Memory   │  Storage Memory     │    │
│  │  (shuffles, joins,  │  (cached/persisted  │    │
│  │   sorts, aggs)      │   partitions)       │    │
│  │                     │                     │    │
│  │  ~1.1 GB baseline   │  ~1.1 GB baseline   │    │
│  │         ◄───── dynamic boundary ─────►    │    │
│  │                                           │    │
│  │  Priority: execution can evict storage    │    │
│  │  Storage CANNOT evict active execution    │    │
│  └─────────────────────┴─────────────────────┘    │
│                                                   │
├───────────────────────────────────────────────────┤
│  User Memory (~1.5 GB)                            │
│  (your objects, non-Spark data structures)        │
└───────────────────────────────────────────────────┘

Borrowing Rules:
  Execution idle → Storage expands:  [====STORAGE=====|exec]
  Storage idle   → Execution expands:[EXECUTION=======|stor]
  Both active    → Execution evicts borrowed storage blocks

Key Takeaway

Spark's unified memory model splits the heap region it manages for data processing into Execution Memory and Storage Memory. Execution Memory holds the transient working set of shuffles, joins, sorts, and aggregations, while Storage Memory holds cached and persisted partitions. The size of the managed region is set by spark.memory.fraction, around sixty percent of usable heap by default, with the rest left as user memory, and within that region spark.memory.storageFraction sets a protected baseline for storage, around half by default. The crucial feature is that the boundary between the two is dynamic rather than a fixed wall.

The borrowing is mutual but the priority is asymmetric. If storage is idle, execution can use the whole region for its shuffles and joins, and if execution is idle, storage can expand for caching. But execution has priority: when it needs space it can evict cached blocks that storage borrowed beyond its protected baseline, because an evicted cached partition can simply be recomputed or re-read, whereas interrupting an active sort or hash table would fail the computation. Storage can never forcibly reclaim memory that execution is actively using.

At the National Bank of India this mattered constantly. A job that cached a reference accounts dataset and then ran a heavy transactions join would see execution evict the borrowed cache when the join needed memory, which was fine because the cache could be rebuilt. But a job that over-cached would push execution into spilling and run slowly. The non-obvious lesson I share in interviews is that the symptom of too much caching is not a cache error but slow shuffles from spilling, so when diagnosing memory pressure I look at the storage tab and the spill metrics in the Spark UI together.

The unified model replaced the old static split precisely so that an idle region never goes to waste, and understanding the execution-over-storage priority is what lets you reason about why a job spills.

Remember one sentence: execution and storage share one unified pool with a dynamic boundary, execution has priority and can evict cached blocks, and over-caching shows up as slow shuffles, not cache errors.

---

Share
Use ← → arrow keys to navigate