Warmup Guide — Apache Spark Internals

Zero-to-principal primer for Phase 06: how Spark turns your code into jobs/stages/tasks, why the shuffle dominates everything, how skew/spill/small-files actually arise, what AQE does, and how to read and fix a 9-hour job.

Table of Contents

Chapter 1: The Execution Hierarchy

Spark turns a program into a hierarchy you must be able to name to debug:

  • Application → one driver + executors. Job → triggered by each action (count, write, collect). Stage → a set of tasks that can run without a shuffle; stages are split at shuffle boundaries (Ch. 2). Task → the unit of execution, one per partition, run on an executor core.
  • So the parallelism of a stage = its partition count, and the wall-clock time of a stage ≈ the slowest task (the straggler — Ch. 4). Two numbers explain most performance: how many partitions, and how evenly the data is spread across them.
  • The DAG scheduler builds the stage graph; the task scheduler places tasks on executors honoring locality. The lab's plan_stages is the DAG scheduler's split-at-shuffle in miniature.

Chapter 2: Narrow vs Wide — Why Shuffles Make Stages

A transformation's dependency determines whether it needs a shuffle:

  • Narrow: each output partition depends on one input partition (map, filter, union). These pipeline within a stage — no data movement across the network.
  • Wide: each output partition depends on many input partitions (groupByKey, reduceByKey, join, repartition, distinct, sort). These require a shuffle: redistribute data by key across the cluster. A shuffle materializes intermediate data (written to disk) and ends a stage — the next stage reads the shuffle output.

So #stages = #shuffles + 1, and every wide op is a place where data crosses the network and hits disk. This is why the principal's first instinct on a slow job is "how many shuffles, and can I remove one?" (e.g. replace a shuffle-join with a broadcast-join, Ch. 5).

Chapter 3: The Shuffle in Detail

The shuffle is the most expensive operation in Spark and the source of most pathologies:

  • Map side: each task partitions its output by the target partition (hash of key), sorts/buffers it, and spills to disk when buffers fill, producing shuffle files.
  • Reduce side: each task in the next stage fetches its partition's data from every map task (an all-to-all transfer) — M × R connections in the worst case. The external shuffle service serves these so executors can die without losing shuffle data.
  • Shuffle explosion: too many partitions (tiny tasks, scheduler overhead, many small files) or too few (huge tasks, spill, OOM). spark.sql.shuffle.partitions defaults to 200 — almost always wrong for your data; size it by bytes (the lab's shuffle_partitions: ceil(total / 128MB)).
  • The cost levers: fewer shuffles (broadcast, pre-partitioning/bucketing), right-sized partitions (128–256 MB), and AQE (Ch. 6).

Chapter 4: Data Skew

Skew is P01's hot-partition problem, now killing a batch job: a few hot keys (a "whale" account, a NULL join key, a Zipfian distribution) send disproportionate data to a few reduce partitions, so a handful of tasks run for hours while the rest finish in seconds — and the stage can't complete until the straggler does.

  • Detect: per-partition (per-task) input sizes in the Spark UI; a few partitions far above the median (the lab's detect_skew, median-multiple). Also: one task at 100% while others idle; massive spill on a few tasks.
  • Fix:
    • Broadcast the other side if it's small (no shuffle at all — Ch. 5).
    • Salting: append a random bucket to the hot key (key#0..key#n), aggregate in two stages (per-bucket then combine) — spreads the whale across partitions (P01's salt).
    • AQE skew-join (Ch. 6): Spark automatically splits skewed partitions into sub-partitions at runtime.
    • Filter/handle the NULL-key case explicitly (a shockingly common cause).

Chapter 5: Join Strategies

Spark picks a join physical plan; knowing them lets you force the right one:

  • Broadcast-hash join: ship the small side to every executor and hash-join locally — no shuffle. The fastest when one side fits spark.sql.autoBroadcastJoinThreshold (default 10 MB; raise it deliberately for medium dims). The lab's should_broadcast.
  • Sort-merge join: shuffle both sides by key, sort, merge. The default for two large tables; robust but pays two shuffles + sorts.
  • Shuffle-hash join: shuffle both, build a hash table on one side; used in narrower cases.
  • Dynamic partition pruning: at runtime, use the dimension's join keys to prune the fact-table partitions scanned — a big win for star schemas.

The principal move: confirm the small side is actually small (stats can be stale → ANALYZE TABLE), broadcast it, and you've turned a two-shuffle sort-merge into a no-shuffle broadcast — often the single biggest speedup.

Chapter 6: Adaptive Query Execution

AQE re-optimizes the plan at runtime using actual shuffle statistics (instead of trusting stale compile-time estimates). Its three tricks, all interview-relevant:

  1. Coalesce shuffle partitions: after a shuffle, merge small adjacent partitions up to a target size — so you can set shuffle.partitions high without ending up with thousands of tiny tasks (the lab's coalesce_partitions). Note it never splits a big partition — that's the skew path.
  2. Switch join strategy: a planned sort-merge join becomes a broadcast join once runtime stats reveal one side is actually small.
  3. Skew-join handling: detect skewed partitions and split them into sub-partitions that run in parallel — automating Ch. 4's salting.

Turn AQE on (spark.sql.adaptive.enabled=true, default in modern Spark) and a lot of manual tuning evaporates — but you still must understand what it's doing to debug when it doesn't help (e.g. it can't fix skew it can't see, or a UDF that blocks pushdown).

Chapter 7: Memory, Spill, and Executor Sizing

  • Unified memory: an executor's heap is split (after reserved memory) into a region governed by spark.memory.fraction (default 0.6), shared between execution (shuffle, join, sort, aggregation buffers) and storage (cached data), which borrow from each other. The rest is user memory.
  • Spill: when a task's working set (e.g. a partition being sorted/aggregated) exceeds the available execution memory, Spark spills to disk — correct but slow (the lab's estimate_spill). The fix is frequently more partitions (smaller per-partition working set), not just more memory — a key insight juniors miss.
  • Executor OOM: too few partitions (giant tasks), a huge collect/broadcast, skew piling data on one executor, or too many cores per executor (each core's task competes for the same heap). Sizing: a common rule is ~4–5 cores/executor and memory sized so target-size partitions fit with headroom; more, smaller executors often beat few giant ones.
  • Off-heap / Tungsten: Spark's binary memory format + code generation reduce GC and object overhead — why Spark SQL is faster than RDD-of-objects.

Chapter 8: File Layout and the Small-File Problem

How Spark writes determines downstream cost (P08–P10):

  • A write produces one file per output partition per task, so 200 shuffle partitions × N tasks can emit millions of tiny files — the small-file problem. Tiny files mean huge metadata, slow listing, poor scan performance, and (on S3) request-cost and throttling pain. The lab's small_file_report quantifies the fix.
  • Target file size: 128 MB – 1 GB. Too small = the tax above; too large = poor read parallelism and memory pressure.
  • Fixes: repartition(n) / coalesce(n) to target the right file count before write (repartition shuffles to balance; coalesce avoids a shuffle but can't increase partitions); partition the output by a sensible column (date) without over-partitioning (which recreates the small-file problem per partition); run compaction (P09) on table formats.

Chapter 9: Reliability, Idempotent Writes, and Backfills

  • Fault tolerance: tasks retry on failure; stages recompute from shuffle data (or recompute the lineage if shuffle data is lost). Speculative execution re-launches slow ("straggler") tasks on other executors and takes whichever finishes first — helps with bad nodes, wastes resources on genuine skew (where the fix is Ch. 4, not speculation).
  • Idempotent writes: a naive job that fails mid-write can leave partial output; a retry then duplicates. The fixes: write to a staging path then atomically rename/commit (the FileOutputCommitter v2 caveats matter on S3 — no atomic rename!), or use a table format (P09) whose commit is atomic (the safe modern answer). Overwrite a partition atomically (INSERT OVERWRITE dynamic partition) for partition-scoped backfills.
  • Safe backfills (a JD deliverable): partitioned by date, resumable (reprocess only failed partitions), reproducible (deterministic logic — event-time, P01 — so a rerun yields identical output), and auditable (record what was reprocessed). The test of a good backfill is the same as a good replay: rerun it and get the same answer.

Lab Walkthrough Guidance

Lab 01 — Spark Execution Simulator, suggested order:

  1. plan_stages/num_stages — new stage at each wide op; #stages = #wide + 1.
  2. detect_skew — median-multiple; the fat partition pops out.
  3. coalesce_partitions — greedy merge to target; the big partition stands alone.
  4. should_broadcast + shuffle_partitions — the join + sizing decisions.
  5. estimate_spill — partition vs execution-memory share.
  6. small_file_report — the compaction pitch (reduction ratio).

Success Criteria

You are ready for Phase 07 when you can, from memory:

  1. Map application → job → stage → task and say what creates a stage.
  2. Explain the shuffle (map write/spill → reduce fetch) and M×R cost.
  3. Diagnose skew (detection + four fixes) and explain salting.
  4. Choose broadcast vs sort-merge and state the threshold.
  5. Explain AQE's three tricks.
  6. Explain spill and why more partitions can fix it; size executors sanely.
  7. Explain the small-file problem and how to fix it on write.
  8. Describe an idempotent, reproducible backfill.

Interview Q&A

Q (Round 3): A Spark job runs 9 hours, spills heavily, writes millions of small Parquet files, and Athena then scans too much. Walk me through it. Four linked symptoms. The 9 hours + heavy spill point at shuffle + skew: I'd check the Spark UI for a stage where one task runs far longer than the rest (skew) and for partitions exceeding execution memory (spill). Fix skew by broadcasting the small side if applicable, salting the hot key, or enabling AQE skew-join; fix spill by raising the partition count (smaller working sets) rather than just memory. The millions of small files come from too many output partitions × tasks — I'd repartition/coalesce to target 128 MB–1 GB files before write (or write to a table format and compact). That same small-file problem is why Athena scans too much and costs too much (P10): few large, well-partitioned, columnar files let Athena prune and skip. So the through-line is: tame the shuffle/skew to fix runtime, fix file sizing to fix both the write and the downstream scan cost.

Q: When does adding executor memory NOT fix spill? When the real problem is too few partitions. Spill happens when a single task's working set exceeds its execution-memory share; if you have 200 partitions over a 1 TB shuffle, each task handles ~5 GB and will spill regardless of reasonable heap sizes. Doubling memory delays it; doubling the partition count halves each task's working set and often eliminates it. So my first move on spill is "are partitions sized to ~128–256 MB?" before "give it more RAM" — more memory is the expensive fix for a free one.

Q: Broadcast vs sort-merge join — how do you decide, and what's the trap? If one side fits the broadcast threshold (default 10 MB, often worth raising for medium dimensions), broadcast it — no shuffle, dramatically faster. Otherwise sort-merge (shuffle both, sort, merge). The trap is stale statistics: Spark might not broadcast because it thinks a side is bigger than it is (or AQE hasn't kicked in), so I ANALYZE TABLE to refresh stats or hint the broadcast explicitly. The bigger trap is broadcasting something that's not actually small and OOMing the executors — confirm the size, don't guess.

Q: How do you make a backfill safe? Idempotent, reproducible, resumable, auditable. Idempotent: write to a staging path with an atomic commit, or use a table format (Iceberg/Delta) whose commit is atomic and overwrite by partition — so a retry can't duplicate. Reproducible: deterministic logic keyed on event time (P01), so a rerun produces identical output. Resumable: partition by date and reprocess only the failed partitions. Auditable: record which partitions were rewritten and from what input version. The acceptance test is the replay test — run it twice, diff the output, expect zero.

References

  • Spark: The Definitive Guide (Chambers & Zaharia); Learning Spark, 2e
  • High Performance Spark (Karau & Warren) — skew, shuffle, memory in depth
  • Spark SQL performance tuning & AQE
  • Spark tuning guide — memory, serialization
  • Zaharia et al., Resilient Distributed Datasets (NSDI 2012) — the RDD/lineage paper
  • PMC track Phase 07 WARMUP — partitioning/shuffle foundations