Spark Tuning Cheat Sheet: Shuffle Partitions, Skew, Broadcast, Persist & Memory Fractions

spark performance tuning is the single most-asked Spark interview prompt in 2026 — and the one where most candidates lose senior signal because they reach for executor-memory first instead of measuring the actual bottleneck. The "my Spark job is slow" question is rarely about CPU; it's almost always about shuffle behaviour, data skew, join strategy, caching discipline, and memory fraction budgeting. Get those five axes right and a job that took 4 hours runs in 40 minutes on the same cluster.

This guide is the senior-DE cheat sheet you wished existed the first time an interviewer asked "how would you tune spark shuffle partitions?" or "your Spark job hits a spark out of memory — what's the first thing you check?" It walks through the shuffle-partition target-size rule and the AQE coalesceShufflePartitions runtime fix, spark skew detection through the Stage tab plus AQE skew-join split and salting fallbacks, broadcast join thresholds with the AQE dynamic SMJ → BHJ switch and the driver-OOM trap that breaks naive broadcast() hints, spark persist discipline (cache() vs persist(StorageLevel.MEMORY_AND_DISK) and the cache-then-count anti-pattern), and spark memory fractions (spark.memory.fraction, storage vs execution, off-heap toggles) for the workloads that don't fit on heap. Each section pairs a teaching block with a Solution-Tail interview answer — 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 ETL practice library →, rehearse on medium-difficulty ETL problems →, and sharpen the optimization axis with the optimization drills →.

---

On this page

• Why tuning is the senior-Spark interview filter
• Shuffle partitions — the most-misunderstood config
• Skew — the silent perf killer
• Broadcast joins + AQE dynamic switch
• Persist / cache + memory fractions
• Cheat sheet — Spark tuning recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why tuning is the senior-Spark interview filter

spark tuning is the senior-DE filter — the question is never "is it slow?" but "what did you measure, and which axis are you tuning?"

The one-sentence invariant: a Spark job's runtime is dominated by shuffle behaviour, skew, join strategy, caching discipline, and memory fractions — every other knob (executor count, partition count, parallelism level) is a downstream consequence of how you set those five axes. When an interviewer says "your Spark job is slow," they are not asking you to recite --executor-memory 16g; they are listening for a methodology that opens with measure first, tune second, then walks the five-axis model.

The five-axis tuning model.

• Shuffle. Every shuffle materialises data to disk, ships it over the network, reads it back. Tuning spark.sql.shuffle.partitions to target ~128-256 MB per partition is the highest-leverage knob in the API.
• Skew. One hot key holds up an entire stage — 99 tasks finish in 30s, one runs for 40 minutes. AQE skew-join split handles most cases since Spark 3.0; salting is the fallback for groupBy skew AQE cannot touch.
• Joins. BroadcastHashJoin (BHJ) beats SortMergeJoin (SMJ) by 10x when one side fits in driver memory. The trap is driver-OOM from broadcasting a "small" table that turns out to be 4 GB after decompression.
• Caching. cache() and persist() consume the storage half of the memory pool and can evict working data under pressure. The "cache then count" anti-pattern hides the cost in a phantom action.
• Memory. spark.memory.fraction = 0.6 by default; raise for cache-heavy workloads, lower for compute-heavy. Off-heap (spark.memory.offHeap.enabled = true + spark.memory.offHeap.size) takes pressure off the JVM heap for very large working sets.

Don't tune what you didn't measure.

• Spark UI first. Stages tab → task-duration histograms. SQL tab → physical plan and per-node metrics. Executors tab → GC time and heap occupancy.
• EXPLAIN FORMATTED second. Tells you whether the join is BroadcastHashJoin, SortMergeJoin, or ShuffledHashJoin, which side is broadcast, and whether AQE is enabled.
• spark.sql.adaptive.logLevel = INFO third. AQE logs every runtime decision — coalesce, skew split, dynamic SMJ → BHJ flip.

What 2026 interviewers actually probe.

• "Measure first, tune second" before reaching for a config knob — senior signal.
• AQE mentioned unprompted as the first defence against skew and over-partitioning — required.
• Pushback on "just bump executor memory" with a five-axis decomposition — senior signal.
• Spill vs OOM with the right vocabulary (spill is a controlled disk write; OOM is a JVM kill) — required.
• coalesce does not shuffle but can reduce upstream parallelism; repartition always shuffles — required.

The 2026 reality — what changed since Spark 2.x.

• AQE on by default since Spark 3.2 — most "manual tuning" advice from 2019 is obsolete.
• Coalesce shuffle partitions runs automatically; hand-tuning spark.sql.shuffle.partitions = 4096 is mostly gone.
• Dynamic SMJ → BHJ — AQE flips a SortMergeJoin to a BroadcastHashJoin at runtime if filters reduce one side below threshold.
• spark.memory.offHeap is mature and supported on every cluster manager.
• Disaggregated storage (S3/GCS/ADLS) has displaced HDFS — shuffle is no longer co-located with data, which makes shuffle tuning even more important.

Worked example — diagnose a 4-hour job in 15 minutes

Detailed explanation. A pipeline that processed 2 TB of input ran in 45 minutes last quarter and now takes 4 hours on the same cluster. The team's instinct is to double executor memory. The senior move is to open the Spark UI Stages tab, find the slow stage, and read the task-duration histogram before changing a single config.

Question. Walk through how you would diagnose a Spark job that suddenly slowed from 45 minutes to 4 hours on the same cluster, the same input volume, and the same code. What do you look at first, second, and third?

Input.

Symptom	Value
Last quarter runtime	45 min
Current runtime	4 hours
Input volume	2 TB (unchanged)
Cluster	50 executors × 8 cores × 32 GB
Code	unchanged

Code.

# Step 1 — confirm AQE is enabled (it usually is by default)
spark.conf.get("spark.sql.adaptive.enabled")            # expect "true"
spark.conf.get("spark.sql.adaptive.skewJoin.enabled")    # expect "true"

# Step 2 — open the Spark UI Stages tab
# Look for: stages where the max task duration >> the median.
# That is the long-tail skew signature.

# Step 3 — inspect the physical plan of the slow query
df.explain("formatted")

# Step 4 — find the offending join
spark.sql("""
  SELECT key, COUNT(*) AS rows
  FROM events
  GROUP BY key
  ORDER BY rows DESC
  LIMIT 20
""").show()

Step-by-step explanation.

1. Confirm AQE is enabled before changing anything. If spark.sql.adaptive.enabled is false, half the modern Spark tuning playbook is dormant.
2. Open the Spark UI Stages tab and rank stages by duration. The slowest stage is the bottleneck. Click into it and look at the task-duration histogram — even bars = healthy, one tall bar = skew.
3. Read the SQL tab to find the physical plan. The plan shows BroadcastHashJoin / SortMergeJoin / ShuffledHashJoin per join node, the shuffle read/write sizes, and the input row counts. A SortMergeJoin where one side is 50 MB and the other is 500 GB is a missed broadcast opportunity.
4. Inspect the keys for skew with a quick GROUP BY key ORDER BY count DESC query. If the top key has 100x the rows of the median key, you have a skew problem.
5. Compare the input volume by partition. The current 4-hour run probably has a new hot key (a country code went viral, a customer started spamming, a bot got through the front door). Identify the key, then decide between AQE skew-join, salting, or filtering the bot.

Output.

Step	Finding	Action
1	AQE is enabled	continue
2	Stage 4: max task 38 min, median task 12 sec	confirmed skew
3	SortMergeJoin on country_id	candidate for skew-join split
4	Top key US: 80% of rows	hot key confirmed
5	Salting + skew-join hint applied	runtime back to 50 min

Rule of thumb. Spend the first 15 minutes in the Spark UI before touching a single config. The slow stage tells you 80% of what you need to know; the physical plan tells you the rest.

Worked example — coalesce(1) vs repartition(1) write trap

Detailed explanation. A team writes 500 GB to S3, then realises they want a single output file for the downstream consumer. They use df.coalesce(1).write.parquet(...). The job that read 500 GB in 8 minutes now runs for 90 minutes on the single-partition write. The senior question is: what changed, and what should you have used instead?

Question. Explain why coalesce(1) before a write of 500 GB causes a job to slow to a crawl, and what the correct pattern is to produce a single output file without losing parallelism on the upstream stages.

Input.

Stage	Records	Time
Read source	500 GB / 200 partitions	4 min
Transformations	500 GB / 200 partitions	4 min
coalesce(1) + write	500 GB / 1 partition	90 min

Code.

# BAD — coalesce(1) collapses parallelism upstream too
df = spark.read.parquet("s3://src/")
df_clean = df.filter("status = 'ok'").withColumn("ts", to_timestamp("ts"))
df_clean.coalesce(1).write.parquet("s3://out/")   # all 500GB through 1 task

# BETTER — let Spark write many files in parallel, merge afterwards
df_clean.write.parquet("s3://out/")               # parallel write, many files

# BEST — write parallel, then merge with a separate compaction job
df_clean.repartition(20).write.parquet("s3://out/")  # 20 medium files

Step-by-step explanation.

1. coalesce(1) does not shuffle; it merges existing partitions into one. But because it is a narrow transformation that fixes parallelism, the entire upstream chain re-plans with parallelism = 1. The 500 GB read, the filter, and the timestamp conversion all run in one task on one executor.
2. repartition(1) would shuffle, but it would also force all 500 GB through one task at the shuffle exchange. Same OOM risk, plus a shuffle write to disk.
3. The correct pattern is to let Spark write many parallel files and then run a separate compaction job that merges them (Spark, AWS Glue compaction, Iceberg rewriteDataFiles, or Delta OPTIMIZE).
4. If you absolutely need one file (downstream consumer can't handle a directory), use df.repartition(1).write.parquet(...) but accept the 90-minute write time, and run the rest of the job before the repartition so the upstream stages stay parallel.
5. The reason coalesce(1) is the senior interview trap: it looks like "free" partition reduction (no shuffle) but it silently strangles upstream parallelism. The rule is coalesce only when you are merging fewer than 10x partitions and the rows fit comfortably in the surviving partitions.

Output.

Pattern	Upstream parallelism	Output files	Runtime
coalesce(1)	collapsed to 1	1	90 min
repartition(1)	parallel up to shuffle	1	60 min
Parallel write + compaction	full parallel	many → 1	12 min

Rule of thumb. Never use coalesce(N) to write to a single file on a job larger than 5 GB. Write parallel, compact separately.

Worked example — the cluster is fine, the job is slow

Detailed explanation. A common interview probe: "my cluster has 50 executors and the job uses 3 of them. What's happening?" The answer is almost always partition count is too low — Spark cannot run more tasks than it has partitions, and idle executors do nothing.

Question. Diagnose a Spark job that allocates 50 executors but only 3 are doing work. What is the cause, and what is the fix?

Input.

Metric	Value
Executors allocated	50 (8 cores each = 400 task slots)
Active tasks	3
Active executors	3
Input partitions	3
Job stage	shuffle read

Code.

# The job reads a Parquet file with only 3 partitions
df = spark.read.parquet("s3://small-parquet/")   # 3 partitions
print(df.rdd.getNumPartitions())                  # 3

# Fix 1 — repartition immediately after read
df = df.repartition(400)                          # match task slots

# Fix 2 — set spark.sql.files.maxPartitionBytes to split big files
spark.conf.set("spark.sql.files.maxPartitionBytes", 128 * 1024 * 1024)

# Fix 3 — for SQL, let AQE coalesce after a shuffle
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

Step-by-step explanation.

1. Spark allocates executors based on spark.dynamicAllocation.maxExecutors, but each task needs a partition to read. If the input has 3 partitions, only 3 tasks run in parallel, and only 3 cores are busy.
2. repartition(400) shuffles the 3 input partitions into 400 output partitions. Now 400 tasks can run, and the 50 executors × 8 cores = 400 slots are fully utilised.
3. spark.sql.files.maxPartitionBytes controls Spark's automatic partition sizing on file reads. The default is 128 MB. If your Parquet files are large, lowering this splits each file into more partitions on read, avoiding the manual repartition.
4. AQE's coalescePartitions will reduce partitions after a shuffle if they end up too small, but it never increases parallelism. The fix for under-partitioning has to happen at the read or via explicit repartition.
5. After fixing partition count, the Spark UI shows all 50 executors active and the stage completes in seconds instead of minutes.

Output.

Metric	Before	After
Active executors	3	50
Active tasks	3	400
Stage runtime	12 min	18 sec
Cluster utilisation	0.75%	100%

Rule of thumb. If executors are idle, the answer is more partitions, not more executors. Match partition count to total core count on the cluster as a starting point.

Senior interview question on Spark tuning methodology

A senior interviewer often opens with: "A team comes to you saying their Spark job is slow. You have 30 minutes. What is your methodology — what do you look at first, what do you change first, and what do you measure before declaring victory?"

Solution Using a five-axis measurement framework

30-minute Spark tuning playbook

Phase 1 — Measure (10 min)
1. Spark UI → Jobs tab → identify the slowest job
2. Stages tab of that job → identify the slowest stage
3. Inspect task-duration histogram for that stage
   - even bars         → not skew; suspect partition count or shuffle
   - one tall bar      → skew; identify the hot key
   - all bars too long → shuffle size; check spill
4. SQL tab → physical plan of the slow query
   - SortMergeJoin where one side is < 100MB → broadcast candidate
   - ShuffleExchange with large bytes        → shuffle target

Phase 2 — Tune one axis (10 min)
- if partition count off:   repartition / files.maxPartitionBytes
- if shuffle partitions:    spark.sql.shuffle.partitions or AQE coalesce
- if skew:                  AQE skewJoin → salt fallback
- if join strategy:         broadcast() hint or autoBroadcastJoinThreshold
- if cache thrash:          unpersist or raise memory.fraction

Phase 3 — Verify (10 min)
- re-run the offending stage
- compare task-duration histogram before/after
- compare stage runtime before/after
- check Executors tab for GC pressure

Step-by-step trace.

Symptom	Axis	First-line fix	AQE assist
Stage runtime dominated by one task	Skew	salt key + repartition	skewJoin.enabled
Stage runtime dominated by all tasks	Shuffle size	raise shuffle.partitions	coalescePartitions
Idle executors	Partition count	repartition or files.maxPartitionBytes	none
SMJ on small × huge	Join strategy	broadcast() or raise threshold	dynamic SMJ → BHJ
Frequent GC or OOM	Memory	memory.fraction or off-heap	none
Spill to disk	Shuffle/Memory	bigger executors or smaller partitions	coalescePartitions

Output:

Phase	Output
Measure	one slow stage identified + suspected axis
Tune	one axis changed at a time
Verify	task-histogram + stage runtime improved

Why this works — concept by concept:

• Measure-first, tune-second — every senior Spark tuning conversation starts in the Spark UI, not in spark-submit args. Without a metric, you cannot tell whether a config change helped or hurt.
• One axis at a time — changing shuffle partitions, broadcast threshold, and memory.fraction in the same deploy makes the result impossible to attribute. Move one knob, measure, then move the next.
• AQE handles 80% of the easy cases — if AQE is enabled and your data is normal-shaped, you may not need to tune spark.sql.shuffle.partitions at all. Disable AQE only when you know you need to.
• The Spark UI is the source of truth — task-duration histograms reveal skew, shuffle metrics reveal data movement cost, the SQL tab reveals join strategy. There is no substitute for reading the UI.
• Cost — Phase 1 measurement is O(1) cluster cost (just open the UI). Phase 2 tuning is O(1) per change. Phase 3 verification is one re-run, which is the only place real cost is spent.

ETL
Topic — etl
Spark tuning and ETL problems

Practice →

Optimization Topic — optimization Performance optimization problems

Practice →

---

2. Shuffle partitions — the most-misunderstood config

spark shuffle partitions is target-size first, not target-count — pick 128-256MB per partition and let AQE finish the job

The mental model in one line: spark.sql.shuffle.partitions is the number of output partitions after a shuffle, and the right number is "whatever produces 128-256 MB per partition for your data" — not 200, not 4096, not "double the cores," but a target-size-driven count that AQE further refines at runtime. Once you internalise "target size, not target count," every shuffle-partition interview becomes a one-line calculation: partitions = total_shuffle_bytes / 200_MB.

The default is wrong for most jobs.

• spark.sql.shuffle.partitions = 200 is the historical Hive-on-Spark default. It assumes 200 MB × 200 = 40 GB total shuffle; anything substantially different makes 200 wrong.
• Two failure modes: too few partitions on a 4 TB shuffle (200 × 20 GB = spill, OOM, skew); too many on a 1 GB shuffle (200 × 5 MB = scheduler overhead).
• The 2026 rule: target 128-256 MB per partition. Compute total_shuffle_bytes ÷ 200_MB and round to a nice number.

AQE coalesce — the runtime fix.

• spark.sql.adaptive.coalescePartitions.enabled = true lets AQE merge small partitions after a shuffle based on actual shuffle-write statistics.
• spark.sql.adaptive.advisoryPartitionSizeInBytes = 64MB is the default target; most teams raise it to 128 MB for throughput.
• AQE never increases partition count — it only merges. Setting spark.sql.shuffle.partitions to a high value and letting AQE coalesce down is safer than guessing a low value.

The three symptoms.

• Too few → spill + OOM + skew. Each partition is too large for execution memory; tasks spill, GC rises, one hot key OOMs.
• Too many → scheduler overhead. Driver spends more time scheduling than tasks spend computing; the UI shows millions of sub-second tasks.
• Just right → ~200 MB per partition. Tasks run for 10-60 seconds, no spill, executors at full utilisation.

The shuffle-partition formula.

• Estimate shuffle-write bytes from the Spark UI Stage tab.
• Compute shuffle_write_bytes ÷ 200_MB.
• Round to a multiple of cluster cores (50 × 8 = 400 cores → round to 400, 800, or 1200).
• Set spark.sql.shuffle.partitions = <that number> and enable AQE coalesce as the safety net.

The 2026 reality.

• AQE on by default since Spark 3.2 mitigates the worst case of spark.sql.shuffle.partitions = 200 by coalescing small partitions automatically.
• For very large shuffles (> 1 TB), keep the initial partition count high so spill doesn't happen before coalesce runs.
• The safest pattern: set spark.sql.shuffle.partitions high (e.g. 2000 for 1 TB) and let AQE coalesce down to ~5000 effective partitions of 200 MB.

Worked example — a 10 TB join sized correctly

Detailed explanation. A nightly join of two large tables produces a 10 TB shuffle. The default spark.sql.shuffle.partitions = 200 puts 50 GB per partition — each task spills repeatedly, runs for 90 minutes, and one task hits OOM and restarts. The fix is to size partitions to 200 MB.

Question. Given a join whose shuffle-write size is 10 TB, compute the right spark.sql.shuffle.partitions value and show the AQE config block that makes the job resilient if your estimate is wrong.

Input.

Metric	Value
Shuffle-write bytes	10 TB
Target partition size	200 MB
Cluster cores	800 (100 executors × 8)
Default shuffle.partitions	200

Code.

# Sizing math
total_shuffle = 10 * 1024 * 1024  # MB
target_size   = 200               # MB
partitions    = total_shuffle // target_size
# 10TB / 200MB = 51200 partitions

# Round to a multiple of cluster cores
spark.conf.set("spark.sql.shuffle.partitions", 51200)

# AQE block — the safety net if 200MB is wrong on this run
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(200 * 1024 * 1024))
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

Step-by-step explanation.

1. Compute the target count: 10 TB ÷ 200 MB = 51,200 shuffle partitions. This is the initial count Spark will request from the shuffle.
2. Set spark.sql.shuffle.partitions = 51200. Without AQE, this would be the final number, and a single skewed partition would still hurt. With AQE, this is the upper bound — AQE will coalesce down where partitions are small.
3. Enable AQE coalesce so that any sub-200 MB partition gets merged with its neighbour into a 200 MB partition. The advisoryPartitionSizeInBytes tells AQE the target.
4. Enable AQE skew-join split. If a partition turns out to be 2 GB (10x the median), AQE will split it into 10 sub-partitions for the join side. Combined with the right initial partition count, this is the senior-Spark "two-line shuffle fix."
5. After the change, the Spark UI shows ~50,000 tasks instead of 200, task runtimes drop to ~30 seconds each, no spill, and the job completes in 45 minutes instead of 4 hours.

Output.

Metric	Default (200)	Tuned (51,200 + AQE)
Partition size	50 GB	~200 MB
Task runtime	90 min (with spill)	~30 sec
Spill bytes	4 TB	0
Job runtime	4 hours	45 min
OOM restarts	2	0

Rule of thumb. Compute shuffle_bytes ÷ 200_MB to set spark.sql.shuffle.partitions. Always pair the explicit count with AQE coalesce enabled — that combination is robust to estimate error.

Worked example — too many partitions on a small job

Detailed explanation. A small reporting query shuffles 1 GB of data with spark.sql.shuffle.partitions = 4000. Each partition holds 250 KB. The driver spends most of the wall-clock time scheduling tasks; CPU on executors is < 5%. The fix is to let AQE coalesce — or lower the initial count.

Question. Diagnose a Spark job where the driver is the bottleneck on a 1 GB shuffle. Show the AQE config block that fixes scheduler overhead.

Input.

Metric	Value
Shuffle-write bytes	1 GB
spark.sql.shuffle.partitions	4000
Per-partition size	~250 KB
Executor CPU	< 5%
Driver CPU	100%

Code.

# Pure config fix — let AQE coalesce small partitions
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", str(64 * 1024 * 1024))
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(128 * 1024 * 1024))

# Alternative: lower the initial count directly
spark.conf.set("spark.sql.shuffle.partitions", 16)  # 1GB / 64MB = 16

Step-by-step explanation.

1. With 4000 partitions for 1 GB of data, each partition is ~250 KB. A 250 KB task runs in ~50 ms but takes ~100 ms to schedule and serialise — the driver is the bottleneck and executors idle.
2. AQE's coalescePartitions reads the per-partition write size after the shuffle and merges small partitions until each reaches advisoryPartitionSizeInBytes (128 MB target). For 1 GB total, that means AQE reduces 4000 → ~8 partitions.
3. Setting minPartitionSize = 64 MB is a safety floor — AQE will never produce a partition smaller than this even if the advisory size logic would prefer to.
4. Alternatively, you can just set spark.sql.shuffle.partitions = 16 if you know the data size at submit time. The AQE coalesce path is preferred because it adapts to actual data sizes — what if the table is 10 GB next week?
5. After the fix, the Spark UI shows ~8 tasks running, each for ~3 seconds, executor CPU at 70%+, and the job finishes in seconds instead of minutes.

Output.

Metric	Before (4000)	After (AQE coalesce)
Task count	4000	~8
Task runtime	50 ms (compute) + 100 ms (overhead)	~3 sec
Driver CPU	100% (scheduling)	10%
Executor CPU	< 5%	70%+
Wall clock	8 min	25 sec

Rule of thumb. Trust AQE coalesce on jobs of unknown size. Set spark.sql.shuffle.partitions only when you have a stable, predictable input volume and the math is easy.

Worked example — when to set partitions explicitly

Detailed explanation. A streaming pipeline with predictable hourly volume hits AQE coalesce-flap — a small variation in input causes AQE to choose 50 partitions one hour and 80 the next, slowing the downstream INSERT because the file count jumps. The fix is to disable AQE coalesce for this stage and pin partition count.

Question. When does AQE coalesce hurt rather than help? Show a streaming INSERT pipeline where pinning spark.sql.shuffle.partitions is preferable.

Input.

Metric	Value
Pipeline	hourly batch INSERT
Volume per hour	80-120 GB
Sink	Iceberg (file-count-sensitive)
AQE coalesce	enabled (default)

Code.

# Disable AQE coalesce for the write stage and pin partitions
spark.conf.set("spark.sql.adaptive.enabled", "true")               # AQE on for skew + dynamic switch
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "false")  # off for stable file count
spark.conf.set("spark.sql.shuffle.partitions", 500)               # 100GB / 200MB ≈ 500

# Iceberg sink with predictable file count
df.writeTo("warehouse.events").append()

Step-by-step explanation.

1. The downstream Iceberg consumer is sensitive to file count for query planning; an unpredictable file count (50 one hour, 80 the next) makes vacuum / compaction harder to size.
2. Disabling coalescePartitions while keeping AQE enabled (for skew-join and dynamic SMJ → BHJ) is a useful middle ground.
3. Setting spark.sql.shuffle.partitions = 500 (100 GB ÷ 200 MB) gives ~500 output files of ~200 MB each — a stable, predictable file count.
4. The trade-off: if the hourly volume swings 4x in one direction (say, 400 GB), each partition becomes 800 MB and you risk spill on the per-partition operations. For that case, the better pattern is to keep coalescePartitions on and accept variable file count.
5. The senior insight: AQE is not always-on by default for every stage. You can scope AQE behaviour by writing two spark.sql statements with different config blocks.

Output.

Setting	File count	File size	Iceberg compaction cost
AQE coalesce on	30-80	1-4 GB	unpredictable
AQE coalesce off + pinned 500	~500	~200 MB	stable

Rule of thumb. Disable AQE coalesce only when downstream consumers are file-count-sensitive (Iceberg, Hive partitioning, BigQuery external tables). Otherwise, AQE coalesce is a free win.

Senior interview question on shuffle partitions

A senior interviewer might ask: "You inherit a Spark job with spark.sql.shuffle.partitions = 4096 hard-coded. The job runs on inputs that range from 1 GB to 10 TB depending on the day. How do you redesign the config so the job is correct on both ends?"

Solution Using AQE coalesce with a high initial bound

# The redesign — AQE-first, single config block, no per-input fork

# 1. Trust AQE for the high end and the coalesce for the low end
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# 2. Target partition size — the actual invariant we care about
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(200 * 1024 * 1024))

# 3. Keep initial partitions HIGH — AQE only coalesces down
#    10TB / 200MB = 51200, so initial = 51200 is fine for the big day
spark.conf.set("spark.sql.shuffle.partitions", 51200)

# 4. AQE skewJoin to handle hot keys on the big day
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# 5. Min partition size so the small day doesn't make sub-MB partitions
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", str(64 * 1024 * 1024))

Step-by-step trace.

Input volume	Initial partitions	AQE coalesce	Final partitions
1 GB	51,200	merges to ~16	~16 (200 MB each)
100 GB	51,200	merges to ~500	~500 (200 MB each)
1 TB	51,200	merges to ~5,000	~5,000 (200 MB each)
10 TB	51,200	no merge	~51,200 (200 MB each)

The same single config block adapts to a 10,000x input swing because AQE coalesce drives toward the target size, not a target count.

Output:

Metric	Before (pinned 4096)	After (AQE + high initial)
1 GB run partition size	250 KB (overhead)	200 MB
10 TB run partition size	2.5 GB (spill)	200 MB
Manual per-input tuning	required	none
Driver scheduling cost (small)	high	low
Executor spill (large)	high	none

Why this works — concept by concept:

• Target size, not target count — the only invariant that matters is "how big is each partition after the shuffle." AQE optimises toward size; you tell it the target via advisoryPartitionSizeInBytes.
• High initial bound + AQE coalesce — AQE only reduces partition count. Setting the initial count high (for the 10 TB day) and letting AQE coalesce down (for the 1 GB day) is the only single-config pattern that handles both ends.
• Min partition size as a safety floor — without minPartitionSize, AQE could occasionally produce a partition under 1 MB. The floor prevents pathological small-partition cases.
• SkewJoin paired with coalesce — coalesce solves under-sized partitions; skewJoin solves over-sized partitions caused by hot keys. The pair covers both ends of the size distribution.
• Cost — AQE adds O(shuffle nodes) extra scheduling logic, but the savings (skipped spill, fewer tasks, balanced load) dwarf the overhead. The cost of not using AQE on a 4-hour job is hours; the cost of AQE is milliseconds.

ETL
Topic — etl · medium
Medium-difficulty ETL tuning problems

Practice →

Optimization Topic — optimization Shuffle and partition optimization drills

Practice →

---

3. Skew — the silent perf killer

spark skew is the long-tail task in the Stage tab — AQE skew-join handles most of it; salting is the surgical fallback

The mental model in one line: skew is when one partition has dramatically more data than the others — the Stage tab shows it as one long bar in the task-duration histogram, and the fix is either AQE skew-join split (for joins) or salting the key (for groupBy or pre-AQE Spark). Once you internalise "one task running 100x longer than the median = one hot key," every skew interview becomes a sequence of consequences from the histogram.

How to detect skew in 30 seconds.

• Open the Spark UI → Jobs tab → click the slow job → Stages tab.
• Click into the slow stage and scroll to the Summary Metrics for Tasks section.
• Compare 75th percentile to max. If the max is > 10x the 75th percentile for Duration or Shuffle Read Size, you have skew.
• Confirm with the task-duration histogram. Even bars + one tall bar = textbook skew. Even bars throughout = no skew (look elsewhere).

What causes skew in practice.

• Hot key in a groupBy. One country, one user, one product accounts for 80% of the rows. The groupBy puts all those rows on one partition.
• Hot key in a join. Same as above, except the matching rows on the other side are also concentrated. The join shuffles all matches into one task.
• NULL or sentinel values. A million rows with customer_id IS NULL all hash to the same partition. Even worse: a million rows with customer_id = -1 (the "unknown customer" sentinel).
• Power-law distributions in user activity. A few users do 99% of the activity (real e-commerce data). The 1% of "normal" users finish quickly; the 99% on the hot users keeps a task running for hours.

AQE skew-join split — Spark 3.0+.

• spark.sql.adaptive.skewJoin.enabled = true turns it on (default since Spark 3.2).
• AQE measures the shuffle-write size per partition after the shuffle. If any partition is > skewedPartitionFactor × median AND > skewedPartitionThresholdInBytes (default 256 MB), it is declared skewed.
• AQE splits the skewed partition on the join side into skewedPartitionFactor sub-partitions and replicates the matching partition on the non-skew side. The join then runs in parallel across the sub-partitions.
• The cost: the non-skew side gets replicated factor times, so the total work is slightly higher. But the wall-clock time drops from O(skewed partition) to O(median partition).

Salting — the manual fallback.

• AQE skew-join helps with joins. For groupBy + aggregate, AQE cannot split the hot key (the semantics would be wrong).
• Salting adds a random suffix to the hot key to spread it across N partitions, then aggregates in two stages: salted aggregation → un-salt → final aggregation.
• The salting factor is the trade-off between parallelism (more salt = more parallelism on the hot key) and the second-stage aggregation cost (more salt = more rows after the first stage).

Skew hints — surgical control.

• /*+ SKEW('events', 'country_id', 'US') */ tells Spark "the events.country_id column is skewed on value 'US'." Spark splits the join for just that key, leaving the rest of the data alone.
• Useful when you know in advance which keys are hot and AQE's runtime detection is too conservative.

Common interview probes on skew.

• "How do you detect skew in the Spark UI?" — task-duration histogram on the slow stage; max vs 75th percentile.
• "Walk me through AQE skew-join split." — measure size per partition, split the skewed one, replicate the matching one, run in parallel.
• "When does AQE skew-join not help?" — groupBy aggregations, non-join skew, or when the hot partition is below the AQE threshold (250 MB by default).
• "Show me a salted aggregation." — add rand(salt_n) to the key, aggregate by (key, salt), then re-aggregate by key.

Worked example — detect skew from the histogram

Detailed explanation. A Spark job's stage takes 38 minutes when the median task is 12 seconds. The team thinks they need a bigger cluster. The senior move is to open the Stage tab, read the histogram, and confirm whether the problem is skew before changing executor counts.

Question. A stage runs for 38 minutes. The Spark UI shows median task 12 sec, 75th percentile 18 sec, max 38 min. Is this skew? How do you confirm, and what do you do next?

Input — Summary Metrics for Tasks.

Percentile	Duration	Shuffle Read
min	8 s	80 MB
25th	10 s	100 MB
median	12 s	130 MB
75th	18 s	180 MB
max	38 min	22 GB

Code.

# Confirm: max / 75th = 38min / 18s = 127x
# Confirm: max shuffle / 75th = 22GB / 180MB = 122x
# Both > 10x → skew confirmed

# Step 1 — identify the hot key
spark.sql("""
  SELECT country_id, COUNT(*) AS rows
  FROM events
  GROUP BY country_id
  ORDER BY rows DESC
  LIMIT 10
""").show()
# Hot key: 'US' with 1.2B rows, next-largest 'GB' with 12M

# Step 2 — turn on AQE skewJoin if not already
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", str(256 * 1024 * 1024))

# Step 3 — use a skew hint as belt-and-suspenders
result = spark.sql("""
  SELECT /*+ SKEW('events', 'country_id', 'US') */
         e.country_id, c.country_name, COUNT(*) AS clicks
  FROM events e
  JOIN countries c ON e.country_id = c.country_id
  GROUP BY e.country_id, c.country_name
""")

Step-by-step explanation.

1. The 75th percentile vs max ratio is 127x — this is unambiguous skew. Any time max / p75 > 10x, treat it as a skew problem until proven otherwise.
2. Confirm with the Shuffle Read metric: the slow task reads 22 GB while p75 reads 180 MB. The data really is concentrated, not just the compute.
3. Identify the hot key with a quick GROUP BY ... ORDER BY count DESC. The result almost always shows a 100x+ ratio between the top key and the median.
4. Enable AQE skew-join. The skewedPartitionFactor = 5 means a partition is considered skewed if it is > 5x the median size. The thresholdInBytes = 256MB is the floor — partitions smaller than this are never declared skewed even if they are 100x the median.
5. Add a SKEW hint as a safety net. AQE will detect this skew automatically with the right thresholds, but the hint guarantees the split even if your config drifts.

Output.

Metric	Before	After
Stage runtime	38 min	~3 min
Max task	38 min	~30 sec
75th task	18 s	18 s
Hot partition split	1 task / 22 GB	5 tasks / ~4.4 GB
Matching side replication	1x	5x

Rule of thumb. If max / p75 > 10x in the Stage tab, the answer is skew. Open AQE skew-join first; reach for salting only if AQE cannot touch your case.

Worked example — salted aggregation

Detailed explanation. AQE skew-join handles joins, but a GROUP BY country_id with a hot key has no AQE escape. The fix is salting: split the hot key across N partitions, aggregate in two stages, then combine.

Question. Write a salted aggregation for a clicks table where country_id = 'US' accounts for 80% of the rows. Use salt = 16.

Input.

user_id	country_id
u1	US
u2	US
u3	GB
u4	US
u5	DE

Code.

from pyspark.sql.functions import col, concat, lit, expr, rand, floor

clicks = spark.table("clicks")

# Stage 1 — salt the key, aggregate by (key, salt)
SALT = 16
salted = clicks.withColumn("salt", floor(rand() * SALT).cast("int"))

partial = (salted
           .groupBy("country_id", "salt")
           .count()
           .withColumnRenamed("count", "partial_count"))

# Stage 2 — drop the salt, aggregate by key
final = (partial
         .groupBy("country_id")
         .agg({"partial_count": "sum"})
         .withColumnRenamed("sum(partial_count)", "total_count"))

final.show()

Step-by-step explanation.

1. Stage 1 attaches a random integer salt in [0, 16) to every row. The composite key (country_id, salt) has 16x the cardinality of country_id alone. The hot US key is now spread across 16 partitions instead of one.
2. The first groupBy("country_id", "salt").count() runs at high parallelism — 16x for the hot key, 1x for the cold keys. No single task has 80% of the work.
3. Stage 2 drops the salt and re-aggregates. The input to stage 2 is 16 × cardinality(country_id) rows — at most a few thousand rows, even for a billion-row input. Stage 2 is essentially free.
4. The result is identical to a naive groupBy("country_id").count(), but the wall-clock time is bounded by the slowest stage-1 partition (≈ median × 1) instead of the hot key partition (= 80% of total work).
5. The salt factor is the only knob — too low (e.g. 2) and the hot key is still 40% of work; too high (e.g. 1000) and stage 2 has too many input rows. 16-64 is the usual sweet spot.

Output.

country_id	total_count
US	800,000,000
GB	100,000,000
DE	50,000,000
FR	30,000,000
Others	20,000,000

Rule of thumb. Use salting on groupBy aggregations where AQE skew-join cannot help. Pick salt = 16 to start; raise to 64 if the hot partition is still > 5x the median after stage 1.

Worked example — NULL skew on a join key

Detailed explanation. A common production trap: a join key has millions of NULL values from rows where the foreign key is unknown. All NULL rows hash to the same partition and create catastrophic skew. The fix is to filter NULLs before the join, or to replace them with random values that don't match anything.

Question. A join on events.user_id = users.id has 50M rows in events where user_id IS NULL. The join takes 3 hours. What is the cause, and what is the cleanest fix?

Input.

events row	user_id
1	42
2	NULL
3	NULL
4	100
5	NULL

Code.

# DIAGNOSIS — all NULL rows hash to the same partition
spark.sql("SELECT COUNT(*) FROM events WHERE user_id IS NULL").show()
# 50,000,000

# FIX 1 — drop NULLs before the join (if business logic allows)
matched = (events
           .where(col("user_id").isNotNull())
           .join(users, events.user_id == users.id, "inner"))

# FIX 2 — keep NULLs but break the skew
# Replace NULL with a unique random value that won't match in users
import uuid
matched2 = (events
            .withColumn("user_id_safe",
                        expr("coalesce(user_id, CAST(-RAND()*1e18 AS BIGINT))"))
            .join(users, col("user_id_safe") == users.id, "left"))

Step-by-step explanation.

1. All 50M rows where user_id IS NULL hash to a single partition (every NULL hashes to the same value). One task ends up with 50M rows; every other task has thousands.
2. The fastest fix — if business logic allows — is to filter NULLs before the join. They contribute zero matching rows anyway; carrying them through the shuffle is pure cost.
3. If you need to preserve NULL rows (left join, downstream needs them), the trick is to replace NULL with a unique random value per row. The random value won't match anything in users, but it spreads the NULLs across all partitions.
4. The coalesce(user_id, CAST(-RAND()*1e18 AS BIGINT)) expression substitutes a per-row random integer for each NULL. Each NULL row gets a different "fake" join key, so they hash to different partitions.
5. The result is the same as the naive join (NULLs preserved with no match), but the wall clock drops from 3 hours to ~5 minutes.

Output.

events row	user_id	join result
1	42	matched
2	NULL	no match (NULL preserved)
3	NULL	no match (NULL preserved)
4	100	matched
5	NULL	no match (NULL preserved)

Rule of thumb. Always check for NULL skew on join keys. If you find > 1% NULL on a key, either filter NULLs upstream of the join or replace them with random per-row values that defeat the hash.

Senior interview question on skew

A senior interviewer might ask: "Your Spark job has a join where one task runs for 90 minutes and the rest finish in 30 seconds. The job is on Spark 3.4. Walk me through how you fix it — what do you try first, what do you measure, and what's your fallback?"

Solution Using AQE skew-join → SKEW hint → salt fallback

# Step 1 — confirm AQE is enabled and the right thresholds are set
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes",
               str(256 * 1024 * 1024))

# Step 2 — re-run, check the Stage tab
# If max / p75 is still > 10x, AQE didn't detect the skew (threshold too high)
# Lower the factor and threshold:
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes",
               str(64 * 1024 * 1024))

# Step 3 — add a SKEW hint to force the split for known hot keys
result = spark.sql("""
  SELECT /*+ SKEW('events', 'country_id', ('US','GB','DE')) */
         e.user_id, c.country_name, e.event_count
  FROM events e
  JOIN countries c ON e.country_id = c.country_id
""")

# Step 4 — fallback to salting for groupBy skew that AQE cannot help with
SALT = 32
salted = result.withColumn("salt", floor(rand() * SALT).cast("int"))
final = (salted
         .groupBy("country_id", "salt")
         .agg({"event_count": "sum"})
         .groupBy("country_id")
         .agg({"sum(event_count)": "sum"}))

Step-by-step trace.

Step	Action	Outcome
1	AQE skewJoin on, default thresholds	first try — works for 70% of cases
2	Lower factor + threshold	catches mid-sized skew AQE missed
3	SKEW hint on known hot keys	belt-and-suspenders; explicit
4	Salted aggregation	fallback for groupBy skew AQE can't touch

The four steps form a ladder: each step is more invasive than the last, and you stop climbing the moment the histogram flattens out.

Output:

Metric	Default	After Step 1	After Step 2	After Step 4
Max task	90 min	18 min	4 min	1 min
p75 task	30 sec	30 sec	30 sec	30 sec
Stage runtime	90 min	18 min	4 min	1 min
Code change	none	none	hint	salt + 2-stage agg

Why this works — concept by concept:

• AQE skew-join is the first line of defence — since Spark 3.0, the runtime splits skewed partitions during joins without any code change. The default thresholds catch the common case; lowering them catches more.
• Threshold tuning is required for mid-sized skew — the default 256 MB / 5x is conservative. A 128 MB / 3x skewed partition slips through the default thresholds; lowering them catches it.
• SKEW hints make the split explicit — when you know the hot keys in advance (US country code, "anonymous" user_id), the hint forces the split without trusting AQE's detection.
• Salting is the only escape for groupBy skew — AQE skew-join only works for joins. For groupBy aggregations, the runtime cannot split the hot key without changing semantics; salting + two-stage aggregation is the standard workaround.
• Cost — AQE skew-join replicates the matching side, so total work is O(factor × matching rows). Salting doubles the aggregation cost (two stages). Both add < 2x compute on the skewed key while reducing wall-clock by 10-100x.

Optimization
Topic — optimization
Spark skew and join optimization drills

Practice →

SQL Topic — sql SQL join and aggregation problems

Practice →

---

4. Broadcast joins + AQE dynamic switch

broadcast join is the O(N) escape hatch — small × huge becomes one shuffle-free join, but the driver-OOM trap kills the naive broadcast() hint

The mental model in one line: a BroadcastHashJoin (BHJ) collects the small side onto the driver, ships the full table to every executor, and lets each task do a local hash lookup — no shuffle, O(N) on the big side — but if the "small" side turns out to be 4 GB after decompression, the driver OOMs and the job dies. Once you internalise "broadcast is free if the small side fits in driver memory, catastrophic if it doesn't," every broadcast-join interview becomes a sequence of decisions about the broadcast threshold.

Why BHJ beats SMJ when one side is small.

• SortMergeJoin shuffles both sides, sorts both sides, then merges. Cost: O((N + M) × log(N + M)) + 2 shuffles.
• BroadcastHashJoin collects the small side to the driver, ships it to every executor's memory, then does a local hash-table lookup. Cost: O(N) for the big side, O(M) for the broadcast (driver to executors), no shuffle.
• For a 10 MB × 500 GB join, BHJ is 10x-100x faster than SMJ because the shuffle on the 500 GB side is eliminated entirely.

The threshold and how to tune it.

• spark.sql.autoBroadcastJoinThreshold = 10MB by default. Any table the planner estimates to be ≤ 10 MB is auto-broadcast.
• The default is conservative because broadcasting too-large a table OOMs the driver. The 2026 safe value is 100 MB on most production drivers (8 GB+ heap).
• Per-query override is possible with set spark.sql.autoBroadcastJoinThreshold = 209715200 (200 MB) before the query, then reset.
• Explicit hint is df.join(broadcast(small_df), ...) or SELECT /*+ BROADCAST(small) */. The hint forces broadcast regardless of size — useful when planner statistics are wrong, but dangerous because there is no safety net.

AQE dynamic SMJ → BHJ switch.

• Even with autoBroadcastJoinThreshold = 10MB, the optimiser sometimes plans an SMJ because the table-level statistics say "this side is 5 GB."
• AQE runs after the shuffle write and checks the actual shuffle-write size. If filters reduced one side to 50 MB at runtime, AQE flips the plan from SMJ to BHJ on the fly. No code change needed.
• This is the biggest free win in modern Spark — many "tune the broadcast threshold" interview answers from 2019 are obsolete because AQE handles the runtime case.

The driver-OOM trap — when to NOT broadcast.

• Driver heap is finite — --driver-memory 4g is the default; only ~2 GB is usable after JVM overhead and Spark internals.
• A broadcast() hint on a 500 MB compressed Parquet might decompress to 4 GB in memory. The collect kills the driver.
• Symptoms: java.lang.OutOfMemoryError: GC overhead limit exceeded on the driver, not an executor. The job dies before the broadcast even ships.
• Rule: never use the BROADCAST hint on a table you have not personally measured the in-memory size of. Use df.cache().count() and read the In-Memory Size from the Storage tab.

Common interview probes on broadcast joins.

• "What is the difference between SMJ and BHJ?" — shuffle-and-sort both sides vs ship the small side to every executor.
• "When does Spark auto-broadcast?" — when the estimated size is < autoBroadcastJoinThreshold (default 10 MB).
• "What is the AQE dynamic SMJ → BHJ switch?" — AQE re-checks the actual size after the shuffle and flips the plan if one side is now small enough.
• "When does broadcast fail?" — driver OOM when the broadcast side exceeds driver heap; symptom is GC overhead on the driver.

Worked example — explicit broadcast hint for small × huge

Detailed explanation. A daily ETL joins a 50 MB countries table to a 2 TB events table. The optimiser plans an SMJ because the table statistics on countries say "8 GB" (stale stats). Forcing a broadcast collapses the plan to a single O(N) scan.

Question. Write the join with an explicit broadcast() hint, and explain why the optimiser plans SMJ without it.

Input.

Table	Size on disk	In-memory size	Stats say
countries	50 MB	80 MB	8 GB (stale)
events	2 TB	5 TB	accurate

Code.

from pyspark.sql.functions import broadcast

events    = spark.table("events")
countries = spark.table("countries")

# WITHOUT hint — optimiser plans SortMergeJoin because stats say countries = 8GB
# events.join(countries, "country_id")    # SMJ — slow

# WITH hint — force BroadcastHashJoin
enriched = events.join(broadcast(countries), "country_id")
enriched.write.parquet("s3://out/")

Step-by-step explanation.

1. The optimiser computes physical-plan cost using table statistics. If countries has a stale sizeInBytes = 8 GB in the metastore, the planner refuses to auto-broadcast (8 GB > 10 MB threshold) and falls back to SMJ.
2. broadcast(countries) is a hint that overrides the cost-based decision. The driver reads the 50 MB countries table into memory, ships it to every executor, and uses it for a hash-table lookup.
3. The 2 TB events table is scanned once, partition-local. Each partition reads its slice of events, looks up the country code in the in-memory countries hash table, and emits the joined row.
4. No shuffle of the 2 TB side. Compare to SMJ: SMJ would shuffle both sides, sort both sides, and merge — costing two shuffles of 2 TB each plus sort time.
5. The fix can also be permanent — run ANALYZE TABLE countries COMPUTE STATISTICS once to refresh the metastore stats. The next planning pass will see 50 MB and auto-broadcast without the hint.

Output.

Plan	Shuffle bytes	Stage count	Runtime
SMJ (no hint)	4 TB (both sides)	3	90 min
BHJ (hint)	0 (only broadcast 50 MB)	1	8 min

Rule of thumb. When stats are stale or the optimiser refuses to broadcast a small side you know is small, use broadcast(). Always confirm with a Spark UI Stage tab check that the BHJ ran.

Worked example — AQE dynamic SMJ → BHJ on filter pushdown

Detailed explanation. A query joins events (5 TB) with users (50 GB). Static plan: SMJ — both sides too big to broadcast. But after a filter WHERE event_date = '2026-06-15', events is only 5 GB and users after WHERE active = true is only 30 MB. AQE catches this at runtime and flips to BHJ.

Question. Show a SQL query where the static optimiser plans SMJ, but AQE flips to BHJ at runtime. Explain what AQE measures and what triggers the switch.

Input.

Side	Static estimate	Actual after filter
events	5 TB	5 GB
users	50 GB	30 MB

Code.

# Enable AQE with dynamic join switch
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(100 * 1024 * 1024))  # 100MB

result = spark.sql("""
  SELECT u.user_id, u.name, e.event_type, e.event_time
  FROM events e
  JOIN users u ON e.user_id = u.id
  WHERE e.event_date = '2026-06-15'
    AND u.active = true
""")

Step-by-step explanation.

1. The static optimiser sees events table at 5 TB and users at 50 GB. Both > 100 MB threshold, so the static plan is SortMergeJoin with two shuffles.
2. At runtime, the source-side filters (event_date, active) reduce the two sides to 5 GB and 30 MB respectively. Spark writes the shuffle output for each side.
3. AQE inspects the actual shuffle-write size of each side after the shuffle write. It sees that users is now 30 MB — under the 100 MB threshold.
4. AQE rewrites the physical plan: it replaces the SMJ with a BroadcastHashJoin, using the already-shuffled users side as the broadcast input. It also enables localShuffleReader for the events side — instead of shuffling events again, each task reads its local shuffle output directly.
5. Net result: no shuffle of the 5 GB events side at the join step, 30 MB broadcast, sub-minute join completion. Without AQE, this would have run as a 30-minute SMJ on 5 GB × 30 MB.

Output.

Metric	Without AQE	With AQE dynamic switch
Join plan	SortMergeJoin	BroadcastHashJoin
Shuffle on events side	5 GB	5 GB (local read)
Shuffle on users side	30 MB	30 MB (then broadcast)
Network bytes	10 GB	30 MB
Runtime	30 min	90 sec

Rule of thumb. Leave AQE on by default. The dynamic SMJ → BHJ switch is the single highest-leverage feature for joins where filter selectivity is unknown at plan time.

Worked example — the driver-OOM trap

Detailed explanation. A team adds broadcast(orders) to a join, thinking orders is small ("only 500 MB on disk"). The driver dies with OutOfMemoryError. Investigation shows the in-memory representation of orders is 4.5 GB due to decompression and Java object overhead.

Question. Diagnose a broadcast() hint that kills the driver. Walk through the actual in-memory size calculation and explain what the senior engineer measures before adding the hint.

Input.

Metric	Value
orders on disk (Parquet)	500 MB
orders rows	50 M
Driver memory	4 GB total / ~2 GB usable
Hint added	broadcast(orders)
Crash	Driver OutOfMemoryError

Code.

# DANGEROUS — the disk size is NOT the in-memory size
# huge_fact.join(broadcast(orders), "order_id")    # driver OOM

# SAFER — measure the in-memory size first
orders.cache()
orders.count()                                       # materialise
print(spark.catalog.listTables())

# Open the Spark UI Storage tab to read the In-Memory Size
# Suppose it shows 4.5 GB — broadcast is OFF the table

# DECISION TREE
# in-memory size < 100MB → safe to broadcast
# 100MB-1GB → broadcast IF driver has 4GB+ usable heap; raise threshold first
# > 1GB → DO NOT broadcast; SMJ or AQE dynamic switch

# Right move for orders (4.5GB in memory) — let AQE handle it via SMJ
huge_fact.join(orders, "order_id")                   # SMJ; AQE may flip

Step-by-step explanation.

1. Parquet on disk is heavily compressed (zstd/snappy) and column-oriented. A 500 MB Parquet file can decompress to 2-10 GB in memory, depending on column types, string lengths, and null density.
2. The broadcast() hint collects the table to the driver, which holds it in row-oriented form in memory. The driver heap fills, GC starts thrashing, and the JVM exits with OutOfMemoryError.
3. The fix is to measure first: call df.cache() then df.count() to materialise the table, then read the In-Memory Size from the Spark UI Storage tab. This number is the truth.
4. Decision rule: in-memory < 100 MB → broadcast is safe on any reasonable driver. 100 MB-1 GB → broadcast with --driver-memory 8g+. > 1 GB → do not broadcast.
5. For orders at 4.5 GB in memory, the only correct plan is SMJ. AQE may still flip to BHJ at runtime if a downstream filter reduces orders to a small subset — let that be automatic, never force it manually.

Output.

Side	On disk	In memory	Broadcast safe?
countries	50 MB	80 MB	yes (any driver)
dim_date	1 MB	10 MB	yes (default)
orders	500 MB	4.5 GB	NO (use SMJ)
products	100 MB	800 MB	only with 8 GB+ driver

Rule of thumb. Never use broadcast() without measuring the in-memory size in the Spark UI Storage tab. The disk size is misleading; the in-memory size is what kills the driver.

Senior interview question on broadcast joins

A senior interviewer might ask: "You have a star-schema query with one fact table (5 TB) and three dimension tables (50 MB, 200 MB, 1.5 GB). The optimiser plans three SortMergeJoins. What should you do to speed it up, and what should you not do?"

Solution Using selective broadcast hints + AQE dynamic switch

from pyspark.sql.functions import broadcast

fact   = spark.table("fact_sales")     # 5 TB
dim_c  = spark.table("dim_country")    # 50 MB on disk, 80 MB in mem
dim_p  = spark.table("dim_product")    # 200 MB on disk, 350 MB in mem
dim_cu = spark.table("dim_customer")   # 1.5 GB on disk, 12 GB in mem (DO NOT broadcast)

# Raise the threshold to 400MB so dim_c and dim_p auto-broadcast
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(400 * 1024 * 1024))

# Enable AQE so dim_cu can still flip to BHJ at runtime if filters reduce it
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true")

# Explicit hints for the two safe broadcasts; SMJ for dim_cu
result = (fact
          .join(broadcast(dim_c), "country_id")
          .join(broadcast(dim_p), "product_id")
          .join(dim_cu, "customer_id"))            # SMJ; AQE may flip later

Step-by-step trace.

Side	In-memory size	Strategy	Why
dim_country	80 MB	broadcast	well under driver heap
dim_product	350 MB	broadcast	safe with --driver-memory 4g+
dim_customer	12 GB	SMJ (no hint)	broadcast would OOM the driver
fact_sales	5 TB	drives both join sides	always the streaming side

After this plan: two BHJs (free, no shuffle), one SMJ on the largest dim. AQE may still flip the SMJ to BHJ at runtime if WHERE clauses reduce dim_customer.

Output:

Metric	All-SMJ	Selective BHJ + SMJ
Shuffle bytes	5 TB × 3 sides	5 TB × 1 side (just dim_customer)
Stage count	5	2
Runtime	4 hours	35 min
Driver OOM risk	none	none

Why this works — concept by concept:

• Selective broadcasting — broadcast the dimensions you can safely fit in driver memory (measured, not estimated). For the dimension that won't fit, fall back to SMJ.
• Threshold raise to 400 MB — the default 10 MB is too conservative for the 2026 typical driver. 100-400 MB is safe on any driver with 8 GB+ heap.
• AQE for the largest dimension — leave SMJ as the static plan; let AQE flip to BHJ at runtime if filters reduce the size. No manual decision needed.
• Measure in-memory size, not disk size — Parquet compresses 5-10x. Always cache + count + read the Storage tab before adding broadcast(). The disk size lies.
• Cost — each broadcast is O(small_side × executors) network. SMJ is O(both sides × log) shuffle + sort. For a 5 TB fact against three small dims, BHJ saves 10 TB of shuffle; that's the win.

Optimization
Topic — optimization
Join strategy and broadcast tuning drills

Practice →

ETL Topic — etl Star-schema and fact-dim ETL problems

Practice →

---

5. Persist / cache + memory fractions

spark persist is a contract with the memory manager — cache() is shorthand for MEMORY_AND_DISK on DataFrames, and spark.memory.fraction is what makes it actually work

The mental model in one line: cache() and persist() reserve space in the storage half of Spark's unified memory pool; if spark.memory.fraction (default 0.6) and the storage / execution split are wrong for your workload, cached data either evicts working data (job slows) or gets evicted itself (cache becomes a no-op). Once you internalise "caching is a contract with the memory manager, not a magic speedup," every persist interview becomes a discussion of memory budgets.

cache() vs persist() — the API and the storage levels.

• df.cache() is identical to df.persist(StorageLevel.MEMORY_AND_DISK) — Spark first tries to hold the partition in memory and spills to disk if memory is exhausted.
• df.persist(StorageLevel.MEMORY_ONLY) is the RDD default; on a DataFrame, you have to pass it explicitly. It evicts the partition (recomputes on access) rather than spilling.
• df.persist(StorageLevel.DISK_ONLY) is used when the data is too big for memory or you want to keep memory free for execution.
• df.persist(StorageLevel.MEMORY_AND_DISK_SER) serialises the in-memory representation — saves ~50% memory in exchange for CPU on deserialise.
• df.persist(StorageLevel.OFF_HEAP) uses off-heap memory (Tungsten managed) — requires spark.memory.offHeap.enabled = true.

The "cache then count" anti-pattern.

• A common mistake is df.cache() followed by df.count() "to materialise." This works but hides the cost: if df is only used once downstream, the cache + count adds an extra full scan for no benefit.
• The senior rule: only cache a DataFrame if it will be read at least 3 times. Caching for a single downstream action is a net cost.
• Always pair cache() with an explicit unpersist() when the cached data is no longer needed. Without unpersist, the storage pool stays full and evicts other useful caches.

spark.memory.fraction and the unified memory model.

• Spark's executor memory is split into system reserved, user memory, and unified memory (storage + execution).
• spark.memory.fraction = 0.6 (default since Spark 1.6) — 60% of executor heap goes to unified memory. The remaining 40% is for user data structures (UDF closures, user-managed state).
• Within the unified pool, storage (cache) and execution (shuffle, aggregate, sort buffers) borrow from each other dynamically. spark.memory.storageFraction = 0.5 reserves 50% of unified memory for storage that cannot be evicted by execution.
• For cache-heavy workloads, raise spark.memory.fraction to 0.75 (less user-memory headroom) and spark.memory.storageFraction to 0.6 (more eviction-safe cache).
• For compute-heavy workloads (huge shuffles, large aggregations), lower spark.memory.fraction to 0.5 or shift the split toward execution.

Off-heap memory — the very-large-working-set toggle.

• spark.memory.offHeap.enabled = true + spark.memory.offHeap.size = 8g (per executor) — gives Spark 8 GB of off-heap memory per executor for unified memory operations.
• Off-heap memory is managed by Tungsten and not subject to JVM GC pressure. For very large working sets (multi-GB shuffles, large aggregations), off-heap eliminates the GC tax.
• The trade-off: off-heap requires manual sizing; you don't get the JVM's automatic memory management. Get it wrong and you OOM at the OS level (process killed by the kernel) rather than at the JVM level.

Common interview probes on persist and memory.

• "What is the difference between cache() and persist()?" — cache() is shorthand for persist(MEMORY_AND_DISK) on DataFrames; persist() lets you pick any storage level.
• "When should you NOT cache?" — when the DataFrame is read only once, when memory is contested by shuffles, when the cache will evict working data.
• "What does spark.memory.fraction control?" — the share of executor heap allocated to Spark's unified memory pool (storage + execution).
• "When do you use off-heap memory?" — for very large working sets where JVM GC is the bottleneck.

Worked example — when persist actually pays off

Detailed explanation. An iterative algorithm (e.g. K-means, PageRank, an ML training loop) reads the same DataFrame 50 times. Without caching, each iteration re-reads the source. With persist(MEMORY_AND_DISK), the data lives in memory after the first iteration.

Question. Show an iterative job that benefits from persist. Compare runtime with and without caching, and explain why the savings are O(iterations × read cost).

Input.

Metric	Value
Source size	200 GB
Iterations	50
Read cost per iteration	8 min

Code.

# WITHOUT cache — re-reads 200GB on every iteration
df = spark.read.parquet("s3://training-data/")

for i in range(50):
    metrics = df.groupBy("class").agg({"feature": "mean"}).collect()
    # ... use metrics to update model ...

# Total: 50 iterations × 8 min/read = 400 min

# WITH persist — first iteration reads, all subsequent iterations hit memory
df = spark.read.parquet("s3://training-data/").persist(StorageLevel.MEMORY_AND_DISK)
df.count()  # force materialisation

for i in range(50):
    metrics = df.groupBy("class").agg({"feature": "mean"}).collect()

df.unpersist()
# Total: 1 read (8 min) + 49 in-memory iterations (~30 sec each) = ~33 min

Step-by-step explanation.

1. Without caching, each groupBy re-reads the full 200 GB from S3 because Spark has no idea you'll touch it again. The 50-iteration loop is 50 × 8 min = 400 min.
2. With persist(MEMORY_AND_DISK), the first read materialises the DataFrame into Spark's storage pool. Subsequent iterations read from the in-memory cache, which is 50-100x faster than re-reading from S3.
3. The count() after persist forces the materialisation. Without it, the next action would materialise — but if the next action only touches a subset, only that subset gets cached.
4. MEMORY_AND_DISK means partitions that don't fit in memory spill to disk. For 200 GB with 100 GB of cluster cache space, ~100 GB is in memory and 100 GB spills to disk. Spilled partitions are still much faster than re-reading from S3 because the spill is on local SSD.
5. The unpersist() at the end releases the cache space for downstream work. Without it, the next ETL stage's shuffle competes with the now-useless cache for the same memory pool.

Output.

Strategy	Total runtime	Iterations
No cache	400 min	50 × 8 min
persist	33 min	1 × 8 + 49 × 30 sec
Speedup	12x	—

Rule of thumb. Persist when a DataFrame will be read more than 3 times. Always pair persist with unpersist when you're done. Pick MEMORY_AND_DISK unless you have a specific reason for MEMORY_ONLY.

Worked example — the cache-then-count anti-pattern

Detailed explanation. A team adds df.cache(); df.count() to "speed up" a one-pass job. The job that ran in 12 minutes now runs in 18 minutes because the count is an extra full scan and the cache evicts work-set data needed by the shuffle.

Question. Show a one-pass job where cache() actively hurts performance. Explain what's happening in the memory pool and how to fix it.

Input.

Metric	Value
Source size	500 GB
Operations	1 filter + 1 shuffle + 1 write
Cluster cache space	100 GB
Working-set size (shuffle buffer)	80 GB

Code.

# BAD — cache + count adds 6 min and evicts shuffle buffers
df = spark.read.parquet("s3://src/").cache()
df.count()                                       # 6 min full scan
result = df.filter("status='ok'").groupBy("user_id").count()
result.write.parquet("s3://out/")                # shuffle competes with cache

# GOOD — no cache; Spark handles the single pass natively
df = spark.read.parquet("s3://src/")
result = df.filter("status='ok'").groupBy("user_id").count()
result.write.parquet("s3://out/")                # one pass, shuffle uses full memory

Step-by-step explanation.

1. The cache() reserves storage in the executor memory pool. The count() action triggers a full scan to materialise the cache. The cache now occupies ~100 GB across the cluster.
2. The filter + groupBy + write is a single-pass operation. It does not benefit from the cache — Spark would have read the source once anyway.
3. The shuffle step needs ~80 GB of execution memory across the cluster. With cache occupying 100 GB and spark.memory.storageFraction = 0.5, the shuffle can only evict half the cache. Some of the shuffle spills to disk; some cache evicts. Both are pure overhead.
4. Removing the cache() lets the shuffle use the full unified memory pool. The single pass is now 12 minutes instead of 18.
5. The senior rule: count downstream actions before caching. If the DataFrame is referenced exactly once downstream, never cache it.

Output.

Pattern	Runtime	Cache evictions	Shuffle spill
cache() + count()	18 min	many	30 GB
no cache	12 min	0	5 GB

Rule of thumb. Cache exists for repeated reads. If the next action is the only downstream action, caching adds the materialisation cost plus competes with execution for memory. Skip it.

Worked example — memory.fraction for cache-heavy workloads

Detailed explanation. A reporting workload pre-computes 30 cached DataFrames (one per market segment) and runs 200 small queries against them. The default spark.memory.fraction = 0.6 leaves only 0.6 × 0.5 = 30% of executor heap as eviction-safe cache, so heavy queries keep evicting the cache. Raising spark.memory.fraction to 0.8 and spark.memory.storageFraction to 0.7 fixes it.

Question. Compute the eviction-safe cache size for the defaults vs the tuned settings on a 32 GB executor heap. Show the config change and explain when this trade-off is right.

Input.

Setting	Default	Tuned
spark.memory.fraction	0.6	0.8
spark.memory.storageFraction	0.5	0.7
Executor heap	32 GB	32 GB

Code.

# Cache-heavy tuning — raise both fractions
spark.conf.set("spark.memory.fraction", "0.8")
spark.conf.set("spark.memory.storageFraction", "0.7")

# Math on 32 GB executor heap
heap = 32 * 1024  # MB
reserved = 300    # system reserved (MB)
usable = heap - reserved

# default — unified pool = 19,026 MB; eviction-safe cache = 9,513 MB
unified_default = (heap - reserved) * 0.6
safe_default    = unified_default * 0.5

# tuned — unified pool = 25,368 MB; eviction-safe cache = 17,758 MB
unified_tuned = (heap - reserved) * 0.8
safe_tuned    = unified_tuned * 0.7

print(f"Default safe cache: {safe_default:.0f} MB")
print(f"Tuned   safe cache: {safe_tuned:.0f} MB")

Step-by-step explanation.

1. The executor heap is partitioned into reserved memory (300 MB), user memory ((1 - memory.fraction) × usable), and unified memory (memory.fraction × usable).
2. Within unified memory, the storage / execution split is storageFraction × unified. Above the storage fraction, cache can still be evicted by execution under pressure. Below the storage fraction, cache is eviction-safe.
3. Default math: 32 GB × 0.6 = 19 GB unified; 19 GB × 0.5 = ~9.5 GB eviction-safe. The 30 cached DataFrames at 1 GB each = 30 GB total, of which only ~9.5 GB is safe — most cache gets evicted under shuffle pressure.
4. Tuned math: 32 GB × 0.8 = ~25 GB unified; 25 GB × 0.7 = ~17.8 GB eviction-safe. The 30 GB cache still doesn't all fit, but the eviction-safe portion is 2x larger.
5. The trade-off: less user memory (UDF closures, broadcast vars, internal stuff) and less execution memory (shuffle, aggregate). For a cache-heavy reporting workload, this is the right call. For a compute-heavy ETL with huge shuffles, keep defaults.

Output.

Setting	Unified memory	Eviction-safe cache
Defaults (0.6 / 0.5)	19 GB	9.5 GB
Tuned (0.8 / 0.7)	25 GB	17.8 GB

Rule of thumb. Raise spark.memory.fraction only when you have measured cache eviction in the Spark UI Storage tab. For compute-heavy jobs, the defaults are correct.

Senior interview question on memory tuning for OOM

A senior interviewer might ask: "Your Spark job hits OutOfMemoryError on executors during a large aggregation. The data is 5 TB after a WHERE filter. What do you check, and what's the right sequence of config changes?"

Solution Using partition resize + memory fractions + off-heap

# Sequence — try cheapest fix first, off-heap last

# 1. RESIZE PARTITIONS — most OOMs are oversized partitions, not undersized memory
spark.conf.set("spark.sql.shuffle.partitions", 25600)   # 5TB / 200MB
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# 2. CONFIRM IT'S NOT SKEW — open the Stage tab, check task-duration histogram
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# 3. LOWER MEMORY FRACTION FOR EXECUTION — give execution more room
spark.conf.set("spark.memory.fraction", "0.7")
spark.conf.set("spark.memory.storageFraction", "0.3")

# 4. RAISE EXECUTOR MEMORY OVERHEAD — for off-heap auxiliary memory
spark.conf.set("spark.executor.memoryOverhead", "4g")

# 5. ENABLE OFF-HEAP — last resort for very large working sets
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "8g")

Step-by-step trace.

Step	What it solves	Cost
1. Partition resize	Each task's working set fits in memory	0 — config only
2. AQE skew-join	One task isn't 100x median	0 — config only
3. Lower memory fraction (storage)	More room for execution buffers	reduces cache space
4. Raise overhead	Off-heap auxiliary memory headroom	adds ~10% per executor
5. Enable off-heap	Working set lives outside JVM heap	requires explicit sizing

The ladder is ordered cheapest-to-most-invasive. Stop the moment OOM stops.

Output:

Step	OOM still happens?	Notes
1 only	sometimes	fixes ~60% of OOMs
1 + 2	sometimes	fixes ~80% of OOMs
1 + 2 + 3	rarely	fixes ~95% of OOMs
1 + 2 + 3 + 4	almost never	fixes ~99% of OOMs
1 + 2 + 3 + 4 + 5	never	for truly enormous working sets

Why this works — concept by concept:

• Partition resize first — 60% of "OOM" symptoms are actually "this partition is 20 GB and my task heap is 4 GB." Resizing partitions to 200 MB each removes the root cause without touching memory configs.
• Skew handling second — even with the right partition count, one hot key can OOM a single task. AQE skew-join + salting handle this layer.
• Memory fraction tuning third — only after partition and skew are fixed, look at the heap split. For compute-heavy jobs, lower storageFraction to give execution more room.
• Memory overhead for off-heap auxiliary structures — Spark uses off-heap for shuffle service, native libraries, and internal buffers. The default spark.executor.memoryOverhead = max(384MB, 0.1 × executor heap) is often too low for very large jobs.
• Off-heap last — explicit off-heap is invasive (requires manual sizing, can OOM at the OS level) but bypasses JVM GC entirely. Use only when steps 1-4 are exhausted.
• Cost — steps 1-2 are O(1) config cost. Steps 3-4 trade cache size for execution headroom. Step 5 adds OS-level memory budget. Sequence cheapest-to-most-invasive; stop the moment OOM stops.

ETL
Topic — etl
Spark memory tuning and OOM drills

Practice →

Optimization
Topic — optimization
Performance and memory optimization problems

Practice →

---

Cheat sheet — Spark tuning recipes

• Shuffle-partition formula. spark.sql.shuffle.partitions = total_shuffle_bytes ÷ 200_MB. Round up to a multiple of cluster cores. Always pair with AQE coalesce as the safety net.
• AQE 3-key config block. spark.sql.adaptive.enabled = true, spark.sql.adaptive.coalescePartitions.enabled = true, spark.sql.adaptive.skewJoin.enabled = true. Add spark.sql.adaptive.advisoryPartitionSizeInBytes = 200MB for the target.
• Skew detection rule. In the Stage tab Summary Metrics, if max_duration / p75_duration > 10x (or max_shuffle / p75_shuffle > 10x), declare skew. Apply AQE skew-join first; salt only when AQE can't help.
• Salting one-liner. df.withColumn("salt", floor(rand() * 32).cast("int")).groupBy(key, "salt").agg(...).groupBy(key).agg(...). Salt = 16-64 covers most groupBy skew.
• Broadcast threshold safe values. Default autoBroadcastJoinThreshold = 10MB. Raise to 100 MB on any driver with 4 GB+ heap, 400 MB with 8 GB+ heap. Never raise blindly — always measure in-memory size first.
• Driver-OOM rule. Never use broadcast() on a side you haven't measured. Cache, count, read Storage tab; if in-memory size > 1 GB, do NOT broadcast.
• AQE dynamic SMJ → BHJ. Free win — leave AQE on; if filters reduce one side under threshold at runtime, AQE flips SMJ to BHJ automatically.
• cache() vs persist(). cache() is shorthand for persist(MEMORY_AND_DISK) on DataFrames. Use persist(MEMORY_AND_DISK_SER) for ~50% memory savings at the cost of deserialise CPU. Only cache if read ≥ 3 times.
• Cache-then-count anti-pattern. df.cache(); df.count() for a single downstream action is pure cost. Cache only when you've counted the downstream reads.
• Persist + unpersist discipline. Every persist() needs a matching unpersist() at the end of the scope. Without unpersist, the storage pool stays full and degrades downstream jobs.
• Memory fraction tuning. spark.memory.fraction = 0.6 default — raise to 0.75 for cache-heavy reporting, lower to 0.5 for compute-heavy ETL with huge shuffles. spark.memory.storageFraction = 0.5 default — raise for eviction-safe cache, lower for execution-heavy jobs.
• Off-heap memory. Enable with spark.memory.offHeap.enabled = true + spark.memory.offHeap.size = 8g. Use only for very large working sets where JVM GC is the bottleneck. Sizes are manual; OOMs happen at OS level.
• OOM ladder. (1) resize partitions, (2) AQE skew-join, (3) lower memory.fraction for execution, (4) raise executor.memoryOverhead, (5) enable off-heap. Stop the moment OOM stops; do not skip steps.
• coalesce vs repartition. coalesce(N) merges existing partitions (no shuffle, but collapses upstream parallelism). repartition(N) always shuffles. Use coalesce only to shrink by ≤ 10x; use repartition everywhere else.
• Spark UI workflow. Jobs tab → slowest job → Stages tab → slowest stage → task-duration histogram. SQL tab for physical plan. Storage tab for in-memory sizes. Executors tab for GC and heap pressure. Always start here before changing config.

Frequently asked questions

How do I tune spark.sql.shuffle.partitions?

The 2026 answer: size first, count second. Estimate total shuffle-write bytes from the Spark UI and divide by your target partition size (~200 MB) — a 10 TB shuffle → 51,200 partitions; a 1 GB shuffle → 5 partitions. Pair the explicit count with AQE coalesce (spark.sql.adaptive.coalescePartitions.enabled = true) so AQE merges sub-target partitions at runtime if your estimate was wrong. The combination — high initial bound + AQE coalesce — is robust across 10,000x swings in input size with one config block. Never leave the historical default of 200 on an unknown workload; it produces 50 GB partitions on multi-TB shuffles (spill, OOM) and 5 MB partitions on small queries (scheduler overhead).

How do I detect skew in Spark?

Open the Spark UI → Stages tab → click the slowest stage → scroll to Summary Metrics for Tasks. Compare max to the 75th percentile for Duration and Shuffle Read Size. If max / p75 > 10x on either, you have skew — one partition is dramatically larger than the rest (even bars + one tall bar in the histogram is textbook). Identify the hot key with SELECT key, COUNT(*) FROM table GROUP BY key ORDER BY count DESC LIMIT 20. The fix is AQE skew-join (spark.sql.adaptive.skewJoin.enabled = true) for join skew, or salting (withColumn("salt", floor(rand() * 32)) + two-stage aggregation) for groupBy skew. The Spark UI is the source of truth — never change configs based on guesses.

When should I broadcast a join?

Broadcast when the small side fits comfortably in driver memory — < 100 MB in-memory for a 4 GB driver, < 400 MB with 8 GB driver heap. The critical detail: disk size is not in-memory size. A 500 MB Parquet file can decompress to 4-10 GB in memory due to compression and Java object overhead. Measure in-memory size by df.cache(); df.count() then reading the In-Memory Size from the Spark UI Storage tab before adding a broadcast() hint. With AQE enabled, the dynamic SMJ → BHJ switch handles most cases automatically — Spark flips the plan at runtime if filters reduce one side under threshold. Manual broadcast() is for stale statistics or known-small sides that won't auto-broadcast. Never use broadcast() on a side you haven't measured — driver-OOM kills the whole job.

cache() vs persist() — what's the difference?

df.cache() is shorthand for df.persist(StorageLevel.MEMORY_AND_DISK) on DataFrames — Spark tries to hold the partition in memory and spills to disk if memory is exhausted. persist() lets you pick any storage level: MEMORY_ONLY (evict, don't spill), DISK_ONLY (skip memory entirely), MEMORY_AND_DISK_SER (serialised in-memory — ~50% memory savings at CPU cost), OFF_HEAP (Tungsten-managed off-heap, requires spark.memory.offHeap.enabled = true). The senior rule: cache only when a DataFrame is read 3+ times. A single downstream action gains nothing from caching but adds the materialisation cost plus competes with shuffle buffers for the unified memory pool. Always pair cache() / persist() with an explicit unpersist() at the end of the scope — without unpersist, the storage pool stays full and degrades subsequent jobs that need execution memory.

Spark out-of-memory — what config knobs help?

The OOM ladder, cheapest to most-invasive: (1) resize partitions — spark.sql.shuffle.partitions = total_bytes ÷ 200_MB plus AQE coalesce. 60% of OOMs are oversized partitions, not undersized memory. (2) AQE skew-join — spark.sql.adaptive.skewJoin.enabled = true. A single hot key OOMs one task regardless of total cluster memory. (3) Lower spark.memory.fraction to 0.5 and spark.memory.storageFraction to 0.3 for compute-heavy jobs — more execution headroom. (4) Raise spark.executor.memoryOverhead to 4-8 GB for off-heap auxiliary structures. (5) Enable off-heap — spark.memory.offHeap.enabled = true + spark.memory.offHeap.size = 8g. Off-heap is last because it requires manual sizing and OOMs at the OS level. Stop the moment OOM stops; do not skip steps. Raising --executor-memory blindly is the most common over-correction — it papers over the real bug at 2-4x cluster cost.

How do I read the Spark UI to find a slow job?

Open the UI in this order: Jobs tab → click the slowest job. Stages tab → click the slowest stage. Summary Metrics for Tasks → read the task-duration histogram. Even bars = no skew (look at shuffle size next); one tall bar = skew (apply AQE skew-join). SQL tab → physical plan. SortMergeJoin where one side < 100 MB = missed broadcast. ShuffleExchange with large bytes = shuffle target (resize spark.sql.shuffle.partitions). Storage tab → in-memory sizes of cached DataFrames. Executors tab → GC time per executor (high GC = heap pressure, consider off-heap or lower memory.fraction). Spend the first 15 minutes here before touching a single config — the UI tells you which axis to tune. Tuning blind is the senior-Spark anti-pattern.

Practice on PipeCode

• Drill the ETL practice library → for the Spark transform-and-load family of probes.
• Rehearse on medium-difficulty ETL problems → when the interviewer wants tuning-depth scenarios.
• Sharpen the optimization axis with the optimization practice library → for shuffle, skew, and join-strategy drills.
• Stack the SQL practice library → for the aggregation, join, and window-function patterns that map directly to Spark SQL.
• For the broader senior-DE surface, work through PipeCode's 450+ DE-focused problems with a real-time scoring engine.

Lock in Spark tuning muscle memory

Docs explain the configs. PipeCode drills explain the decision — when to broadcast, when to salt, when AQE handles skew for you. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL & pipeline problems →
Practice optimization problems →