Spark Memory Architecture and OOM Debugging

“The job died with an OutOfMemoryError. What do you do?” is one of the most common senior Spark questions, and one of the most commonly fumbled. “Add more memory” is the junior answer. The senior answer identifies which process ran out (driver or executor), which memory region (heap, off-heap, overhead, Python), why (skew, too-large partitions, collect, broadcast, caching, UDFs), and fixes the cause. This module gives you the mental model and the debugging path.

1. The memory layout of an executor

flowchart TB
    subgraph CONT["Container / pod limit = heap + overhead (+ off-heap + PySpark memory)"]
        subgraph HEAP["JVM heap: spark.executor.memory"]
            RES["Reserved: 300 MB<br/>Spark internals"]
            subgraph UNI["Unified memory = (heap − 300 MB) × spark.memory.fraction (0.6)"]
                EXE["Execution<br/>shuffle buffers, sort, hash aggregation,<br/>join hash tables"]
                STO["Storage<br/>cached blocks, broadcast variables,<br/>unrolled blocks"]
            end
            USR["User memory = (heap − 300 MB) × 0.4<br/>your objects, UDF data structures,<br/>Spark internal metadata"]
        end
        OFF["Off-heap (spark.memory.offHeap.size)<br/>optional Tungsten off-heap pool"]
        OVH["memoryOverhead: max(384 MB, 10%)<br/>JVM native memory, thread stacks, NIO/Netty buffers,<br/>metaspace, Python workers*, Arrow buffers*"]
        PYM["spark.executor.pyspark.memory (optional)<br/>caps Python worker memory"]
    end

* If spark.executor.pyspark.memory isn’t set, Python workers draw from the overhead allowance.

Worked example: spark.executor.memory=16g, 4 cores.

• Usable heap ≈ 16 GB − 300 MB ≈ 15.7 GB.
• Unified (execution + storage) ≈ 15.7 × 0.6 ≈ 9.4 GB; user memory ≈ 6.3 GB.
• spark.memory.storageFraction=0.5 makes ≈ 4.7 GB of storage protected from eviction by execution; the rest is borrowable either way.
• Per running task: execution memory is shared fairly among active tasks, so each of 4 tasks gets between 1/(2N) and 1/N of the execution pool, roughly 1.2-2.3 GB.
• Container request ≈ 16 GB + overhead (1.6 GB by default) ≈ 17.6 GB.

The borrowing rules (the asymmetry interviewers ask about)

• Execution and storage share one pool and borrow each other’s free space.
• Execution can evict cached blocks (storage beyond the protected storageFraction) when it needs memory.
• Storage cannot evict execution memory: if execution holds it, new cache blocks are dropped or spilled to disk (per storage level).
• Reasoning: execution memory is needed now to finish tasks; cached data can always be recomputed.

The driver

The driver has the same JVM structure (spark.driver.memory, spark.driver.memoryOverhead), but it holds very different things:

• query plans and task metadata;
• results of collect(), toPandas(), take();
• broadcast relations, which are collected to the driver before being shipped;
• accumulators, listener events, and the Spark UI’s job and stage history.

spark.driver.maxResultSize (default 1 GB) caps the total serialized size of results collected from tasks, so a failing job is better than a dead driver.

2. Every kind of out-of-memory failure, decoded

3. The debugging path

flowchart TD
    A[Job failed with OOM / container killed] --> B{Driver or executor?}
    B -->|"driver log, 'Driver stacktrace', app dies at once"| D1[Driver path]
    B -->|"task failures, ExecutorLostFailure, exit 137 on executors"| E1[Executor path]
    D1 --> D2{"collect / toPandas / take?<br/>broadcast of a big table?<br/>millions of tiny partitions?<br/>huge plan?"}
    D2 --> D3[Remove the collect, limit/aggregate,<br/>fix broadcast stats/hints, coalesce, simplify plan,<br/>then raise driver memory if still needed]
    E1 --> E2{Which stage and task?<br/>Stages tab: max vs median}
    E2 -->|"one or few tasks huge"| SK[Skew: see the skew guide]
    E2 -->|"all tasks large"| E3{Spill and GC metrics}
    E3 -->|"heavy spill everywhere"| P[Partitions too big:<br/>more shuffle partitions / AQE advisory size]
    E3 -->|"high GC time"| G[Too many objects per task:<br/>fewer cores per executor, serialized cache,<br/>DataFrames over RDDs, G1GC]
    E3 -->|"container killed, heap looks fine"| O[Overhead / Python memory:<br/>raise overhead or pyspark.memory,<br/>smaller Arrow batches, avoid UDFs]

Step 1: Driver or executor?

• Driver: the whole application dies, the error appears in the driver log, and often right after an action like collect() or during planning or broadcasting.
• Executor: individual tasks fail and are retried (Lost task … ExecutorLostFailure), and the error appears in executor logs; the job fails after spark.task.maxFailures (4).

Step 2: Which stage and task?

In the Stages tab, open the failing stage and look at the Summary Metrics for completed tasks: duration, shuffle read size, records, spill, GC time (min / 25th / median / 75th / max).

• max ≫ median on shuffle read or records → skew: one task got a giant partition. Go to skew.
• all tasks similar but big → partitions too large for per-task memory.

Step 3: Spill

Spill (Memory) vs Spill (Disk): spill means operators already ran out of execution memory and wrote to disk. A ratio of memory to disk of roughly 3-10× is normal (deserialized vs compressed). Heavy spill on every task is the precursor to OOM: increase partitions or memory per task.

Step 4: GC time

GC time above ~10% of task time signals memory pressure; above 20-30% is a problem. Patterns:

• High GC everywhere: too many cores per executor for its heap, deserialized caches, object-heavy code (RDDs, Python objects in the JVM).
• High GC on a few tasks: skew.
• GC rising over the job: cached data accumulating, or a leak in a long-lived session.

Step 5: GC logs (when it isn’t obvious)

Add -Xlog:gc*:file=… (JDK 11+) or -verbose:gc via spark.executor.extraJavaOptions. Look for back-to-back full GCs that reclaim little (heap genuinely too small for the working set), long pauses, or humongous allocations with G1 (very large arrays such as big join hash tables or large records).

Step 6: Fix the cause, then size

Only after removing causes (skew, collect, oversized broadcast, UDF memory) should you resize: more memory per task (fewer cores or more memory), more overhead for PySpark, more partitions.

4. GC and serialization tuning

• G1GC (the default on modern JVMs) suits most Spark workloads; tune -XX:InitiatingHeapOccupancyPercent (start concurrent marking earlier, e.g. 35) and -XX:G1HeapRegionSize for very large objects. ZGC/Shenandoah reduce pause times for very large heaps or latency-sensitive jobs, at some throughput cost.
• Fewer, smaller objects: DataFrames (Tungsten binary rows) instead of RDDs of objects; serialized storage levels for RDD caches; Kryo for RDD serialization.
• Cached data lives in the old generation: large MEMORY_ONLY caches increase full-GC cost. Prefer DataFrame columnar caching or MEMORY_AND_DISK_SER, and unpersist when done.
• Avoid giant heaps per executor (say over 64 GB) with many cores: GC pauses grow. Prefer more medium executors (4-5 cores).

5. Observability: where to look

6. Memory anti-patterns

1. collect() / toPandas() on unbounded data. Aggregate or write out instead.
2. Broadcasting a table that’s large after decompression (or is large because statistics are stale).
3. collect_list/collect_set over unbounded groups (one key → millions of elements in one row).
4. Python UDFs on large columns without budgeting memory overhead.
5. Building plans in a loop of withColumn (hundreds of nested projections → driver memory and planning time); use one select with all expressions.
6. Fat executors (all cores, huge heap) → long GC pauses and shared-overhead OOMs.
7. Caching everything “just in case”: evicts useful data and adds GC pressure.
8. Too few shuffle partitions for the data (each task gets GBs).
9. Ignoring skew and “fixing” it with bigger executors, which only moves the cliff.

7. Configuration reference

Interview questions