Chapter 21 — Key Takeaways (Apache Spark)

The page to have open next to the Spark UI.

First, do you need it?

Spark's overheads are fixed and do not shrink with cluster size. A job that takes 8 s in DuckDB takes 90 in Spark, and most of the 90 is not compute.

💸 Kestrel's dim_product job — 340 MB in, 180 MB out, hourly — cost $6,307 a year on Spark, with cluster startup at 67% of elapsed time. The counterweight is real: moving it added a second execution engine to the DAG.

Three non-size reasons that are legitimate: the transform is not expressible in SQL · you need one codebase for 1 GB and 4 TB · your organization already runs Spark, and people are oddly embarrassed by the third.

The model, in four nouns

Driver plans and collects — one JVM, not your cluster's memory. Executors hold cores and heap. A partition is one task on one core — the number of partitions is the number of tasks. A stage is work with no data movement; stage boundaries are shuffles.

Lazy evaluation means the failing line is almost never the wrong line.

Narrow and wide

Narrow (select, filter, withColumn, union) — one input partition per output. Free, fused. Wide (groupBy, join, distinct, orderBy, repartition, any PARTITION BY window) — a shuffle: write every partition to disk, read the pieces back over the network.

Count the Exchange nodes. That is the job's cost in one number. Five stages down to three removes roughly forty percent of the real cost, and no cluster sizing achieves the same thing.

🔎 Read a plan in this order — and note it is not the order it prints in:

1. Exchange nodes — the shuffle count
2. Join strategies — BroadcastHashJoin cheap, SortMergeJoin two shuffles
3. PushedFilters — empty means you are reading everything
4. only then partition counts and memory ← where most people start, because that is where the knob is

Partitions

$$\text{target} \approx 128\text{–}200\ \text{MB each} \qquad\text{or}\qquad 2\text{–}4 \times \text{cores} \qquad\textbf{take the larger}$$

341 GB across 192 cores → 2,131 (size wins). 2 GB across 192 cores → 576 (cores win).

⚠️ spark.sql.shuffle.partitions = 200 has caused more slow Spark jobs than any other setting. At 341 GB it gives 1.7 GB partitions and spills on every task. The tell: sort tasks by Shuffle Spill (Disk). Non-zero on most tasks means partitions are too large — five seconds, and almost nobody sorts by it.

coalesce versus repartition

⚠️ coalesce(n) limits the parallelism of every narrow transformation between it and the nearest upstream shuffle. It reads like formatting. It is a constraint on the stage.

The same line is fine in one job and disastrous in the next. The tell: stage count unchanged, task count collapsed — adjacent numbers on the job page, and the second is the one nobody reads.

Size the write separately from the compute, and compute it from the data — a hardcoded file count is right until a 6.28× day.

Skew

📏 Skew is a property of the key, not the data size. A bigger cluster does not help.

The fifteen-second diagnostic: median task duration versus max. ~1–3× even · ~3–10× mild · >10× skew, and capacity will not fix it.

Sources, in order of frequency: a null or default key · a genuine heavy hitter · a low-cardinality key · a truncated timestamp.

Three fixes, in this order:

1. Filter the pathological key — cheapest, and frequently a modelling correction
2. Broadcast the other side — removes the partitioning entirely
3. Salt — split the heavy key N ways; replicates the other side N times, so it trades a skewed shuffle for a bigger one

⚠️ Not every aggregate survives salting. SUM/COUNT/MIN/MAX/collect_set decompose. COUNT(DISTINCT) and percentiles do not. AVG is the dangerous one: averaging partial averages runs, returns a plausibly wrong number, and fails nothing.

📐 Skew is often a modelling signal. Kestrel's heavy key was a distributor modelled as a retail customer — distorting AOV, cohort retention, a top-customers report, two dashboards, and an SCD2 snapshot. The Spark job was the only consumer whose failure mode was loud. Salting works either way, which is exactly why the question stops being asked.

AQE (on by default from 3.2)

Does: coalesce small shuffle partitions · switch join strategies at runtime · split skewed join partitions.

🧭 Does not: skew in a groupBy · skew in the source before any shuffle · oversized partitions outside the skew-join path. And it cannot help a job with no shuffle at all, because it works from shuffle statistics.

Memory: four different failures, one message

⚠️ Raising executor memory buys a reprieve for two, nothing at scale for the first, and literally nothing for skew. The diagnostic is three questions: where did it fail, when, and how many tasks.

Execution and storage memory share one pool, which is why a cache() added last month is a plausible cause of a join failing today.

UDFs

built-in expression      1×      compiled, fused, optimizable
pandas / Arrow UDF     ~2-5×     batched serialization
plain Python UDF      ~10-100×   per-row round trip, optimizer blind

📏 The gap between rows two and three decides jobs. And check row one first — a surprising share of production UDFs reimplement something in pyspark.sql.functions.

The tell: a Filter sitting above a BatchEvalPython node when you wrote it below. That is the optimizer telling you it gave up.

Caching

Helps only when a DataFrame is used more than once and recomputing costs more than storing. That is the entire condition, and most cache() calls do not meet it.

It is lazy (a cache followed by one action was populated and never read), it evicts shuffle memory, and the upstream is often cheap anyway.

Reading and writing

Confirm pushdown in the plan — PushedFilters: [] means you are reading everything. A filter wrapped in a Python UDF cannot push down.

The small-files problem is about the accumulated directory. 730,000 files × 467 KB cost $0.29 in GET requests and 39.9 minutes of per-file overhead — 0.63 s each, and superlinear, because throttling retries grow with the request rate. OPTIMIZE cost $80.64 once and took a full-year read from 51 minutes to 5.8.

The two case studies, compressed

47 minutes, one core. Two cluster doublings took cost from $13.44 to $90.24 a night and runtime from 28 minutes to 47. Median 1.1 s, max 46 minutes — a ratio of 2,531×. Nobody looked at the one number that would have stopped it, because "slow job, add capacity" is right most of the time.

The optimization that serialized the cluster. .coalesce(1), applied correctly to a real problem, ran the whole job on 1 of 192 cores for eleven days at 4.2× duration — and it succeeded every night, finishing inside the SLA.

🏭 A regression that stays inside the SLA reports nothing. Alert on duration change against a trailing median, and publish the margin as a tracked number. The margin is what you are actually protecting, and a pass/fail check cannot see it.