Spark on the JVM: Tungsten Memory, Serialization (Kryo) & Debugging Executors

spark on the jvm is the single fact about Apache Spark that most engineers know but never operationalise — and it is the reason the same job that ran fine on a laptop against 10 GB explodes with java.lang.OutOfMemoryError: Java heap space the moment it meets 2 TB in production. Every Spark executor is a Java Virtual Machine process. Every row you cache, every record you shuffle, every object you broadcast lives — at least for a moment — as bytes managed by that JVM's heap, its garbage collector, and its serialization machinery. When a Spark job is slow, the cause is almost never "Spark is slow"; it is object overhead inflating a 40-byte record into 200 bytes of heap, or Java's default serializer writing full class names into every shuffle block, or an executor's execution-memory pool exhausting and spilling gigabytes to disk, or the garbage collector burning 40% of every task's wall-clock time trying to keep up.

This guide is the walkthrough you wished existed the first time an interviewer asked "your executors keep getting killed by YARN — walk me through why," or "what does Tungsten actually do and why does off-heap memory help," or "you switched to Kryo and the shuffle got smaller — explain the mechanism." It works through the four JVM-shaped forces that govern every Spark job — how the heap is laid out and where object overhead comes from, how serialization (Java versus Kryo) dominates the cost of shuffle, cache, and broadcast, how the unified memory model partitions an executor into reserved, user, storage, and execution regions and when each spills, and how to debug a failing executor from the Spark UI and the GC log down to the exact line of tuning that fixes it. Each section pairs a teaching block with a Solution-Tail interview answer — runnable config and code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the optimization practice library →, rehearse on the data-processing practice library →, and sharpen the pipeline axis with the ETL practice library →.

---

On this page

• Why Spark's JVM foundation leaks into every performance problem
• Tungsten — off-heap, binary rows, whole-stage codegen
• Serialization — Java vs Kryo
• Executor memory model — unified, off-heap, overhead, spill
• Debugging executors — OOM, GC, the Spark UI
• Cheat sheet — Spark-on-the-JVM recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why Spark's JVM foundation leaks into every performance problem you'll debug

Every executor is a JVM — so every OOM, GC pause, and slow shuffle is a JVM story wearing a Spark costume

The one-sentence invariant: an Apache Spark cluster is a fleet of JVM processes (one driver, N executors) coordinated by a scheduler, and every performance problem you will ever debug — a job that OOMs, a stage that GC-thrashes, a shuffle that takes ten times longer than the compute — is downstream of how those JVMs lay out objects on the heap, serialize records to move them between processes, partition a fixed memory budget between caching and computation, and reclaim dead objects with a garbage collector. You do not need to be a JVM internals expert to write correct Spark, but you cannot tune Spark — or answer a senior Spark interview — without holding the mental model of the JVM underneath. The reason "just add more memory" fails as often as it works is that the memory problem usually isn't a shortage of RAM; it's object overhead, serialization bloat, an unbalanced memory split, or GC pressure — four different diseases with four different cures, all of which look identical from the outside (the job is slow or the container died).

The four JVM-shaped forces every Spark job fights.

• Heap layout and object overhead. A JVM object is not just its data. Every java.lang.Object carries a 12–16 byte header (mark word + class pointer), pointers are 4 or 8 bytes, String is a char[] plus a wrapper, and boxed Integer costs ~16 bytes to hold a 4-byte value. A dataset that is 40 bytes/row on disk can balloon to 200+ bytes/row as live JVM objects. This overhead is why caching a "small" DataFrame can OOM an executor.
• Serialization. The JVM cannot send an object graph over the network or write it to disk as-is — it must be serialized into bytes and deserialized back. Every shuffle, every cache() with a serialized storage level, every broadcast, and every task result crosses this boundary. Java's built-in serializer is correct but slow and fat; Kryo is the faster, smaller alternative. Serialization cost is the most under-diagnosed Spark bottleneck.
• Memory partitioning. Each executor's heap is carved into regions: a small reserved slice, a user-memory slice for your objects and UDF state, and a large unified pool shared between storage (cached blocks) and execution (shuffle/sort/join/aggregation buffers). When execution memory runs out, Spark spills to disk. Getting the partitioning wrong means either needless spill or OOM.
• Garbage collection. The JVM reclaims dead objects automatically, but reclamation is not free — it costs CPU and, with the wrong collector or heap size, long stop-the-world pauses. A Spark task that spends 40% of its time in GC is not compute-bound; it is allocation-bound, and the fix is usually fewer/leaner objects (Tungsten, Kryo, off-heap), not a bigger heap.

The 2026 reality — the levers are known, the microscope is the Spark UI.

• Tungsten is the engine that lets Spark SQL / DataFrame / Dataset code sidestep JVM object overhead: it stores rows in a compact off-heap binary format (UnsafeRow) and generates specialised bytecode (whole-stage codegen) instead of walking a chain of iterator objects. If you use the DataFrame API, you already benefit; if you use raw RDDs of case classes, you mostly don't.
• Kryo is the serializer you flip on to shrink and speed up every shuffle, cache, and broadcast. It is not the default (for backward-compatibility reasons), so a huge fraction of "slow shuffle" problems are solved by two config lines plus a class-registration list.
• The unified memory manager (default since Spark 1.6) auto-balances storage and execution, which removed a whole category of manual tuning — but you still choose the executor size, the memory.fraction, whether to enable off-heap, and the memoryOverhead that keeps your container alive.
• The Spark UI (Stages, Executors, SQL tabs) plus the GC log is the microscope. Every diagnosis in this guide ends at a number you can read there: shuffle spill (memory/disk), GC time as a fraction of task time, peak execution memory, task-duration skew.

What interviewers actually probe.

• Do you say "executors are JVMs" in the first sentence when asked about Spark memory? — required framing.
• Can you distinguish heap OOM from container-overhead OOM (OutOfMemoryError versus Container killed by YARN for exceeding memory limits)? — senior signal.
• Do you name Kryo and why it's smaller (int class IDs vs full class names), not just "it's faster"? — senior signal.
• Do you describe spill as a normal, bounded mechanism (execution memory exhausted → sort/agg spills sorted runs to disk) rather than as a failure? — required answer.
• Do you reach for the Spark UI and GC log before reaching for --executor-memory 32g? — senior signal.

Worked example — the JVM-forces diagnostic table

Detailed explanation. The most useful artifact for a Spark-tuning interview is a table that maps a symptom to the JVM force behind it and the lever that fixes it. Interviewers hand you a symptom ("the job is slow / the container died") and grade whether you can name the mechanism instead of guessing. Build the table for a canonical failing job: a nightly aggregation of a 2 TB event log that OOMs on some executors and GC-thrashes on the rest.

• Symptom surface. "Job is slow and some executors die" — deliberately ambiguous, like a real ticket.
• Candidate causes. Object overhead, serialization bloat, memory-split imbalance, GC pressure — the four forces.
• Evidence source. Each cause has a distinct fingerprint in the Spark UI or GC log.
• Goal. Produce a symptom → cause → evidence → lever table you can recite.

Question. Map the four JVM forces to their symptom, their evidence in the Spark UI / GC log, and the primary lever that addresses each.

Input.

JVM force	Typical symptom	Evidence to look at
Object overhead	cache OOMs; RDD jobs bloat	Storage tab size >> on-disk size
Serialization	slow shuffle; large shuffle write	Shuffle Write bytes; task serialization time
Memory split	frequent spill or OOM	Shuffle Spill (Memory/Disk); peak exec memory
GC pressure	high task time, low CPU work	GC Time / Task Time ratio; GC log pauses

Code.

// Turn on the diagnostics you need BEFORE you tune anything.
// spark-submit flags (Scala or PySpark; these are engine-level).
//   --conf spark.eventLog.enabled=true              // persist the Spark UI
//   --conf spark.eventLog.dir=hdfs:///spark-logs
//   --conf spark.executor.extraJavaOptions=
//       "-XX:+UseG1GC -Xlog:gc*:file=/tmp/gc-%p.log:time,uptime:filecount=5,filesize=20m"

// Programmatic view of the same knobs (Scala Spark):
val spark = org.apache.spark.sql.SparkSession.builder()
  .appName("jvm-forces-diagnostic")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .config("spark.sql.adaptive.enabled", "true")   // AQE: coalesce + skew handling
  .getOrCreate()

// Read the four forces back at runtime:
val sc = spark.sparkContext
println(s"serializer      = ${sc.getConf.get("spark.serializer", "JavaSerializer(default)")}")
println(s"memory.fraction = ${sc.getConf.get("spark.memory.fraction", "0.6")}")
println(s"offHeap.enabled = ${sc.getConf.get("spark.memory.offHeap.enabled", "false")}")

Step-by-step explanation.

1. Object overhead shows up in the Storage tab: a DataFrame that is 8 GB on disk (Parquet, columnar, compressed) can occupy 30 GB+ as deserialized JVM objects when cached with the default MEMORY_AND_DISK level. The lever is to cache with MEMORY_ONLY_SER (serialized, so it stores bytes not objects) or to prefer the DataFrame API where Tungsten already stores compact binary rows.
2. Serialization bloat shows up as large Shuffle Write bytes and non-trivial task serialization time. Java serialization writes the fully-qualified class name into every serialized object; a shuffle of 100M small records pays that tax 100M times. The lever is Kryo plus registering the hot classes so they are keyed by a small integer instead of a class-name string.
3. Memory-split problems show up as Shuffle Spill (Memory) and Shuffle Spill (Disk) being large, or as heap OOM when execution memory can't grow. The lever is right-sizing the executor and, when appropriate, enabling off-heap execution memory so large sort/aggregate buffers live outside the GC-managed heap.
4. GC pressure shows up as a high GC Time relative to Task Time in the Executors tab, and as frequent or long pauses in the GC log. The lever is reducing allocation (Tungsten, Kryo, off-heap, fewer wide UDF objects) and using G1GC with a sane heap — not blindly increasing --executor-memory, which can make full-GC pauses longer.
5. The discipline the table enforces: name the force from the evidence before choosing the lever. "Add memory" is the right fix for exactly one of the four rows and the wrong fix for the other three.

Output.

Symptom seen	JVM force named	Primary lever
Cached DF 4× bigger than Parquet	object overhead	serialized cache / DataFrame API / off-heap
Shuffle write huge, tasks slow	serialization	Kryo + registerKryoClasses
Spill (Disk) in the GBs	memory split	right-size executor / enable off-heap exec
GC time > 25% of task time	GC pressure	fewer objects (Tungsten/Kryo) + G1GC

Rule of thumb. Before you touch --executor-memory, read four numbers: cached size vs on-disk size, shuffle write bytes, shuffle spill, and GC-time fraction. Each points at a different one of the four JVM forces, and only one of them is fixed by more RAM.

Worked example — driver JVM vs executor JVM

Detailed explanation. A subtle but frequently-tested distinction: the driver is also a JVM, and it fails for different reasons than executors. collect() on a 5 GB result, a broadcast that's too big, an accumulator that grows unbounded, or a query plan with tens of thousands of partitions can OOM the driver while every executor is healthy. Candidates who conflate the two mis-diagnose half of all real incidents.

• Driver responsibilities. Builds the DAG, schedules tasks, holds broadcast variables before they're shipped, collects results, maintains accumulators and the block-manager master.
• Executor responsibilities. Runs tasks, holds cached blocks, performs shuffle read/write, executes the actual transformations.
• Different failure fingerprints. Driver OOM often follows a collect, toPandas, broadcast, or a huge plan; executor OOM follows a shuffle, join, or cache.

Question. For four common failures, decide whether it is a driver JVM problem or an executor JVM problem, and name the lever.

Input.

Failure event	Driver or executor?
df.collect() on a 6 GB result	?
Shuffle stage: Container killed ... exceeding memory	?
spark.sql.autoBroadcastJoinThreshold join, one side 2 GB	?
toPandas() on a wide 3 GB frame	?

Code.

# PySpark — the driver-side vs executor-side failure lines
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
         .appName("driver-vs-executor")
         # driver gets its OWN heap + overhead, sized separately from executors:
         .config("spark.driver.memory", "4g")
         .config("spark.driver.maxResultSize", "2g")   # caps collect() before it OOMs the driver
         .config("spark.executor.memory", "8g")
         .config("spark.executor.memoryOverhead", "2g")
         .getOrCreate())

big = spark.read.parquet("s3://events/2026/")

# DRIVER-side risk: pulls every row to the driver JVM heap.
# result = big.collect()          # 6 GB -> driver OOM (or maxResultSize abort)

# EXECUTOR-side pattern: aggregation stays distributed, only small result returns.
agg = (big.groupBy("event_type")
          .agg(F.count("*").alias("n"))
          .orderBy(F.desc("n")))
agg.show(20)                       # safe: tiny result crosses back to the driver

Step-by-step explanation.

1. df.collect() on a 6 GB result is a driver failure: collect streams every partition's rows back to the driver JVM and materialises them in a single Array. The spark.driver.maxResultSize guard aborts the job with a clear message instead of letting the driver OOM silently. The lever is: don't collect big results — aggregate/write distributed, or limit() first.
2. Container killed ... exceeding memory during a shuffle is an executor failure — specifically an overhead OOM (the container's total footprint, heap + off-heap + native, exceeded the YARN/K8s limit). The lever is raising spark.executor.memoryOverhead or reducing per-task footprint, not raising heap.
3. A broadcast join where the "small" side is 2 GB is a driver problem first: the driver collects the broadcast relation to its heap before shipping it. Above spark.sql.autoBroadcastJoinThreshold (default 10 MB) Spark won't auto-broadcast; forcing a 2 GB broadcast OOMs the driver. The lever is: don't broadcast large sides; let Spark pick a sort-merge join.
4. toPandas() on a 3 GB frame is a driver failure — it is collect plus a conversion into a single pandas object on the driver heap. The lever is Arrow-optimised conversion for smaller frames, or don't pull it to the driver at all.
5. The through-line: driver failures follow "bring data to one JVM" operations (collect, toPandas, broadcast, giant plans); executor failures follow "process data in many JVMs" operations (shuffle, join, cache). Size and tune them independently.

Output.

Failure event	Which JVM	Lever
collect() 6 GB	driver	avoid collect / maxResultSize guard
Container killed in shuffle	executor	raise memoryOverhead / cut footprint
2 GB broadcast	driver	don't broadcast large side
toPandas() 3 GB	driver	Arrow for small frames only

Rule of thumb. Ask "does this operation bring data to one JVM or process it across many?" Bring-to-one failures are driver JVM problems; process-across-many failures are executor JVM problems. They have separate memory settings for a reason.

Senior interview question on Spark's JVM foundation

A senior interviewer often opens with: "One of our nightly Spark jobs used to finish in 30 minutes and now takes three hours and occasionally loses executors. Nothing in the code changed — only the data grew about 4×. Walk me through how you'd reason about the problem given that Spark runs on the JVM, and which numbers you'd pull before changing a single config."

Solution Using a JVM-forces triage against the Spark UI

# triage.py — read the four JVM-force fingerprints from the event log,
# then decide the lever. Run against a finished (or running) app's UI/REST API.
import requests

APP = "http://spark-history:18080/api/v1/applications/app-20260803-0001"

def get(path):
    return requests.get(f"{APP}{path}").json()

execs = get("/executors")
stages = get("/stages")

# 1. GC pressure: total GC time vs total task time across executors
gc_ms   = sum(e["totalGCTime"]      for e in execs if e["id"] != "driver")
task_ms = sum(e["totalDuration"]    for e in execs if e["id"] != "driver")
gc_frac = gc_ms / max(task_ms, 1)

# 2. Serialization/shuffle: total shuffle write across stages
shuffle_write = sum(s.get("shuffleWriteBytes", 0) for s in stages)

# 3. Memory split: total spill across stages
spill_mem  = sum(s.get("memoryBytesSpilled", 0) for s in stages)
spill_disk = sum(s.get("diskBytesSpilled", 0)   for s in stages)

print(f"GC fraction        : {gc_frac:6.1%}")
print(f"Shuffle write (GB) : {shuffle_write/1e9:6.1f}")
print(f"Spill mem/disk (GB): {spill_mem/1e9:5.1f} / {spill_disk/1e9:5.1f}")

# 4. Decision
if gc_frac > 0.20:
    print("-> GC-bound: reduce allocation (Kryo + off-heap), switch to G1GC, do NOT just add heap")
elif spill_disk > 5e9:
    print("-> spill-bound: raise exec memory fraction / enable off-heap / more partitions")
elif shuffle_write > 200e9:
    print("-> shuffle-heavy: enable Kryo + register classes; reduce shuffle via broadcast/repartition")
else:
    print("-> not memory-shaped: check skew (max vs median task time) and input partition count")

// The same triage, expressed as the mental checklist you say out loud:
//   1. Did GC time as a fraction of task time cross ~20%?   -> allocation problem
//   2. Is disk spill in the multi-GB range?                 -> execution-memory shortage
//   3. Is shuffle write enormous relative to input?         -> serialization / wide shuffle
//   4. Is max task time >> median task time?                -> data skew, not memory at all
// Only AFTER naming the force do you pick: Kryo, off-heap, memory.fraction, partitions, or skew join.

Step-by-step trace.

Step	What you pull	What it tells you
1	GC time / task time from /executors	allocation pressure (GC-bound?)
2	diskBytesSpilled from /stages	execution-memory shortage
3	shuffleWriteBytes from /stages	serialization + shuffle width
4	max vs median task duration	data skew (a non-memory cause)
5	cached size vs input size (Storage tab)	object-overhead blow-up

Walking the triage on the "4× data, 6× time" job: GC fraction reads 34%, disk spill reads 22 GB, shuffle write reads 310 GB. That fingerprint says allocation- and spill-bound, not "needs a bigger box." The plan writes itself: flip on Kryo and register the record classes (shrinks shuffle write and per-record allocation), enable off-heap execution memory (moves the large aggregation buffers off the GC-managed heap so spill and GC both drop), and switch to G1GC. Only if spill persisted afterwards would you touch executor size — and even then you'd add cores/partitions before raw heap.

Output:

Metric	Before triage-driven fix	After
GC time fraction	34%	8%
Disk spill	22 GB	3 GB
Shuffle write	310 GB	140 GB
Wall-clock runtime	~3 h	~50 min
Executors lost to OOM	2–3 / run	0

Why this works — concept by concept:

• Executors are JVMs — because each executor is a JVM process, the four levers (object layout, serialization, memory split, GC) are the only things you can tune; every symptom maps to one of them, which turns "the job is slow" into a bounded decision.
• Evidence before levers — pulling GC fraction, spill, and shuffle write first prevents the classic anti-pattern of throwing memory at a GC or serialization problem, where more heap makes pauses longer, not shorter.
• GC fraction as the allocation signal — GC time / task time is the single best proxy for "am I allocating too many objects?" Above ~20% you are allocation-bound and the fix is fewer/leaner objects (Kryo, Tungsten, off-heap), not more RAM.
• Skew is not a memory bug — the max-vs-median task-time check catches the common misdiagnosis where one giant partition (not a memory shortage) is the real cause; adding memory never fixes skew.
• Cost — the triage is O(1) API calls and zero cluster cost, and it replaces an O(N) trial-and-error loop of "add memory, resubmit, wait, repeat." The payoff is picking the correct one of four levers on the first try.

Optimization
Topic — optimization
Optimization problems on compute + memory tuning

Practice →

Data Processing Topic — data-processing Data-processing problems on distributed compute

Practice →

---

2. Tungsten — off-heap memory, binary row format, cache-aware compute, whole-stage codegen

Tungsten exists to make Spark stop thinking like a JVM — compact binary rows off the heap, and generated code instead of iterator chains

The mental model in one line: tungsten is Spark's execution engine rewrite (Spark 1.4–2.0) whose entire purpose is to escape the JVM's object-model tax — instead of representing each row as a graph of Row/String/boxed-Integer objects on the garbage-collected heap, Tungsten stores rows as a compact, schema-aware binary layout (UnsafeRow) that it can put in off-heap memory, operate on cache-efficiently, and process with specialised generated bytecode (whole-stage codegen) rather than a chain of virtual next() calls. Everything you get "for free" from the DataFrame, Dataset, and Spark SQL APIs — the reason they routinely beat hand-written RDD code — is Tungsten. Understanding it is what lets you explain why the DataFrame API is faster and when you accidentally fall off the Tungsten fast path.

The problem Tungsten solves — the JVM object tax.

• Header overhead. Every JVM object carries a 12–16 byte header. A row modelled as objects pays that header per field-wrapper. Tungsten's UnsafeRow has one small fixed region (null-tracking bitset + fixed-length values) plus a variable-length region — no per-field object headers.
• Pointer chasing. An object graph is scattered across the heap; reading a row means following pointers, which defeats CPU cache prefetching. UnsafeRow is a single contiguous byte buffer, so a row read is a linear scan — cache-friendly.
• GC pressure. Millions of short-lived row objects per second is exactly the allocation pattern that keeps the garbage collector busy. Storing rows as bytes (especially off-heap) means the GC never sees them, so GC time drops sharply on Tungsten-heavy stages.

The four Tungsten mechanisms.

• Binary row format (UnsafeRow). A row is [null-bit set][fixed-length values, 8 bytes each][variable-length blob]. Fixed-width fields (long, double, the offset+length of a string) live inline; variable-width payloads (string/array/map bytes) live in the trailing region. Reading field k is pointer arithmetic, not object dereference.
• Off-heap allocation. With spark.memory.offHeap.enabled=true, Tungsten allocates these binary rows via sun.misc.Unsafe in memory outside the JVM heap. That memory is not garbage-collected and not counted against the heap — so it neither triggers GC nor causes heap OOM (though it is counted against the container's total, which matters for section 4 and 5).
• Cache-aware compute. Tungsten's sort and aggregation operators are written to exploit CPU cache lines — e.g. sorting arrays of (key-prefix, pointer) so comparisons hit a prefix in-cache before dereferencing the full row. This is why Tungsten sort beats a naive object-comparator sort by multiples.
• Whole-stage codegen. Instead of executing an operator tree as a chain of iterator objects each calling next() (one virtual dispatch per row per operator), Tungsten generates a single Java method that fuses an entire stage — filter, project, aggregate — into one tight loop, then compiles it with Janino. The generated code looks like what a competent engineer would hand-write, minus the object churn.

When you fall OFF the Tungsten fast path.

• Typed Dataset operations with lambdas. ds.map(x => ...) on a Dataset[CaseClass] forces Spark to deserialize each UnsafeRow back into a JVM object, run your lambda, then re-serialize — you pay the object tax you were avoiding. The DataFrame/Column-expression API (df.filter($"x" > 3)) stays in codegen; the typed lambda doesn't.
• Python/Scala UDFs (non-vectorised). A plain Python UDF ships each row out of the JVM to a Python worker and back — off the Tungsten path entirely. Prefer built-in functions or pandas/Arrow UDFs.
• Very wide plans. Whole-stage codegen has a method-size limit (the JVM's 64 KB bytecode-per-method ceiling); extremely wide stages fall back to the interpreted path. spark.sql.codegen.wholeStage and the *(n) markers in explain tell you what got fused.

Worked example — measuring the object tax vs UnsafeRow

Detailed explanation. To make the object tax concrete, estimate the live-object size of a row modelled as JVM objects versus the same row as an UnsafeRow. This is exactly the calculation that explains why a cached DataFrame's Storage-tab size can be several times its Parquet size, and why serialized/off-heap storage shrinks it.

• Row schema. (id: Long, name: String[12 chars], amount: Double, active: Boolean).
• Object model. A wrapper object + boxed fields + a String (char[] + object header).
• UnsafeRow model. Null bitset + fixed 8-byte slots + inline string bytes.

Question. Estimate bytes-per-row for the JVM-object representation versus the UnsafeRow representation, and explain the gap.

Input.

Field	Java-object cost (approx)	UnsafeRow cost
object header	16 B	0 (one row-level null bitset, 8 B)
id (Long)	16 B (boxed)	8 B inline
name (String, 12 chars)	16 hdr + 24 char[] + 24 = ~64 B	8 B (offset+len) + 12 B bytes
amount (Double)	16 B (boxed)	8 B inline
active (Boolean)	16 B (boxed)	in null bitset / 8 B slot

Code.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.UnsafeRow

val spark = SparkSession.builder().appName("object-tax").getOrCreate()
import spark.implicits._

case class Txn(id: Long, name: String, amount: Double, active: Boolean)

val ds = Seq(
  Txn(1L, "alice_smith", 42.50, true),
  Txn(2L, "bob_jones",   17.00, false)
).toDS()

// DataFrame path keeps rows as UnsafeRow internally.
val df = ds.toDF()

// Estimate the object-model size (SizeEstimator walks the live object graph):
import org.apache.spark.util.SizeEstimator
val oneObject = Txn(1L, "alice_smith", 42.50, true)
println(s"JVM object graph bytes/row ~ ${SizeEstimator.estimate(oneObject)}")

// Show whether the plan stays in Tungsten codegen (look for WholeStageCodegen + '*'):
df.filter($"amount" > 20).groupBy($"active").count().explain(true)

Step-by-step explanation.

1. SizeEstimator.estimate walks the actual JVM object graph and typically reports ~80–120 bytes for this four-field row, because every field that is boxed or wrapped drags in a 16-byte header and the String alone is a char[] plus its wrapper.
2. The same row as an UnsafeRow is roughly 8 (null bitset) + 3×8 (fixed slots for id, amount, and the string's offset+length) + 12 (the actual "alice_smith" bytes, padded to 8) ≈ 48 bytes — and, crucially, it is one contiguous buffer, so there is no pointer chasing and no per-field GC bookkeeping.
3. The gap (roughly 2–3×) is the object tax. Multiply by hundreds of millions of rows and it is the difference between a cache that fits in memory and one that spills, and between a GC that idles and one that thrashes.
4. explain(true) shows *(1) HashAggregate / *(2) ... markers — the *(n) prefix means that operator was fused by whole-stage codegen into generated method n. If you see operators without the *, that part fell back to the interpreted iterator path (often because of a UDF or a typed lambda).
5. Enabling off-heap (spark.memory.offHeap.enabled=true, spark.memory.offHeap.size=...) tells Tungsten to place these UnsafeRow buffers outside the heap entirely — the GC never scans them, which is why off-heap most helps GC-bound, allocation-heavy stages.

Output.

Representation	Approx bytes/row	GC-visible?	Cache-friendly?
JVM object graph (RDD of case class)	~80–120 B	yes (heavy)	no (pointer chase)
UnsafeRow on-heap (DataFrame)	~48 B	yes (light)	yes (contiguous)
UnsafeRow off-heap (offHeap enabled)	~48 B	no	yes

Rule of thumb. Prefer DataFrame/Column-expression operations over typed lambdas and Python UDFs so your rows stay as UnsafeRow inside whole-stage codegen. Check explain for *(n) markers; every operator that loses the * is an operator that fell back to the object-tax path.

Worked example — reading a whole-stage codegen plan

Detailed explanation. Interviewers love to put an explain output in front of you and ask "which parts are codegen'd and which aren't, and why?" Being able to read the physical plan — the *(n) fusion markers, exchange (shuffle) boundaries, and the operators that break codegen — is a concrete senior signal.

• Fusion markers. *(1), *(2) group operators into generated methods; a new number appears after a stage boundary.
• Codegen breakers. Exchange (shuffle), some joins, and any operator wrapping a UDF or typed lambda break the fused loop.
• Goal. Point at each operator and say "fused / not fused, and here's why."

Question. Given a filtered, grouped aggregation over a Parquet source, identify the whole-stage-codegen boundaries and the shuffle.

Input.

Query step	Expectation
scan parquet	codegen'd (*) column read
filter amount > 20	fused into scan stage
exchange hashpartitioning(active)	shuffle boundary (not codegen'd)
partial + final HashAggregate	codegen'd on each side of the exchange

Code.

val plan = spark.read.parquet("s3://txns/")
  .filter($"amount" > 20)
  .groupBy($"active")
  .agg(org.apache.spark.sql.functions.sum($"amount").as("total"))

plan.explain("formatted")
/*
== Physical Plan ==
* HashAggregate (final)                <- *(2): fused
+- Exchange hashpartitioning(active)   <- shuffle boundary (NOT fused)
   +- * HashAggregate (partial)        <- *(1): fused
      +- * Filter (amount > 20)        <- *(1): fused into the scan stage
         +- * ColumnarToRow            <- *(1)
            +- Scan parquet            <- vectorised columnar read
*/

Step-by-step explanation.

1. The Scan parquet + ColumnarToRow + Filter + partial HashAggregate all carry the *(1) marker: Tungsten fused them into one generated method, so a row is read, filtered, and partially aggregated in a single tight loop with no intermediate object materialisation.
2. Exchange hashpartitioning(active) is the shuffle — it cannot be fused because it moves data across the network between executors. This is the stage boundary; everything above it runs in a new task set on (possibly) different executors.
3. The final HashAggregate gets a fresh *(2) marker: it is fused on the read side of the shuffle, combining the partial aggregates into final results.
4. The partial-then-final aggregation split is a map-side combine: Spark aggregates locally before the shuffle so it ships far fewer rows across the network. This is a serialization win (section 3) and a shuffle-size win at once.
5. If you added a Python UDF in the middle — say .withColumn("bucket", my_udf($"amount")) — the operators around it would lose the * marker, because the row must leave the JVM for the Python worker, breaking the fused loop. That's the visual signature of falling off the Tungsten path.

Output.

Operator	Codegen?	Reason
Scan parquet / ColumnarToRow	yes *(1)	vectorised columnar read
Filter amount > 20	yes *(1)	fused into scan stage
HashAggregate (partial)	yes *(1)	map-side combine before shuffle
Exchange hashpartitioning	no	shuffle crosses the network
HashAggregate (final)	yes *(2)	fused on the read side

Rule of thumb. Read explain("formatted") and treat every *(n) as "Tungsten fused this into fast generated code" and every un-starred operator as either a legitimate shuffle boundary or a codegen breaker to investigate. A UDF or typed lambda that strips the * off nearby operators is a performance smell.

Scala/Spark interview question on Tungsten

A senior interviewer might ask: "A colleague rewrote a DataFrame aggregation as an RDD of case classes with a reduceByKey, expecting it to be faster because 'RDDs are lower level.' It's actually 3× slower and GC time went from 5% to 35%. Explain in JVM terms why, and show the fix — including how off-heap memory changes the picture."

Solution Using the DataFrame/Tungsten path with off-heap execution memory

// SLOW: RDD of case-class objects — every record is a JVM object graph,
// reduceByKey shuffles serialized objects, GC scans millions of short-lived rows.
case class Sale(region: String, amount: Double)
val slow = sc.textFile("s3://sales/")
  .map { line =>
    val p = line.split(",")
    (p(0), p(1).toDouble)                 // Tuple2 + boxed Double per record
  }
  .reduceByKey(_ + _)                     // shuffles JVM objects (Java serialization by default)
  .collect()

// FAST: DataFrame path — rows stay as UnsafeRow, aggregation is codegen'd,
// map-side partial aggregate shrinks the shuffle, and off-heap keeps buffers off the GC heap.
val spark = org.apache.spark.sql.SparkSession.builder()
  .appName("tungsten-fix")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .config("spark.memory.offHeap.enabled", "true")
  .config("spark.memory.offHeap.size", "4g")   // Tungsten binary rows live here, GC never sees them
  .getOrCreate()
import spark.implicits._
import org.apache.spark.sql.functions._

val fast = spark.read
  .schema("region STRING, amount DOUBLE")
  .csv("s3://sales/")
  .groupBy($"region")
  .agg(sum($"amount").as("total"))        // whole-stage codegen + partial/final aggregate

# spark-submit flags that make the fast path even faster on a GC-bound job
--conf spark.memory.offHeap.enabled=true
--conf spark.memory.offHeap.size=4g
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer
--conf spark.executor.extraJavaOptions=-XX:+UseG1GC

Step-by-step trace.

Step	RDD/case-class path	DataFrame/Tungsten path
Row representation	Tuple2 + boxed Double objects	UnsafeRow binary buffer
Aggregation	reduceByKey over objects	codegen'd partial + final HashAggregate
Shuffle payload	Java-serialized objects	compact rows (Kryo), map-side combined
GC exposure	millions of short-lived objects	rows off-heap; GC idle
CPU behaviour	pointer chasing, virtual dispatch	contiguous scan, fused loop

Tracing the two on the same 500M-row input: the RDD path allocates a Tuple2 and a boxed Double per record, so the young generation fills constantly and GC time climbs to 35%; the shuffle ships full serialized objects. The DataFrame path represents each record as an UnsafeRow, does a map-side partial aggregate so the shuffle carries one row per region per partition (kilobytes, not gigabytes), and — with off-heap enabled — holds the aggregation buffers outside the heap so the GC has almost nothing to scan.

Output:

Metric	RDD/case-class	DataFrame/Tungsten + off-heap
GC time fraction	~35%	~5%
Shuffle write	~120 GB	~0.4 GB (map-side combined)
Peak heap pressure	high (object churn)	low (rows off-heap)
Wall-clock	3× baseline	1× baseline

Why this works — concept by concept:

• UnsafeRow binary format — representing a row as one contiguous byte buffer removes per-field object headers and pointer chasing, so the same data occupies less memory and scans cache-efficiently; the RDD-of-objects path pays the full object tax.
• Whole-stage codegen — fusing filter/aggregate into one generated loop eliminates the per-row virtual next() dispatch of the iterator model, which is why the DataFrame aggregate beats a hand-written reduceByKey.
• Map-side partial aggregate — the codegen'd partial HashAggregate combines locally before the shuffle, cutting shuffle bytes by orders of magnitude; the RDD reduceByKey combines too, but ships fat serialized objects instead of compact rows.
• Off-heap execution memory — placing Tungsten buffers outside the GC-managed heap means the collector never scans them, which is the direct cause of the GC-time collapse from 35% to 5% on allocation-heavy stages.
• Cost — off-heap adds a fixed reservation (offHeap.size) that counts against the container, so you trade a slice of container budget for near-zero GC on those buffers. Net effect on a GC-bound aggregate: same hardware, ~3× throughput, O(regions) shuffle instead of O(rows).

Optimization
Topic — optimization
Optimization problems on execution-engine tuning

Practice →

Data Processing Topic — data-processing Data-processing problems on aggregation at scale

Practice →

---

3. Serialization — Java vs Kryo, registering classes, why serialization dominates shuffle cost

Every shuffle, cache, and broadcast serializes — so the serializer you choose is a tax you pay on every byte that crosses a boundary

The mental model in one line: serialization is the conversion of live JVM objects into a byte stream (and back) that Spark performs every single time data must leave one JVM's heap — a shuffle write/read, a serialized cache level, a broadcast variable, a task result — and because Java's default serializer writes verbose type metadata (full class names) into the stream while kryo serialization writes compact integer class IDs, switching to Kryo and registering your hot classes routinely cuts shuffle bytes by 2–4× and serialization CPU by more, with two config lines and a registration list. Serialization is the most under-appreciated Spark bottleneck because it is invisible in the code — nothing in your transformation says "serialize here" — yet it runs on the hottest path in the whole system.

Where serialization happens (the hot paths).

• Shuffle. Between a map stage and a reduce stage, every record is serialized to shuffle files on the map side and deserialized on the reduce side. On a wide join or aggregation this is billions of serialize/deserialize calls.
• Cache with a _SER level. persist(MEMORY_ONLY_SER) stores blocks as serialized bytes (smaller, GC-cheaper) instead of live objects — you trade CPU (deserialize on read) for memory and GC.
• Broadcast. A broadcast variable is serialized on the driver, shipped, and deserialized once per executor. A slow serializer makes a broadcast join's setup slow.
• Task closures and results. The closure you pass to map/foreach is serialized to ship to executors; task results (small) serialize back to the driver.

Java serialization vs Kryo.

• Java (JavaSerializer, the default). Uses java.io.Serializable. Correct and zero-config, but writes the fully-qualified class name and a lot of type metadata into the stream, and is comparatively slow. It's the default only for backward compatibility.
• Kryo (KryoSerializer). Third-party, much faster, and much smaller — but it doesn't write full class names if you register your classes; instead each registered class gets a small integer ID. Unregistered classes still work under Kryo but fall back to writing the class name once, losing much of the benefit.
• The registration lever. spark.kryo.registrationRequired=true turns "unregistered class" into a hard error at runtime — a deliberately strict mode that guarantees no hot class silently falls back to the fat encoding. You register the record types, key types, and any custom classes on the shuffle path.

Why registration matters so much.

• Class name vs int ID. com.acme.model.TransactionRecord is ~30 bytes as a string, written per object under naive encoding; as a registered class it's a 1–2 byte varint ID. On 10^9 shuffled records that difference alone is tens of GB.
• Determinism and safety. registrationRequired=true makes an unregistered class fail loudly in dev/CI instead of silently bloating production shuffle. It's the config that separates "we turned on Kryo" from "we turned on Kryo correctly."
• Custom serializers. For a hot, oddly-shaped class you can register a hand-written Kryo serializer that encodes only the fields that matter — the last mile of shuffle-size tuning.

Worked example — measuring the shuffle-size drop from Kryo + registration

Detailed explanation. The canonical demonstration: run the same shuffle-heavy job three ways — Java serialization, Kryo without registration, Kryo with registration — and read Shuffle Write from the Spark UI. This is the exact experiment an interviewer wants you to have run.

• Workload. A groupByKey/join over 200M records of a custom Event class with a few string and numeric fields.
• Three configs. Java (default), Kryo unregistered, Kryo registered.
• Metric. Shuffle Write bytes and serialization time.

Question. Configure the three serializer setups and predict the relative shuffle-write sizes.

Input.

Setup	Class encoding on the wire	Expected shuffle size
Java serializer	full class name + metadata per object	1.0× (baseline, largest)
Kryo, unregistered	class name once, then Kryo body	~0.7×
Kryo, registered	int class ID + Kryo body	~0.3–0.4×

Code.

import org.apache.spark.sql.SparkSession
import org.apache.spark.serializer.KryoRegistrator
import com.esotericsoftware.kryo.Kryo

// Domain classes on the shuffle path
case class Event(userId: Long, kind: String, amount: Double, ts: Long)
case class UserKey(userId: Long)

// Central registrator so every hot class gets a small integer ID
class AppKryoRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo): Unit = {
    kryo.register(classOf[Event])
    kryo.register(classOf[UserKey])
    kryo.register(classOf[Array[Event]])
    kryo.register(classOf[scala.collection.mutable.WrappedArray.ofRef[_]])
  }
}

val spark = SparkSession.builder()
  .appName("kryo-shuffle")
  // 1. turn on Kryo
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  // 2. point Kryo at the registrator (int IDs instead of class-name strings)
  .config("spark.kryo.registrator", "com.acme.AppKryoRegistrator")
  // 3. STRICT: fail loudly if any hot class hits the shuffle unregistered
  .config("spark.kryo.registrationRequired", "true")
  // 4. bigger Kryo buffer for wide rows (avoids "buffer overflow" errors)
  .config("spark.kryoserializer.buffer.max", "128m")
  .getOrCreate()

# The same three setups as spark-submit flags (language-agnostic engine config):
#
# Java (baseline):
#   (no serializer flag -> JavaSerializer)
#
# Kryo, unregistered:
#   --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
#
# Kryo, registered + strict:
#   --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
#   --conf spark.kryo.registrator=com.acme.AppKryoRegistrator
#   --conf spark.kryo.registrationRequired=true
#   --conf spark.kryoserializer.buffer.max=128m

Step-by-step explanation.

1. Setup A (Java) is the baseline: every serialized Event on the shuffle path carries com.acme.Event as a string plus Java's stream metadata. On 200M records that metadata is a large fraction of the shuffle bytes.
2. Setup B (Kryo, unregistered) is faster than Java and a bit smaller, but because the classes aren't registered, Kryo still writes the class name (once per class per stream, better than Java but not optimal) and can't use its most compact encodings for your types.
3. Setup C (Kryo, registered) is the real win: each of Event, UserKey, and the array/collection wrappers gets a small integer ID, so the wire encoding is [varint classId][packed fields] with no class-name strings. This is the 0.3–0.4× shuffle-size regime.
4. registrationRequired=true is what makes setup C stay correct: if a new class hits the shuffle without being registered, the job fails immediately in CI with Class is not registered: com.acme.NewThing, instead of silently regressing to the setup-B encoding in production.
5. spark.kryoserializer.buffer.max must exceed your largest single serialized record; the classic Kryo error com.esotericsoftware.kryo.KryoException: Buffer overflow means a wide row didn't fit — raise this, don't disable Kryo.

Output.

Setup	Shuffle write (relative)	Serialization CPU (relative)	Correctness guard
Java serializer	1.00×	1.00×	none
Kryo, unregistered	~0.70×	~0.5×	none
Kryo, registered + strict	~0.35×	~0.35×	fails on unregistered class

Rule of thumb. Turning on Kryo without registering classes captures maybe half the benefit; the other half — and the correctness guarantee — comes from spark.kryo.registrator plus spark.kryo.registrationRequired=true. Register every class that rides the shuffle, cache, or broadcast path.

Worked example — a custom Kryo serializer for a hot class

Detailed explanation. When one class dominates the shuffle and its default Kryo encoding is wasteful (e.g. it carries a fat field you can recompute, or an enum stored as a string), a hand-written Kryo serializer is the last-mile lever. This is an advanced but interview-relevant technique.

• Target. A GeoPoint(lat: Double, lon: Double, label: String) where label is redundant downstream.
• Custom serializer. Write only lat and lon; drop label on the wire and reconstruct it lazily.
• Effect. Shuffle payload per point drops from ~40 bytes to 16 bytes.

Question. Write and register a custom Kryo serializer that encodes only the two doubles for GeoPoint.

Input.

Field	Default Kryo bytes	Custom serializer bytes
lat (Double)	8	8
lon (Double)	8	8
label (String)	~len+overhead	0 (dropped)

Code.

import com.esotericsoftware.kryo.{Kryo, Serializer}
import com.esotericsoftware.kryo.io.{Input, Output}

case class GeoPoint(lat: Double, lon: Double, label: String)

// Custom serializer: write only the coordinates; reconstruct label lazily.
class GeoPointSerializer extends Serializer[GeoPoint] {
  override def write(kryo: Kryo, out: Output, gp: GeoPoint): Unit = {
    out.writeDouble(gp.lat)
    out.writeDouble(gp.lon)              // label deliberately NOT written
  }
  override def read(kryo: Kryo, in: Input, tpe: Class[GeoPoint]): GeoPoint = {
    val lat = in.readDouble()
    val lon = in.readDouble()
    GeoPoint(lat, lon, label = "")       // label recomputed downstream if needed
  }
}

class GeoKryoRegistrator extends org.apache.spark.serializer.KryoRegistrator {
  override def registerClasses(kryo: Kryo): Unit = {
    kryo.register(classOf[GeoPoint], new GeoPointSerializer)   // bind the custom serializer
  }
}

// Wire it up:
//   spark.serializer                = org.apache.spark.serializer.KryoSerializer
//   spark.kryo.registrator          = com.acme.GeoKryoRegistrator
//   spark.kryo.registrationRequired = true

Step-by-step explanation.

1. write emits only the two 8-byte doubles, so each GeoPoint is exactly 16 bytes on the wire regardless of how long label is — a fixed, predictable shuffle payload.
2. read reconstructs the object with an empty label; downstream code that needs the label recomputes it (e.g. a reverse-geocode lookup) rather than paying to shuffle it billions of times.
3. Registering the class with the serializer instance (kryo.register(classOf[GeoPoint], new GeoPointSerializer)) binds the custom codec; without the second argument Kryo would use its reflective default and still serialize label.
4. This is only worth doing for a genuinely hot class — profile first (Shuffle Write per stage) and reserve custom serializers for the one or two classes that dominate. Over-engineering serializers for cold classes is wasted effort and a maintenance liability.
5. Because registrationRequired=true is on, forgetting to register GeoPoint would fail the job immediately — which is exactly the safety you want when hand-rolling serializers.

Output.

Metric	Default Kryo	Custom serializer
Bytes per GeoPoint on shuffle	~40	16
Shuffle write (1B points)	~40 GB	~16 GB
CPU per record	reflective	direct field writes
Trade-off	none	label recomputed downstream

Rule of thumb. Reach for a custom Kryo serializer only after the Spark UI proves one class dominates shuffle write. When you do, encode the minimum fields needed downstream and keep registrationRequired=true so the binding can never silently fall back.

Data-processing interview question on serialization

A senior interviewer might ask: "A broadcast-hash join in one of our jobs spends most of its time in the broadcast phase, and the shuffle for a downstream aggregation is 4× the input size. The team's fix was to add executors, which barely helped. Explain in serialization terms what's happening, and give the config + code that actually fixes it."

Solution Using Kryo with registration, strict mode, and a serialized cache level

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

val spark = SparkSession.builder()
  .appName("serialization-fix")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .config("spark.kryo.registrator", "com.acme.AppKryoRegistrator")
  .config("spark.kryo.registrationRequired", "true")
  .config("spark.kryoserializer.buffer.max", "256m")     // wide broadcast rows fit
  // broadcast the SMALL side explicitly and keep it below a sane threshold:
  .config("spark.sql.autoBroadcastJoinThreshold", "64m")
  .getOrCreate()
import spark.implicits._

val facts = spark.read.parquet("s3://events/")          // large
val dims  = spark.read.parquet("s3://dim_users/")       // small enough to broadcast

// Cache the reused dimension as SERIALIZED bytes: smaller + Kryo-encoded + GC-cheap
val dimsCached = dims.persist(StorageLevel.MEMORY_ONLY_SER)
dimsCached.count()  // materialise

val joined = facts.join(broadcast(dimsCached), Seq("user_id"), "left")
  .groupBy($"kind")
  .agg(sum($"amount").as("total"))

# Equivalent engine-level flags for a PySpark submit of the same job:
#   --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
#   --conf spark.kryo.registrator=com.acme.AppKryoRegistrator
#   --conf spark.kryo.registrationRequired=true
#   --conf spark.kryoserializer.buffer.max=256m
#   --conf spark.sql.autoBroadcastJoinThreshold=64m
# (PySpark rows in DataFrames are already Tungsten UnsafeRows; Kryo governs the
#  cache/broadcast/RDD paths and any Python-object closures on the JVM boundary.)

Step-by-step trace.

Step	Before (Java serializer)	After (Kryo + registered)
Broadcast encode	full class names + metadata	int class IDs, compact
Broadcast size	large; slow ship	~0.35×; fast ship
Downstream shuffle write	4× input	~1.4× input
Dimension cache	live objects (fat, GC)	MEMORY_ONLY_SER Kryo bytes
Effect of adding executors	marginal (bottleneck is bytes/CPU)	not needed

Tracing the fix: the original job was serialization-bound, not compute-bound, so adding executors barely moved it — each executor still paid the fat-encoding tax. Switching to Kryo with a registrator shrinks the broadcast payload (faster broadcast phase) and the downstream shuffle write (from 4× to ~1.4× of input). Caching the reused dimension as MEMORY_ONLY_SER stores it as compact Kryo bytes instead of live objects, cutting both its memory footprint and the GC it would otherwise cause.

Output:

Metric	Before	After
Broadcast phase time	dominant	~1/3
Downstream shuffle write	4× input	~1.4× input
Dimension cache footprint	100% (objects)	~35% (_SER Kryo)
Executors needed	more (didn't help)	same/fewer

Why this works — concept by concept:

• Kryo integer class IDs — replacing per-object class-name strings with 1–2 byte registered IDs is the direct cause of the shuffle- and broadcast-size drop; the encoding is [classId][packed fields] instead of [className][metadata][fields].
• registrationRequired = strict correctness — forcing every hot class to be registered guarantees no class silently falls back to the fat encoding, so the win is durable across code changes, not a one-time fluke.
• MEMORY_ONLY_SER cache — storing the reused dimension as serialized Kryo bytes trades a little deserialize-on-read CPU for a much smaller, GC-invisible cached footprint, which is exactly right for a hot, broadcast-sized table.
• Adding executors can't fix a bytes/CPU tax — because the bottleneck was serialization work per byte, horizontal scaling only adds more JVMs each paying the same tax; the fix has to lower the per-byte cost, which Kryo does.
• Cost — Kryo adds a registration list to maintain and a strict-mode failure surface in CI, in exchange for O(0.35) shuffle/broadcast bytes and proportionally lower serialization CPU. The maintenance cost is a few lines per new hot class; the payoff is paid on every shuffled byte for the life of the job.

Data Processing
Topic — data-processing
Data-processing problems on shuffle + serialization

Practice →

ETL Topic — etl ETL problems on join + broadcast pipelines

Practice →

---

4. Executor memory model — unified memory, execution vs storage, off-heap, overhead, spill

An executor's memory is a budget carved into reserved, user, and a unified storage+execution pool — and knowing the split is how you predict spill and OOM

The mental model in one line: an executor's JVM heap is partitioned by the unified memory manager into a fixed reserved slice, a user-memory slice for your objects/UDF state, and a single unified pool that storage (cached blocks) and execution (shuffle/sort/join/aggregation buffers) borrow from each other dynamically — and around that heap sits off-heap memory (Tungsten buffers, optional) and memoryOverhead (JVM-internal, native, and Python memory that lives outside the heap but inside the container), so executor memory sizing is really the arithmetic of fitting all of those into the container limit while leaving execution enough room to avoid spill. Get this arithmetic right and jobs stop OOMing; get it wrong and you either waste RAM or watch YARN/K8s kill containers.

The heap regions (Spark's unified memory manager).

• Reserved memory. A fixed 300 MB carved off the top, held back so Spark's own internals never starve. You don't tune it; you just subtract it.
• Usable memory. (spark.executor.memory − 300 MB). Everything below is a fraction of this.
• Unified region (storage + execution). spark.memory.fraction (default 0.6) of usable memory is the unified pool. Within it, spark.memory.storageFraction (default 0.5) sets the baseline split, but the two sides borrow from each other: execution can evict cached (storage) blocks when it needs room, and storage can grow into unused execution space. Execution memory is "greedier" — it can force-evict storage, but storage cannot evict active execution buffers.
• User memory. The remaining (1 − spark.memory.fraction) (default 0.4) of usable memory holds your own data structures, RDD-lineage bookkeeping, and non-Tungsten objects. UDFs that build big hash maps live here — and can OOM here even when the unified pool has room.

Outside the heap.

• Off-heap memory. spark.memory.offHeap.enabled=true + spark.memory.offHeap.size gives Tungsten a slab of memory outside the JVM heap for its binary buffers. It participates in the execution/storage accounting but is GC-invisible. It counts against the container total.
• Memory overhead. spark.executor.memoryOverhead (default max(384 MB, 0.10 × executor.memory)) is a non-heap reservation for JVM metadata, thread stacks, native libraries, netty buffers, and — critically — PySpark's Python worker processes. This is the number that, when too small, produces Container killed by YARN for exceeding memory limits even though the heap never threw OutOfMemoryError.
• The container total. YARN/K8s enforces executor.memory + offHeap.size + memoryOverhead ≤ container limit. Blow that and the resource manager kills the container regardless of what the heap thinks.

Spill — the normal safety valve.

• What it is. When an execution operator (sort, aggregate, join build) needs more memory than the unified pool can give, it writes sorted/partitioned runs to local disk and continues — this is spill, and it is a feature, not a crash. The job completes; it's just slower.
• Memory spill vs disk spill. The Spark UI shows both. "Shuffle Spill (Memory)" is the in-memory size of data that was spilled; "Shuffle Spill (Disk)" is its (usually smaller, serialized) on-disk size. Large disk spill = execution memory too small for the working set.
• When spill becomes a problem. A little spill is fine. Multi-GB, repeated spill means the operator is thrashing memory↔disk; the fix is more execution memory (bigger executor, higher memory.fraction, or off-heap), more partitions (smaller per-task working set), or reducing the working set (map-side combine, better join strategy).

Worked example — sizing an executor from first principles

Detailed explanation. The single most-tested memory skill is computing the actual execution-memory budget from an executor configuration. Interviewers give you --executor-memory and the fractions and ask "how much can a sort use before it spills?" Do the arithmetic explicitly.

• Config. --executor-memory 8g, defaults for memory.fraction=0.6, storageFraction=0.5.
• Goal. Compute reserved, usable, unified, baseline storage, baseline execution, and user memory.
• Then. Predict spill for a 3 GB sort working set.

Question. For an 8 GB executor with default fractions, compute every region and state whether a 3 GB sort spills.

Input.

Quantity	Formula	Value
executor.memory	given	8192 MB
reserved	fixed	300 MB
usable	mem − 300	7892 MB
unified	0.6 × usable	~4735 MB
baseline storage	0.5 × unified	~2368 MB
baseline execution	0.5 × unified	~2368 MB
user memory	0.4 × usable	~3157 MB

Code.

def executor_budget(executor_mem_mb: int,
                    memory_fraction: float = 0.6,
                    storage_fraction: float = 0.5,
                    reserved_mb: int = 300) -> dict:
    usable   = executor_mem_mb - reserved_mb
    unified  = usable * memory_fraction
    storage  = unified * storage_fraction
    execution = unified * (1 - storage_fraction)
    user     = usable * (1 - memory_fraction)
    return {
        "reserved_mb":         reserved_mb,
        "usable_mb":           round(usable),
        "unified_mb":          round(unified),
        "baseline_storage_mb": round(storage),
        "baseline_execution_mb": round(execution),
        "user_mb":             round(user),
        # execution can borrow ALL of unified if storage is empty:
        "max_execution_if_no_cache_mb": round(unified),
    }

b = executor_budget(8 * 1024)
for k, v in b.items():
    print(f"{k:32s}: {v}")
# baseline_execution_mb            : 2368
# max_execution_if_no_cache_mb     : 4735

Step-by-step explanation.

1. Subtract the fixed 300 MB reserved first: an 8 GB executor has 7892 MB usable, not 8192. This surprises people — the fractions apply to usable, not to executor.memory.
2. The unified pool is 0.6 × 7892 ≈ 4735 MB. This is the total that storage and execution share.
3. The storageFraction=0.5 split is only a baseline: it means storage is guaranteed ~2368 MB it can hold against eviction, and execution starts at ~2368 MB. But if nothing is cached, execution can borrow the entire 4735 MB.
4. So the 3 GB sort: if the executor is caching nothing, execution can grow to ~4735 MB, and 3 GB fits — no spill. If the executor is simultaneously holding ~2.4 GB of pinned cached blocks (storage using its protected baseline), execution is capped nearer ~2368 MB and the 3 GB sort spills the excess ~600+ MB to disk.
5. This is why "does it spill?" depends on what else the executor is doing: caching and execution compete for the same unified pool. The lever to stop spill is either more executor memory, a higher memory.fraction, fewer cached blocks competing, off-heap execution memory, or more partitions so each task's sort is smaller.

Output.

Region	8 GB executor	Notes
reserved	300 MB	fixed, untunable
usable	7892 MB	fractions apply here
unified (storage+exec)	~4735 MB	shared, borrowable
baseline execution	~2368 MB	grows to 4735 if no cache
user memory	~3157 MB	UDF maps, non-Tungsten objects
3 GB sort verdict	fits (no cache) / spills (with cache)	depends on storage pressure

Rule of thumb. Always subtract the 300 MB reserved first, then apply memory.fraction to what's left. Execution can borrow the whole unified pool only when nothing is cached; the moment you cache() a big table, you shrink the memory a concurrent sort/aggregate can use before it spills.

Worked example — the overhead OOM that isn't a heap OOM

Detailed explanation. The most misdiagnosed executor failure is Container killed by YARN for exceeding memory limits. X GB of Y GB physical memory used. This is not a heap OutOfMemoryError — the heap was fine; the container total (heap + off-heap + overhead + native + Python) exceeded the limit. The fix is memoryOverhead, not executor.memory.

• Symptom. Container killed, no java.lang.OutOfMemoryError in the executor log.
• Common causes. PySpark Python workers, large off-heap, native libs (e.g. a compression codec), too many netty shuffle buffers.
• Fix. Raise spark.executor.memoryOverhead (and/or spark.executor.pyspark.memory for Python), not the heap.

Question. Given a killed PySpark container, compute a corrected container budget and the flags to set.

Input.

Component	Before	Problem
executor.memory (heap)	8 GB	fine (no heap OOM)
memoryOverhead	0.8 GB (default 10%)	too small for Python workers
Python worker RSS	~1.5 GB	not counted in heap
container limit	9 GB	exceeded by Python + native

Code.

# BEFORE — container = 8g heap + 0.8g overhead = 8.8g, but Python workers
# push real RSS to ~9.5g -> "Container killed by YARN for exceeding memory limits".
--conf spark.executor.memory=8g
# (memoryOverhead defaults to max(384m, 0.10*8g) = ~820m -> too small for PySpark)

# AFTER — give the non-heap side explicit room for Python + native + netty:
--conf spark.executor.memory=8g
--conf spark.executor.memoryOverhead=3g          # JVM-internal + native + netty
--conf spark.executor.pyspark.memory=2g          # cap/budget the Python workers
# New container ask = 8 (heap) + 3 (overhead) ~= 11g; request an 11-12g container.

# Detecting which side blew up, from the executor log lines:
#   HEAP OOM  -> "java.lang.OutOfMemoryError: Java heap space"      => raise executor.memory
#   OVERHEAD  -> "Container killed by YARN for exceeding memory ...  => raise memoryOverhead
#   OFF-HEAP  -> "OutOfMemoryError: Cannot allocate ... off-heap"   => raise offHeap.size
#   DRIVER    -> OOM right after collect()/toPandas()               => raise driver.memory / avoid collect

Step-by-step explanation.

1. The tell is the absence of java.lang.OutOfMemoryError in the executor log combined with a YARN/K8s "container killed / exceeding memory limits" message. Heap was healthy; the container's total footprint wasn't.
2. In PySpark, each executor spawns Python worker processes whose memory lives entirely outside the JVM heap — it's counted in the container total but not in executor.memory. Default overhead (10%) rarely covers real Python workers, so the container quietly exceeds its limit.
3. The fix raises spark.executor.memoryOverhead to cover JVM-internal + native + netty, and optionally sets spark.executor.pyspark.memory to budget the Python side explicitly. The heap stays at 8 GB because the heap was never the problem.
4. The container request must grow to match: heap + overhead (+ offHeap). Raising overhead without asking the resource manager for a bigger container just moves the ceiling — you must request ~11–12 GB.
5. This is the single most common "we added executor memory and it didn't help" story: they raised the heap to fight an overhead OOM, which does nothing (and can make GC pauses worse).

Output.

Setting	Before	After
executor.memory (heap)	8 GB	8 GB (unchanged)
memoryOverhead	~0.8 GB	3 GB
pyspark.memory	unset	2 GB
container request	9 GB	~11–12 GB
Result	killed by YARN	stable

Rule of thumb. "Container killed for exceeding memory limits" with no heap OutOfMemoryError means overhead, not heap. Raise memoryOverhead (and pyspark.memory for Python jobs) and enlarge the container request — never reach for --executor-memory to fix an overhead kill.

Spark interview question on the executor memory model

A senior interviewer might ask: "You have a 200-node cluster with 64 GB / 16 cores per node. A join+aggregate job spills 40 GB to disk per run and occasionally loses executors to OOM. Walk me through how you'd size executors, set the memory fractions and overhead, and decide whether to enable off-heap — from first principles, not by guessing."

Solution Using first-principles executor sizing with off-heap and tuned fractions

# Node: 64 GB, 16 cores. Leave ~1 core + ~1 GB for the OS/NM daemons.
# Rule of thumb: ~5 cores/executor for good HDFS/shuffle throughput.
#   usable cores/node = 15  -> 3 executors/node (5 cores each)
#   usable mem/node   = 63 GB -> 21 GB per executor slot
# Split that 21 GB into heap + overhead (+ off-heap):

--num-executors 600                      # 200 nodes * 3 executors
--executor-cores 5
--executor-memory 15g                    # JVM heap
--conf spark.executor.memoryOverhead=3g  # native + netty + JVM-internal
--conf spark.memory.offHeap.enabled=true
--conf spark.memory.offHeap.size=3g      # Tungsten buffers off the GC heap
# container per executor ~= 15 + 3 + 3 = 21 GB  (fits the 21 GB slot)

# Attack the 40 GB spill:
--conf spark.memory.fraction=0.6         # keep unified pool generous
--conf spark.sql.shuffle.partitions=2000 # smaller per-task working set (was 200)
--conf spark.sql.adaptive.enabled=true   # AQE coalesces + handles skew
--conf spark.sql.adaptive.skewJoin.enabled=true
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer
--conf spark.executor.extraJavaOptions=-XX:+UseG1GC

# Verify the budget the config actually yields (reuse the earlier helper):
heap_mb = 15 * 1024
b = executor_budget(heap_mb)                 # from the sizing worked example
print("unified pool (MB):", b["unified_mb"])          # ~8815
print("baseline exec (MB):", b["baseline_execution_mb"])
# With off-heap 3g added, execution has ~8.8g heap-unified + 3g off-heap to work in
# before a single-task working set must spill. Raising shuffle.partitions from 200
# to 2000 cuts each task's working set ~10x, which is what actually kills the 40 GB spill.

Step-by-step trace.

Decision	Reasoning
5 cores/executor	balances HDFS/shuffle throughput vs too-many-tiny-executors overhead
3 executors/node	15 usable cores / 5 = 3; leaves a core for daemons
15g heap + 3g overhead + 3g off-heap	sums to the 21 GB slot; overhead covers native/netty
shuffle.partitions 200→2000	10× smaller per-task working set — the primary spill fix
off-heap on	large aggregation buffers leave the GC heap
AQE + skewJoin	coalesces small partitions, splits skewed ones

Tracing the spill fix: 40 GB of disk spill on 200 shuffle partitions means each task's working set is ~200 MB+ over what execution memory can hold. Cutting the working set per task ~10× (2000 partitions) is what actually eliminates most spill; off-heap and a generous memory.fraction give execution more room; AQE prevents a few skewed partitions from re-creating the spill on their own. Raising raw heap alone would not have fixed it — the working set per task was the lever.

Output:

Metric	Before	After
Disk spill / run	~40 GB	~2 GB
Executors lost to OOM	1–3	0
shuffle.partitions	200	2000
GC time fraction	~22%	~7%
Runtime	baseline	~0.5×

Why this works — concept by concept:

• Reserved-then-fraction arithmetic — sizing execution memory correctly starts from (heap − 300 MB) × memory.fraction, so you know the real budget a per-task sort has before it spills, instead of guessing from --executor-memory.
• Partitions as the working-set lever — spill is driven by per-task working-set size; raising shuffle.partitions shrinks each task's set linearly and is usually a bigger spill lever than adding heap.
• Off-heap for execution buffers — moving Tungsten aggregation/sort buffers off-heap gives execution more room without enlarging the GC-scanned heap, cutting both spill and GC together.
• Overhead sized for native/netty — a 3 GB overhead prevents the container-kill class of failure that a bigger heap would never fix, keeping executors alive under shuffle-buffer and native pressure.
• Cost — the config trades a fixed off-heap + overhead reservation (part of the 21 GB slot) and 10× more shuffle partitions (more, smaller tasks = slightly more scheduling overhead) for near-zero spill and zero OOM. Net: same hardware, ~2× throughput, O(working-set/partitions) spill instead of O(working-set).

Optimization
Topic — optimization
Optimization problems on memory + partition sizing

Practice →

Exception Handling Topic — exception-handling Exception-handling problems on failure recovery

Practice →

---

5. Debugging executors — OOM diagnosis, GC tuning, Spark UI, the interview signals

Debugging an executor is a fixed procedure — name the exact OOM, read the Spark UI signals, read the GC log, then apply the one lever the evidence points to

The mental model in one line: debugging executors is not guesswork — it is a fixed procedure that starts by classifying the failure (heap oom, overhead OOM, off-heap OOM, or driver OOM), then reads four Spark-UI signals (shuffle spill, GC time fraction, task-duration skew, peak execution memory) and, when needed, the garbage collection log, and only then applies the single lever the evidence indicates — because the wrong lever (usually "add heap") is not just useless but can make GC pauses longer. The senior signal in a Spark interview is walking this procedure out loud on a described failure without jumping to a fix.

The OOM taxonomy — four distinct failures that look alike.

• Heap OOM. Executor log shows java.lang.OutOfMemoryError: Java heap space (or GC overhead limit exceeded). The JVM heap genuinely filled. Fix: more heap, fewer/leaner objects, more partitions, or serialized caching.
• Overhead OOM. Resource manager shows Container killed ... exceeding memory limits, no heap OutOfMemoryError. Non-heap footprint (native, netty, Python) blew the container total. Fix: raise memoryOverhead / pyspark.memory, enlarge the container.
• Off-heap OOM. OutOfMemoryError: Cannot allocate ... off-heap — Tungsten's off-heap slab (offHeap.size) is too small for the working set. Fix: raise offHeap.size or reduce per-task working set.
• Driver OOM. OOM on the driver, typically right after collect(), toPandas(), a big broadcast, or a huge plan. Fix: don't bring big data to the driver; raise driver.memory/maxResultSize only as a guard.

The four Spark-UI signals.

• Shuffle Spill (Memory/Disk). Large disk spill → execution memory too small for the working set (section 4 levers).
• GC Time / Task Time. In the Executors tab, GC time as a fraction of task time; above ~20% you're allocation-bound (Kryo/Tungsten/off-heap, not more heap).
• Task-duration skew. In the Stages tab, compare max vs median (75th/max percentile) task time. Max ≫ median = data skew — one giant partition — which no amount of memory fixes; the lever is salting, AQE skew-join, or repartitioning.
• Peak Execution Memory. Per-task peak execution memory tells you how close a task came to spilling; near the pool size means you're on the edge.

GC tuning essentials.

• Use G1GC for large heaps. -XX:+UseG1GC is the sane default above a few GB; it targets bounded pause times and handles large heaps better than the old ParallelGC for Spark's allocation pattern.
• Read the GC log. -Xlog:gc* (JDK 11+) or -XX:+PrintGCDetails (JDK 8) shows pause durations and frequency. Frequent long pauses + rising heap-after-GC = the executor can't reclaim fast enough → reduce allocation, don't just enlarge the heap (a bigger heap = longer full-GC pauses).
• Don't oversize the heap. Beyond ~31–32 GB the JVM loses compressed oops (pointers become 8 bytes), so a 32 GB heap can hold fewer effective objects than a 31 GB one. Prefer more, smaller executors over a few giant ones.

Worked example — reading a GC-thrash + spill Spark UI

Detailed explanation. Given the numbers from a failing stage, walk the diagnosis to a single prescription. This is the interview exercise in its purest form: here are the signals, name the disease and the cure.

• Signals. GC 31% of task time, disk spill 18 GB, max task 9 min vs median 40 s, no container kills.
• Read. GC-bound AND spill-bound AND skewed — but skew is the primary driver here (max ≫ median).
• Prescribe. Fix skew first (it's causing both the spill and the GC on the giant task), then confirm.

Question. From the four signals, rank the causes and give the ordered fix.

Input.

Signal	Value	Reading
GC time / task time	31%	allocation pressure (secondary)
Disk spill	18 GB	execution memory short (secondary)
max vs median task	9 min vs 40 s	severe skew (PRIMARY)
container kills	0	not an overhead/off-heap OOM

Code.

from pyspark.sql import functions as F

# 1. CONFIRM skew: how lopsided are the join keys?
key_counts = (df.groupBy("join_key").count()
                .orderBy(F.desc("count")))
key_counts.show(10, truncate=False)
# e.g. one key has 240M rows; the next has 3M -> classic skew (that one key = the 9-min task)

# 2. FIX skew primarily with AQE skew-join (Spark 3+):
#    --conf spark.sql.adaptive.enabled=true
#    --conf spark.sql.adaptive.skewJoin.enabled=true
#    --conf spark.sql.adaptive.skewJoin.skewedPartitionFactor=5
#    --conf spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256m

# 3. FIX skew manually if AQE isn't enough — salt the hot key:
salted = (df.withColumn("salt", (F.rand() * 16).cast("int"))
            .withColumn("join_key_salted", F.concat_ws("_", "join_key", "salt")))
# join against a dimension exploded across the same 16 salt values, then drop salt.

# 4. Only AFTER skew is fixed, re-check GC + spill; if still high, THEN Kryo/off-heap/partitions.

Step-by-step explanation.

1. The dominant signal is max task 9 min vs median 40 s — a 13× ratio. That is not a memory shortage; it is one partition doing 200× the work of the others. Adding memory or executors would leave that one task just as slow.
2. Confirming with a key-count query shows a single join key with hundreds of millions of rows. That one giant partition is also why disk spill is 18 GB (its working set doesn't fit) and why GC is 31% (it allocates furiously) — the skew is the root cause of the other two signals.
3. The primary fix is AQE skew-join (spark.sql.adaptive.skewJoin.enabled=true), which detects the oversized partition at runtime and splits it into several sub-partitions joined independently — turning one 9-minute task into several 1-minute tasks.
4. If AQE isn't sufficient (e.g. an aggregation rather than a join, or an extreme single key), salt the hot key: append a random bucket to the key so the 240M-row key spreads across 16 tasks, then aggregate/join per bucket and combine.
5. Only after the skew is resolved do you re-read GC and spill. Frequently they drop on their own, because the giant task that drove both is gone. If residual GC/spill remains, then apply Kryo, off-heap, or more partitions — in that order.

Output.

Cause	Rank	Fix
Data skew (one hot key)	1 (primary)	AQE skew-join / salting
Disk spill	2 (symptom of #1)	resolves largely with #1; else more partitions/off-heap
GC pressure	3 (symptom of #1)	resolves largely with #1; else Kryo/off-heap
Container kill	n/a	none observed

Rule of thumb. When max task time ≫ median, fix skew first — it is usually the hidden root cause of the spill and GC you also see. Adding memory to a skewed job just makes the same giant task fail more slowly.

Worked example — the OOM decision tree

Detailed explanation. Codify the OOM classification as a decision tree so any described failure resolves to a lever in under a minute. This is the artifact interviewers reward — a reproducible procedure, not a hunch.

• Q1. Is there a java.lang.OutOfMemoryError: Java heap space in the executor log? → heap OOM.
• Q2. Is it a resource-manager "container killed / exceeding memory" with no heap OOM? → overhead OOM.
• Q3. Does the error say "off-heap" / "Cannot allocate"? → off-heap OOM.
• Q4. Did it happen on the driver right after collect/toPandas/broadcast? → driver OOM.

Question. Walk four described incidents through the tree and name the lever for each.

Input.

Incident	Log signature	Node
A	OutOfMemoryError: Java heap space in shuffle	executor
B	Container killed by YARN ... exceeding memory	executor
C	OutOfMemoryError: Cannot allocate ... off-heap	executor
D	OOM immediately after df.toPandas()	driver

Code.

def classify_oom(has_heap_oom: bool,
                 container_killed: bool,
                 mentions_offheap: bool,
                 after_collect_on_driver: bool) -> str:
    if after_collect_on_driver:
        return "DRIVER OOM -> avoid collect/toPandas; raise driver.memory/maxResultSize as guard"
    if mentions_offheap:
        return "OFF-HEAP OOM -> raise spark.memory.offHeap.size or cut per-task working set"
    if container_killed and not has_heap_oom:
        return "OVERHEAD OOM -> raise executor.memoryOverhead (+pyspark.memory); enlarge container"
    if has_heap_oom:
        return "HEAP OOM -> more heap OR fewer objects (Kryo/off-heap) OR more partitions OR _SER cache"
    return "not an OOM -> check skew (max vs median task) and shuffle spill"

print(classify_oom(True,  False, False, False))  # A
print(classify_oom(False, True,  False, False))  # B
print(classify_oom(False, False, True,  False))  # C
print(classify_oom(False, False, False, True))   # D

Step-by-step explanation.

1. Incident A has an explicit heap OutOfMemoryError during a shuffle → heap OOM. Levers in preference order: more partitions (smaller per-task set) and serialized caching first, more heap last. If the heap is already near 31 GB, add executors instead of enlarging it (compressed-oops cliff).
2. Incident B is a container kill with no heap OOM → overhead OOM. The heap was fine; native/netty/Python blew the container total. Raise memoryOverhead (and pyspark.memory for PySpark) and enlarge the container request — do not touch heap.
3. Incident C names off-heap explicitly → off-heap OOM. Tungsten's off-heap slab is too small for the working set. Raise spark.memory.offHeap.size or shrink the per-task working set (more partitions, map-side combine).
4. Incident D OOMs on the driver right after toPandas() → driver OOM. The lever is behavioural: don't pull big data to the driver. Raising driver.memory only buys headroom; the real fix is aggregating/writing distributed instead of collecting.
5. The tree's value is that each leaf points to a different config, and three of the four leaves are not "add executor heap." Saying "add memory" to all four is the exact anti-pattern interviewers screen for.

Output.

Incident	Classification	Lever
A	heap OOM	partitions / _SER cache / more heap (last)
B	overhead OOM	memoryOverhead + container size
C	off-heap OOM	offHeap.size / smaller working set
D	driver OOM	avoid collect/toPandas; guard with maxResultSize

Rule of thumb. Read the exact error string before choosing a lever: "Java heap space" (heap), "exceeding memory limits" (overhead), "off-heap" (off-heap), or "right after collect on the driver" (driver). Four failures, four different fixes — only one of which is more executor heap.

Spark interview question on debugging executors

A senior interviewer might ask: "A production PySpark job on a K8s Spark cluster keeps losing executors. The logs show a mix of Container killed ... exceeding memory and, on other runs, java.lang.OutOfMemoryError: Java heap space during a large join. GC time is around 28%. Walk me through the full debugging procedure end to end and the config you'd land on."

Solution Using the full executor-debugging procedure end to end

# STEP 1 — classify each failure from the exact log line.
#   "Container killed ... exceeding memory"   -> OVERHEAD OOM (raise memoryOverhead)
#   "java.lang.OutOfMemoryError: Java heap"   -> HEAP OOM during the join
#   GC ~28%                                    -> allocation-bound (reduce objects, G1GC)

# STEP 2 — read the Spark UI signals.
#   Stages tab: max task time vs median  -> check skew on the join key
#   Stages tab: Shuffle Spill (Disk)     -> execution memory vs working set
#   Executors tab: GC Time / Task Time   -> confirm the 28%

# STEP 3 — apply the ordered levers (config below).
--conf spark.executor.memory=12g
--conf spark.executor.memoryOverhead=4g          # fixes the OVERHEAD kills (Python + native)
--conf spark.executor.pyspark.memory=2g          # budget PySpark workers explicitly
--conf spark.memory.offHeap.enabled=true
--conf spark.memory.offHeap.size=3g              # join/agg buffers off the GC heap -> less GC
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer
--conf spark.kryo.registrationRequired=true
--conf spark.sql.shuffle.partitions=1600         # smaller per-task working set (kills heap OOM + spill)
--conf spark.sql.adaptive.enabled=true
--conf spark.sql.adaptive.skewJoin.enabled=true  # split the skewed join partition
--conf spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:MaxGCPauseMillis=200

# STEP 4 — verify each failure class is gone (REST API, from the triage in section 1):
import requests
APP = "http://spark-history:18080/api/v1/applications/app-latest"
execs  = requests.get(f"{APP}/executors").json()
stages = requests.get(f"{APP}/stages").json()

gc_frac = (sum(e["totalGCTime"] for e in execs if e["id"] != "driver")
           / max(sum(e["totalDuration"] for e in execs if e["id"] != "driver"), 1))
spill   = sum(s.get("diskBytesSpilled", 0) for s in stages) / 1e9
print(f"GC fraction now: {gc_frac:.1%}   disk spill now: {spill:.1f} GB")
# Expect GC fraction < 10% and spill < a few GB once the levers land.

Step-by-step trace.

Step	Action	Failure class addressed
1	classify from exact log line	overhead vs heap vs off-heap vs driver
2	read spill / GC / skew signals	pick primary vs secondary causes
3a	raise memoryOverhead + pyspark.memory	kills the container-exceeded OOMs
3b	shuffle.partitions up + AQE skewJoin	kills heap OOM + spill on the join
3c	off-heap + Kryo + G1GC	drops GC from 28%
4	re-read GC fraction + spill via API	confirm each class resolved

Tracing end to end: the overhead kills came from PySpark workers overflowing the default overhead, fixed by memoryOverhead=4g + pyspark.memory=2g and a matching container size. The heap OOM on the join came from too-large per-task working sets on a skewed key, fixed by raising shuffle.partitions to 1600 and enabling AQE skew-join to split the hot partition. The 28% GC came from object allocation during the join, dropped by off-heap execution buffers, Kryo, and G1GC. Each lever targets a specific failure class the classification step identified — nothing is thrown at the wall.

Output:

Metric	Before	After
Container-exceeded kills	frequent	0
Heap OOM on join	intermittent	0
GC time fraction	~28%	~8%
Disk spill	large	few GB
Runtime	unstable / long	stable, ~0.6×

Why this works — concept by concept:

• Classify before you tune — reading the exact error string routes each failure to its correct lever (overhead vs heap vs off-heap vs driver), so you never "add heap" to an overhead kill or a skew problem.
• memoryOverhead for the container total — sizing the non-heap reservation to cover Python + native + netty is the only thing that stops "exceeding memory limits" kills; heap changes are irrelevant to that failure class.
• partitions + AQE skew-join for heap OOM/spill — shrinking per-task working sets and splitting the skewed partition attacks the actual cause of the join's heap OOM and disk spill, which more heap would only delay.
• off-heap + Kryo + G1GC for GC — moving buffers off-heap and shrinking allocations lowers the collector's workload, and G1GC with a pause target keeps individual pauses bounded — the correct response to a 28% GC fraction.
• Cost — the procedure adds config surface and a slightly larger container request, in exchange for turning three intermittent, hard-to-reproduce failure classes into zero. It is O(1) analyst time per incident (read four numbers, classify, apply the mapped lever) versus an open-ended trial-and-error loop.

Exception Handling
Topic — exception-handling
Exception-handling problems on OOM + failure diagnosis

Practice →

Optimization
Topic — optimization
Optimization problems on GC + executor tuning

Practice →

---

Cheat sheet — Spark-on-the-JVM recipes

• Executors are JVMs — the four forces. Every Spark performance problem is one of: object overhead (heap layout), serialization (Java vs Kryo), memory split (unified pool + off-heap + overhead), or garbage collection. Read four numbers before tuning: cached-size vs on-disk-size (object overhead), shuffle write bytes (serialization), disk spill (memory split), GC time / task time (GC). Only one of the four is fixed by more RAM.
• Turn on Kryo correctly. spark.serializer=org.apache.spark.serializer.KryoSerializer, spark.kryo.registrator=<your registrator>, spark.kryo.registrationRequired=true, spark.kryoserializer.buffer.max=128m (raise for wide rows). Register every class that rides the shuffle/cache/broadcast path — registered classes get 1–2 byte int IDs instead of full class-name strings, cutting shuffle write ~2–4×.
• Enable off-heap for GC-bound stages. spark.memory.offHeap.enabled=true + spark.memory.offHeap.size=<N>g. Tungsten binary buffers move outside the GC-managed heap, so the collector never scans them — the direct fix for a high GC-time fraction on allocation-heavy aggregations/sorts. Remember off-heap counts against the container total.
• Executor memory arithmetic. usable = executor.memory − 300 MB (reserved); unified = 0.6 × usable (shared storage+execution); storage baseline = execution baseline = 0.5 × unified; user = 0.4 × usable. Execution can borrow the whole unified pool only when nothing is cached. Fractions apply to usable, not to executor.memory.
• OOM taxonomy — four failures, four fixes. java.lang.OutOfMemoryError: Java heap space → heap (partitions / _SER cache / more heap last). Container killed ... exceeding memory with no heap OOM → overhead (memoryOverhead + bigger container). Cannot allocate ... off-heap → off-heap (offHeap.size). OOM after collect/toPandas on the driver → driver (don't collect big data). Never answer "add heap" to all four.
• memoryOverhead sizing. Default max(384 MB, 0.10 × executor.memory) is too small for PySpark. Set spark.executor.memoryOverhead to cover native + netty + JVM-internal, and spark.executor.pyspark.memory to budget Python workers. Then enlarge the container request to heap + overhead + offHeap — raising overhead without a bigger container does nothing.
• Spill is a feature, not a crash. When execution memory is exhausted, sort/aggregate/join buffers spill sorted runs to disk and the job completes — just slower. A little spill is fine; multi-GB repeated spill means the per-task working set is too big. Fix with more spark.sql.shuffle.partitions (smaller working set), off-heap, or a higher memory.fraction — usually partitions first.
• Skew is not a memory bug. If max task time ≫ median (check the Stages tab), one giant partition is the problem and no amount of memory fixes it. Enable spark.sql.adaptive.enabled=true + spark.sql.adaptive.skewJoin.enabled=true, or salt the hot key. Fix skew before re-examining spill/GC — it's often the root cause of both.
• Stay on the Tungsten fast path. Prefer DataFrame/Column-expression ops over typed Dataset lambdas (ds.map(x => ...) deserializes each UnsafeRow to an object and back) and over non-vectorised Python UDFs (row leaves the JVM). Read explain("formatted"): *(n) markers = whole-stage-codegen'd; un-starred operators near a UDF = fell off the fast path.
• GC flags for large heaps. -XX:+UseG1GC -XX:MaxGCPauseMillis=200 is the sane default above a few GB. Log with -Xlog:gc* (JDK 11+) or -XX:+PrintGCDetails (JDK 8). Frequent long pauses + rising post-GC heap = reduce allocation, don't enlarge the heap — a bigger heap means longer full-GC pauses.
• Don't cross the 32 GB heap cliff. Above ~31–32 GB the JVM drops compressed ordinary object pointers (oops), so pointers become 8 bytes and a 32 GB heap can hold fewer effective objects than a 31 GB one. Prefer more, smaller executors (~5 cores, 15–20 GB) over a few giant ones — better parallelism and no oops cliff.
• Cache with the right storage level. MEMORY_ONLY_SER stores compact (Kryo) bytes — smaller and GC-cheap, at the cost of deserialize-on-read; ideal for a hot, reused, broadcast-sized table. MEMORY_AND_DISK (default for DataFrames) keeps deserialized rows and spills to disk under pressure. Never cache what you read once; only cache what you reuse.
• First-config-minute answer. "Spark executors are JVMs, so I'd classify the failure first — heap vs overhead vs off-heap vs driver — read shuffle spill, GC-time fraction, and task skew from the Spark UI, and only then apply the one lever the evidence points to: Kryo + registration for fat shuffles, off-heap + G1GC for GC pressure, more shuffle partitions for spill, AQE skew-join for skew, and memoryOverhead for container kills — never blanket --executor-memory."

Frequently asked questions

Why is Spark built on the JVM at all?

Spark runs on the JVM because it grew out of the Hadoop ecosystem (HDFS, YARN, the Java/Scala big-data stack) and Scala — a JVM language — gave it a concise functional API over that ecosystem with full Java interoperability. The consequence every data engineer inherits is that each Spark executor and the driver are JVM processes, so their performance is governed by JVM realities: object headers and pointers inflate in-memory data (jvm heap object overhead), moving data between processes requires serialization, a fixed heap must be partitioned between caching and computation (spark memory management), and dead objects are reclaimed by a garbage collection cycle that costs CPU and can pause the process. PySpark doesn't change this — the DataFrame engine still runs in the JVM; Python only drives it and runs UDFs in separate Python worker processes (whose memory lives in memoryOverhead, not the heap). Understanding spark on the jvm is what turns "the job is slow" into a specific, tunable JVM force.

What does Tungsten actually do?

tungsten is Spark's execution-engine rewrite whose job is to make Spark stop paying the JVM's object tax. Instead of representing each row as a graph of JVM objects (a Row wrapper, boxed numbers, a String that is itself a char[] plus a header), Tungsten stores rows in a compact binary layout called UnsafeRow — a single contiguous byte buffer with a null-tracking bitset, fixed-width inline slots, and a trailing variable-length region. It can place those buffers in off-heap memory (via sun.misc.Unsafe) so the garbage collector never scans them, it operates on them cache-efficiently (cache-aware sort/aggregate), and it uses whole-stage codegen to compile an entire stage's operators (filter, project, aggregate) into one generated Java method instead of a chain of iterator objects each doing a virtual next() call. The net effect is less memory per row, far less GC pressure, and CPU that behaves like hand-written code — which is why the DataFrame/Dataset/SQL APIs routinely beat equivalent RDD-of-objects code. You fall off the Tungsten fast path with typed Dataset lambdas and non-vectorised Python UDFs, both of which force rows back into JVM objects.

Java vs Kryo serialization — which should I use and why?

Use kryo serialization for essentially every non-trivial Spark job. Java serialization (the default, JavaSerializer) is correct and zero-config but writes the fully-qualified class name plus verbose type metadata into the byte stream for every object and is comparatively slow — so on a shuffle of a billion records it pays that metadata tax a billion times. Kryo (spark.serializer=org.apache.spark.serializer.KryoSerializer) is much faster and much smaller, especially when you register your classes via spark.kryo.registrator: a registered class is encoded as a 1–2 byte integer ID instead of a ~30-byte class-name string, which typically cuts shuffle write 2–4×. Turn on spark.kryo.registrationRequired=true so that any hot class hitting the shuffle unregistered fails loudly in CI instead of silently regressing to the fat encoding in production. Kryo governs the shuffle, serialized-cache, broadcast, and RDD-closure paths; DataFrame rows are already compact Tungsten UnsafeRows, so the biggest Kryo wins are on RDD workloads, serialized caching, broadcasts, and custom classes on those paths.

What is executor memory overhead and why does YARN kill my container?

executor memory overhead (spark.executor.memoryOverhead) is a reservation for memory that lives outside the JVM heap but inside the container: JVM-internal structures, thread stacks, native libraries, netty shuffle buffers, and — for PySpark — the Python worker processes. YARN/K8s enforces a container limit of roughly executor.memory (heap) + offHeap.size + memoryOverhead, and it kills the container the moment the total footprint exceeds that limit, even if the JVM heap never threw an OutOfMemoryError. That's why the message Container killed by YARN for exceeding memory limits with no heap OOM in the log is an overhead problem, not a heap problem — the fix is to raise memoryOverhead (and spark.executor.pyspark.memory for Python jobs) and enlarge the container request, not to raise --executor-memory. The default overhead of max(384 MB, 10% of heap) is frequently too small for PySpark because Python workers can easily use a gigabyte or more per executor that the heap accounting never sees.

What is a spill and is it bad?

A spill is Spark writing execution data to local disk when an operator (a sort, an aggregation, or a join's build side) needs more memory than the unified execution pool can give it. It is a safety valve, not a crash: the operator writes sorted/partitioned runs to disk and merges them, so the job still completes correctly — just slower. The Spark UI reports two numbers: "Shuffle Spill (Memory)" (the in-memory size of the spilled data) and "Shuffle Spill (Disk)" (its serialized on-disk size). A little spill is completely normal and not worth chasing. Multi-gigabyte, repeated spill is a signal — it means each task's working set is too large for the execution memory available, and the levers are, in rough priority: raise spark.sql.shuffle.partitions so each task handles a smaller slice, enable off-heap memory for execution buffers, raise spark.memory.fraction, or (last) enlarge the executor. First, though, check for skew — if one partition is huge, that giant task spills while the rest don't, and the real fix is AQE skew-join or salting, not more memory.

How do I debug a Spark OOM step by step?

debugging executors is a fixed procedure, not guesswork. Step 1 — classify the oom from the exact log line: java.lang.OutOfMemoryError: Java heap space is a heap OOM; Container killed ... exceeding memory limits with no heap OOM is an overhead OOM; Cannot allocate ... off-heap is an off-heap OOM; an OOM right after collect()/toPandas()/a big broadcast is a driver OOM. Step 2 — read four Spark-UI signals: shuffle spill (execution memory shortage), GC time / task time (allocation pressure, bad above ~20%), max-vs-median task duration (data skew), and peak execution memory. Step 3 — apply the one lever the evidence points to: more shuffle.partitions and serialized caching for heap OOM, memoryOverhead + a bigger container for overhead OOM, offHeap.size for off-heap OOM, avoiding collect for driver OOM, off-heap + Kryo + G1GC for high garbage collection, and AQE skew-join or salting for skew. Step 4 — re-read the signals to confirm the specific failure class is gone. The anti-pattern the whole procedure exists to prevent is answering "add more memory" to every failure — it's the correct fix for exactly one of the four OOM classes.

Practice on PipeCode

• Drill the optimization practice library → for the executor-sizing, memory-fraction, off-heap, and GC-tuning problems senior Spark interviewers love to open with.
• Rehearse on the data-processing practice library → for the aggregation-at-scale, shuffle, and serialization scenarios where Tungsten and Kryo earn their keep.
• Sharpen the pipeline axis with the ETL practice library → for the join, broadcast, and incremental-load patterns that stress the JVM memory model in production.
• Stress-test failure handling with the exception-handling practice library → for the OOM-taxonomy, container-kill, and spill-diagnosis reps that separate a fluent debugging answer from a guessing one.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-force JVM model (object overhead, serialization, memory split, garbage collection) against real graded inputs.

Lock in Spark-on-the-JVM muscle memory

Docs explain the flags. PipeCode drills explain the decision — when Tungsten's off-heap fixes a GC-bound stage, when Kryo registration shrinks a fat shuffle, when a container kill is overhead and not heap, when a spill is skew wearing a memory costume. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs data engineers actually face when a job OOMs at 3 a.m.

Practice optimization problems →
Practice data-processing problems →