Chapter 21: Apache Spark: Distributed Data Processing at Scale

"Most Spark jobs are slow for one of four reasons, and three of them are the same reason."

Overview

Spark is the default answer to "the data does not fit in one machine," and it is the wrong answer more often than its popularity suggests. This chapter covers when to reach for it, the execution model you must hold in your head to debug it, and the small number of failure modes that account for nearly all production Spark pain.

The organizing fact is this: Spark's difficulty is not the API. The DataFrame API is a pleasant SQL-shaped thing you can learn in an afternoon. The difficulty is that the API hides a distributed execution model, and every performance problem you will have is a property of that model rather than of your code. A job that reads correctly and returns the right answer can take forty minutes instead of four because of a decision the API did not ask you to make.

So this chapter spends most of its length on the model — partitions, stages, shuffles, and skew — and comparatively little on syntax. Chapter 4's material on distributed systems is the foundation, and Chapter 4 §4.7 promised the salting treatment that arrives here in §21.7.

Chapter 20 ended by handing over partition alignment as "the dominant concern," which is a fair summary of everything below.

21.1 Do You Need Spark?

The honest starting question, and it has become much easier to answer wrongly over the last few years because single-machine tools got dramatically better while Spark's reputation stayed the same.

A modern single machine handles more than most people think. A 64-core cloud instance with 512 GB of RAM costs about $3 an hour and will process a hundred gigabytes of Parquet with DuckDB or Polars in minutes, single-threaded code and all. Chapter 22 covers those tools properly.

Spark's overheads are real and fixed. A cluster takes 60–180 seconds to start. Every shuffle writes to disk and reads back. The JVM's serialization costs are paid on every boundary. A job that takes 8 seconds in DuckDB routinely takes 90 in Spark, and the 90 does not shrink with a bigger cluster because most of it is not compute.

Three reasons to use Spark that are not about size, and all three are legitimate:

The data is already in a lake and the transformation is not expressible in SQL. A Python UDF over a hundred million rows is a Spark job.

You need the same code to run on 1 GB in a test and 4 TB in production. Spark's scale-out is transparent in a way a single-machine tool's is not.

Your organization already runs Spark, and the marginal cost of one more job is far below the cost of introducing a second execution engine with its own deployment, monitoring, and on-call story. This is the most common good reason, and people are oddly embarrassed by it.

💸 Cost Check — the job that should not have been a Spark job

Kestrel's dim_product refresh reads 340 MB of Parquet, joins three small tables, and writes 180 MB. It ran on Spark because everything ran on Spark.

Spark: 4 executors, minimum billable, cluster start plus 90 seconds of work:

$$4 \times \tfrac{4.5}{60}\ \text{h} \times \$2.400 = \$0.72\ \text{per run}$$

DuckDB, on the orchestrator's own worker, 11 seconds: $0.003.

It ran hourly. $0.72 × 24 × 365 = $6,307 a year to avoid a pip install.

The cluster startup was 67% of the elapsed time, which is the shape to look for: when the fixed overhead dominates, you are paying for distribution you are not using.

The honest counterweight: moving it introduced a second execution engine into the DAG, with its own container image, its own failure modes, and one more thing for the on-call engineer to know. Kestrel judged that worth $6,307. A team of two might reasonably not, and that is a real argument rather than a rationalization.

21.2 The Execution Model, in the Only Detail That Matters

Four nouns. If you can place a performance problem against these, you can debug Spark; if you cannot, you are guessing.

                 ┌─────────────────────────────────────────┐
   your code ───▶│ DRIVER   plans, schedules, collects      │
                 └────────────────┬────────────────────────┘
                                  │  tasks
              ┌───────────────────┼───────────────────┐
              ▼                   ▼                   ▼
        ┌───────────┐       ┌───────────┐       ┌───────────┐
        │ EXECUTOR  │       │ EXECUTOR  │       │ EXECUTOR  │
        │ ┌───┬───┐ │       │ ┌───┬───┐ │       │ ┌───┬───┐ │
        │ │ P │ P │ │       │ │ P │ P │ │       │ │ P │ P │ │  ← PARTITIONS
        │ └───┴───┘ │       │ └───┴───┘ │       │ └───┴───┘ │
        └───────────┘       └───────────┘       └───────────┘

The driver runs your program, builds the plan, and schedules work. It is a single point of failure and a single point of memory pressure. collect() brings data to the driver, and the driver's heap is not your cluster's memory — it is one JVM, usually the smallest one you provisioned.

Executors are JVMs holding memory and CPU slots. A task occupies one core.

A partition is the unit of parallelism: a chunk of data processed by one task on one core. The number of partitions is the number of tasks, and this is the single most consequential number in any Spark job.

A stage is a set of tasks that can run without moving data between executors. Stage boundaries are shuffles, which is why counting stages tells you how many times your data crossed the network.

Lazy evaluation ties it together. Transformations (select, filter, join, groupBy) build a plan and do nothing. Actions (count, write, collect, show) execute it. This is why a malformed transformation appears to succeed and then fails, confusingly, at the write — and why the line number in a stack trace is almost never the line that is wrong.

21.3 Narrow and Wide

Every transformation is one of two kinds, and the distinction predicts your job's cost better than anything else you could know about it.

Narrow: each output partition depends on one input partition. select, filter, withColumn, map, union. No data moves. They are effectively free and they fuse together into one pass.

Wide: each output partition depends on many input partitions. groupBy, join, distinct, orderBy, repartition, and every window function with a PARTITION BY. Data crosses the network, and that crossing is a shuffle.

NARROW                              WIDE  (a shuffle)
P1 ──▶ P1'                          P1 ──┐   ┌──▶ P1'
P2 ──▶ P2'                          P2 ──┼─╳─┼──▶ P2'
P3 ──▶ P3'                          P3 ──┘   └──▶ P3'
one input, one output               every input feeds every output

A shuffle writes every partition to local disk, then every executor reads the pieces it needs over the network. It is the most expensive thing Spark does, by a wide margin, and it is the thing your code does not visibly ask for.

Counting shuffles is the highest-value habit in this chapter:

df.explain(mode="formatted")   # look for Exchange -- each one is a shuffle

Every Exchange node is a shuffle. A five-stage job has four of them. If you can reduce five stages to three, you have removed roughly forty percent of the job's real cost, and no amount of cluster sizing achieves the same thing.

🔎 Read the Plan — three shuffles where one would do

python (events .groupBy("session_id").agg(F.count("*").alias("n")) # shuffle 1 .join(sessions, "session_id") # shuffle 2 .groupBy("customer_id").agg(F.sum("n").alias("total")) # shuffle 3 .orderBy(F.desc("total"))) # shuffle 4

Four shuffles. Two are avoidable:

• Shuffle 2 disappears if sessions is small enough to broadcast (§21.6). At Kestrel it is 700,000 rows and about 40 MB — comfortably broadcastable, and Spark will do it automatically if the statistics are present and will not if they are missing.
• Shuffle 1 and 3 can sometimes merge if the data is already partitioned by customer_id, because a groupBy on the partitioning column needs no exchange.

The general reading order for a Spark plan, and it is not the order the plan prints in:

1. Count the Exchange nodes. That is the job's cost in one number.
2. Find the join strategies. BroadcastHashJoin is cheap; SortMergeJoin means two shuffles.
3. Check PushedFilters on the scans. An empty list means you are reading everything.
4. Only then look at partition counts and memory.

Most people start at step 4, because it is the one with a knob attached.

21.4 Partitions: The Number That Decides Everything

Too few partitions and you have idle cores and partitions too large for executor memory. Too many and scheduling overhead dominates — each task has a fixed cost of a few milliseconds, and 200,000 tasks of 8 ms each is more scheduler than work.

The rule of thumb, and it is genuinely just a rule of thumb:

$$\text{target partition size} \approx 128\ \text{MB}\ \text{to}\ 200\ \text{MB in memory}$$

$$\text{partitions} \approx 2\ \text{to}\ 4 \times \text{total cores}$$

Take the larger. For Kestrel's clickstream — 341 GB of Parquet for the year, 24 executors × 8 cores = 192 cores:

$$\frac{341{,}000\ \text{MB}}{160\ \text{MB}} = 2{,}131 \quad\text{versus}\quad 3 \times 192 = 576$$

2,131, because partition size wins when the data is large. Spark's default spark.sql.shuffle.partitions is 200, which would give 1.7 GB partitions and spill catastrophically.

⚠️ Failure Mode — spark.sql.shuffle.partitions = 200

This default has caused more slow Spark jobs than any other single setting. It was chosen when clusters were small, it is not adaptive without AQE, and it applies to every shuffle in the job regardless of the data's size.

At 20 GB shuffled, 200 partitions is 100 MB each. Fine. At 341 GB, it is 1.7 GB each. Every task spills to disk, and the job takes six times longer than it should while every dashboard shows the cluster as busy. At 200 MB, it is 1 MB each — 200 tasks doing nothing, dominated by scheduling.

Set it per job, from the data, or enable AQE (§21.8) and let Spark coalesce after the fact.

The tell in the UI: open the stage, sort tasks by duration, and look at Shuffle Spill (Disk). Any non-zero spill on a majority of tasks means your partitions are too large. That one column diagnoses this in about five seconds and most people have never sorted by it.

repartition versus coalesce:

df.repartition(2000)              # full shuffle. Even distribution. Expensive.
df.repartition("customer_id")     # shuffle by key -- co-locates a key's rows
df.coalesce(10)                   # NO shuffle. Merges partitions. Can be very uneven.

coalesce is cheap and dangerous. It avoids the shuffle by merging adjacent partitions, which means it cannot balance them — and, worse, it propagates its parallelism backwards. A .coalesce(1) before a write does not just write one file; it makes the entire upstream stage run with one task. Jobs that mysteriously stopped using the cluster are usually this.

Use coalesce to reduce file count after a filter that removed most rows. Use repartition when you need balance.

21.5 Skew: The Failure That Looks Like Slowness

Skew is uneven partition sizes, and it is the most common Spark performance problem in any system with real-world data, because real-world data is never uniform.

The symptom is unmistakable once you know it:

Stage 7: 2000/2000 tasks    Duration: 47 min
  Task duration:  min 0.4 s   median 1.1 s   p75 1.4 s   max 46 min
                                                          ^^^^^^

1,999 tasks finished in a second. One ran for 46 minutes, and the stage is not done until it is. The cluster is 99.95% idle for most of the stage, and every dashboard shows the job as running normally.

Where skew comes from, in order of how often you will meet it:

A null or default key. customer_id is null for guest checkouts, so every guest order hashes to the same partition. At Kestrel this is 4.2% of orders — 100,000 rows in one partition against 50 in each of the others.

A genuine heavy hitter. Kestrel's largest wholesale customer accounts for about 8% of order lines (Chapter 7 §7.4). Grouping by customer_id puts 8% of the data on one core.

A low-cardinality key. Grouping by country when 71% of your traffic is one country.

A timestamp truncated to a low-resolution unit, so a backfill puts a year of rows on one date.

📏 Scale Note — skew is a property of the join key, not of the data size

A skewed job does not get better with a bigger cluster, and this is the single most expensive misunderstanding in this chapter.

Doubling the executors halves the time for the 1,999 healthy tasks — which were never the problem — and does nothing at all to the one that takes 46 minutes, because that task is one core processing one partition and no amount of hardware splits it.

The observable consequence is a job that costs twice as much and finishes at the same time, which is reliably diagnosed as "Spark is slow."

The check, before you resize anything: open the slowest stage in the UI and compare the median task duration to the max. A ratio above about 10× is skew and a bigger cluster will not help. A ratio near 1 is a genuine capacity problem and it will.

That one comparison decides whether the next thing you do is worth doing, and it takes fifteen seconds.

21.6 Joins: Broadcast, Sort-Merge, and the One That Explodes

Sort-merge join is the default for two large tables: shuffle both sides by the join key, sort each partition, merge. Two shuffles, and it is correct at any scale.

Broadcast hash join sends the small side to every executor in full. Zero shuffles — the large side is read where it already is.

from pyspark.sql import functions as F
big.join(F.broadcast(small), "customer_id")

Spark broadcasts automatically when it believes the small side is under spark.sql.autoBroadcastJoinThreshold (default 10 MB). It only believes that when statistics exist, and they frequently do not — a Parquet directory written by another job, a view, or anything behind a UDF often has no size estimate, so Spark falls back to sort-merge on a 3 MB table. Hinting explicitly is cheap insurance.

The limit is the driver and the executor heap. A broadcast is collected to the driver and then sent to every executor, so a 2 GB broadcast on a 4 GB driver is an OutOfMemoryError that names the driver rather than the join.

⚠️ Failure Mode — the join that multiplies

python orders.join(order_items, "order_id") # fine: 1-to-many, expected orders.join(promotions, "promotion_id") # is promotion_id unique in promotions?

If the right side has duplicate keys, the output has more rows than the input, and Spark will do it without comment. This is Chapter 6 §6.9's fan-out, and at Spark's scale it turns a 40-minute job into an out-of-memory failure four hours in.

Assert the grain of the right side before joining. Two lines:

python assert promotions.count() == promotions.select("promotion_id").distinct().count()

That costs a full pass and is still worth it on any join whose right side you did not build yourself in the same job. In production, make it a data test on the source rather than a runtime assert — but write the assert while developing, because the alternative is discovering it from a memory error whose message mentions neither the join nor the table.

21.7 Salting, and the Other Skew Fixes

Chapter 4 §4.7 deferred this. Three fixes, in the order you should try them.

Fix 1: filter the pathological key. If the skew is a null or a sentinel, handle it separately. This is the cheapest fix by a wide margin and it is frequently the right one:

# Guest checkouts have no customer. They do not belong in a per-customer
# aggregate at all, so the "fix" is a modelling correction.
real = orders.filter(F.col("customer_id").isNotNull())
guests = orders.filter(F.col("customer_id").isNull())   # aggregate separately

Fix 2: broadcast the other side. If the skew is in a join and the other side is small, broadcasting eliminates the shuffle entirely and skew stops mattering — there is no partitioning by key any more.

Fix 3: salt the key. When the skew is genuine, the key is needed, and the other side is large.

SALT = 16

# Add a random salt to the skewed side, splitting each key into 16 keys.
left = big.withColumn("salt", (F.rand() * SALT).cast("int"))

# Replicate the small side 16 times, once per salt value, so every salted
# key still finds its match.
right = (small
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(SALT)]))))

joined = left.join(right, ["customer_id", "salt"]).drop("salt")

What that does: the heavy key's 100,000 rows spread across 16 partitions instead of one. The 46- minute task becomes sixteen 3-minute tasks that run in parallel.

What it costs, and both halves matter:

The replicated side gets 16× larger. If small is 40 MB it becomes 640 MB, which is fine. If it is 4 GB it becomes 64 GB, which is not. Salting trades a skewed shuffle for a bigger one, and above a certain size that is a bad trade.

The code becomes harder to read and the salt factor is a magic number that was right for the data distribution on the day it was chosen. Write down the measurement that produced it.

📐 Design Decision — salt, or fix the model?

Kestrel's skew came from one wholesale customer holding 8% of order lines. Salting worked and cut the job from 47 minutes to 9.

A year later they removed the salting, because the underlying issue was a modelling one: that customer is not a customer in the sense the rest of the model means. They are a distributor placing consolidated orders, and treating them as a peer of a retail shopper distorted six different models, of which the Spark job was the only one that complained.

Splitting customer_type into retail and wholesale and processing them separately removed the skew, simplified four models, and made three reports correct that nobody had known were wrong.

The general point: skew is often a modelling signal. A key whose distribution breaks your execution engine is frequently a key that is conflating two different kinds of thing. Before you salt, ask whether the heavy key belongs in the same population as the others — and note that salting works either way, which is exactly why the question stops being asked.

When salting is genuinely right: the distribution is naturally heavy-tailed and the entities really are the same kind of thing. Web traffic by URL, events by device model, transactions by merchant. Zipf is not a modelling error.

21.8 Adaptive Query Execution

AQE re-plans the query at runtime using statistics from completed stages, and it is on by default from Spark 3.2. It addresses three of this chapter's problems directly.

Coalescing shuffle partitions. After a shuffle, AQE looks at the actual partition sizes and merges small ones. This is what makes spark.sql.shuffle.partitions = 200 survivable — set it high and let AQE reduce it.

Switching join strategies. If a side turns out smaller than estimated, AQE converts a sort-merge join to a broadcast join mid-flight. This rescues the missing-statistics case from §21.6.

Skew join handling. AQE detects partitions much larger than the median and splits them automatically — salting, done by the engine.

spark.conf.set("spark.sql.adaptive.enabled", True)
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", True)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)

🧭 Version Note — Spark 3.5.x, and what AQE does not do

This book pins PySpark 3.5.3 and delta-spark 3.2.1 (requirements.txt). AQE is on by default from 3.2; a great deal of skew-tuning advice online predates it and describes manual work the engine now does.

Three things AQE does not fix, and mistaking its coverage is the current version of the old mistake:

• Skew in a groupBy, as opposed to a join. skewJoin is named accurately. An aggregation on a heavy key is still one task.
• Skew present in the source data before any shuffle. AQE works from shuffle statistics; a skewed read has none yet.
• A partition too large to fit in memory in the first place. AQE coalesces small partitions; it does not split large ones outside the skew-join path.

So §21.7 is not obsolete. AQE has made it needed less often, which is a different thing, and the cases it does not cover are the ones you will meet on a Tuesday.

21.9 Memory, and the Four Ways a Job Runs Out of It

OutOfMemoryError is Spark's least informative failure, because four unrelated problems produce it and the message names none of them. Learning to tell them apart from the shape of the failure is worth more than any memory setting.

One: the driver collected too much.

rows = df.collect()               # brings the ENTIRE DataFrame to one JVM
pdf  = df.toPandas()              # same thing, wearing a hat
print(df.count(), df.take(5))     # fine -- bounded

The tell: the stack trace is on the driver, the executors are idle, and it fails at an action rather than during a stage. The driver's heap is not your cluster's memory. It is one JVM, usually the smallest thing you provisioned, and collect() on a 40 GB DataFrame will not become survivable by adding executors.

Two: a partition does not fit. §21.4. One task must hold its partition plus working memory; at 1.7 GB partitions on a 4 GB executor running 4 tasks, it cannot.

The tell: it fails inside a stage, on a minority of tasks, and the stage shows heavy disk spill before it dies. Fix the partition count, not the memory.

Three: a broadcast was too large. §21.6. The small side is collected to the driver and then held in full on every executor.

The tell: it fails at the start of a join stage, and the plan says BroadcastHashJoin on something you did not expect to be small. A broadcast that Spark chose from a bad statistic is the usual cause, which is why autoBroadcastJoinThreshold is worth lowering rather than raising.

Four: skew. §21.5. One partition is fifty times the others and the executor holding it dies while the rest of the cluster idles.

The tell: 1,999 tasks succeeded and one failed, twice, and then the stage failed. The retry is the giveaway — Spark reruns the failed task on another executor, which also dies, because the problem travelled with the data.

⚠️ Failure Mode — raising executor memory is the wrong first response to three of the four

It is the first thing everyone tries, it occasionally works, and the occasions when it works are the ones where it was cheapest to be patient.

Cause	What memory buys	What actually fixes it
driver collect()	nothing at real scale	do not collect; write instead
oversized partition	a temporary reprieve	more partitions
oversized broadcast	a temporary reprieve	disable the broadcast, or shrink the side
skew	nothing	§21.7

For skew it buys literally nothing, because the failing task's partition grows with the data and the ratio to the others is unchanged.

The diagnostic order is fixed and takes two minutes: where did it fail (driver or executor), when (action, stage start, mid-stage), and how many tasks failed (one, some, all). Those three answers identify which of the four it is, every time, and each has a different fix.

A useful habit: Spark's unified memory manager splits an executor's heap between execution (shuffles, joins, sorts) and storage (cached data), and they borrow from each other. That is why §21.11's caching can cause an out-of-memory failure in a shuffle — the two are drawing on one pool, and a cache() added last month is a plausible cause of a join failing today.

21.10 UDFs: The Cost That Is Not in the Plan

A user-defined function is the escape hatch from Spark's built-ins, and it is expensive in a way the query plan does not show you.

@F.udf("double")
def score(views, purchases):              # a plain Python UDF
    return (purchases or 0) / max(views or 1, 1)

What that costs per row. The data lives in the JVM. For each row, Spark serializes the arguments, sends them to a Python worker process, waits, and deserializes the result back. The arithmetic is trivial; the round trip is not, and it happens once per row.

Catalyst cannot see inside it either, so the optimizer stops: no predicate pushdown through it, no constant folding, no expression fusion. A filter written after a UDF cannot be moved before it, even when it obviously could.

Three tiers, in the order to try them:

Built-in expressions. Always first. Spark's built-ins compile to JVM bytecode and cost roughly nothing. The example above is a built-in expression:

df.withColumn("score", F.col("purchases") / F.greatest(F.col("views"), F.lit(1)))

Pandas (Arrow) UDFs. When the logic genuinely is not expressible in built-ins. These pass a whole batch through Arrow, so the serialization cost is amortized across thousands of rows instead of paid per row, and the function receives a Series:

@F.pandas_udf("double")
def score(views: pd.Series, purchases: pd.Series) -> pd.Series:
    return purchases.fillna(0) / views.fillna(1).clip(lower=1)

Plain Python UDFs. Last resort — an external library call, a model inference, a parser with no vectorized form.

📏 Scale Note — the order-of-magnitude worth remembering

The precise numbers depend on the row width, the Python version, and what the function does, so measure your own. The magnitudes are stable enough to plan with:

text 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 the second and third rows is the one that decides jobs. Converting a plain UDF to a pandas UDF is usually a ten-line change and frequently the largest single speedup available in a job that has one.

And check the first row before either. A surprising share of production UDFs reimplement something in pyspark.sql.functions — date arithmetic, string splitting, null coalescing, JSON extraction — written by someone who did not know the built-in existed. Read the function list once, properly. It is an hour, and it pays back on the first job.

The tell in the plan: a BatchEvalPython or ArrowEvalPython node, and a Filter sitting above it that you wrote below it. That second detail is the optimizer telling you it gave up.

21.11 Caching, and Why It Usually Does Not Help

df.cache()          # == persist(MEMORY_AND_DISK)
df.persist(StorageLevel.MEMORY_ONLY)
df.unpersist()

Caching is the most over-applied optimization in Spark, and the reason is that it feels like it should always help.

It helps when a DataFrame is used more than once in the same job and recomputing it costs more than storing it. That is the entire condition.

It hurts, or does nothing, in the common cases:

Used once. The overwhelming majority. Caching adds a serialization pass and buys nothing.

Cached data evicts executor memory that the shuffle needed, converting a fast job into a spilling one. The job gets slower and the cause is the thing you added to speed it up.

It is lazy. df.cache() marks the DataFrame; nothing is stored until an action runs. A cache followed by one action is a cache that was populated and never read.

The upstream might be cheap. Caching a filtered Parquet read saves you a Parquet read, which with predicate pushdown may be nearly free.

🧪 Try It — measure before you cache

Take a job you believe benefits from caching. Run it three ways and record wall-clock and peak executor memory:

1. As written.
2. With every cache() removed.
3. With caching only on DataFrames used two or more times, which you find by counting references rather than by intuition.

In the author's experience the third is usually fastest and the second is usually faster than the first. Do not take that on trust — the point of the exercise is that you can settle it for your own job in twenty minutes, and that almost nobody has.

If you find caching helping where this section says it should not, you have found something worth understanding rather than a contradiction: the likely cause is an expensive upstream, a non- deterministic source, or a UDF, and knowing which is more valuable than the speedup.

21.12 Reading and Writing: Where the Time Actually Goes

Predicate pushdown and partition pruning are the difference between reading 341 GB and reading 900 MB, and both depend on how the data was written (Chapter 9 §9.4, Chapter 11 §11.5).

# Reads only the matching date directories. Nothing else is touched.
spark.read.parquet("s3://lake/events/") \
     .filter(F.col("event_date") == "2026-03-17")

Confirm it rather than assuming:

df.explain(mode="formatted")
# PushedFilters: [IsNotNull(event_date), EqualTo(event_date,2026-03-17)]
# ...an EMPTY PushedFilters list means you are reading everything.

A filter that cannot push down — one wrapped in a Python UDF, or applied to a column the file format cannot index — reads the whole dataset and then discards most of it. The plan says so plainly and nobody looks.

The small files problem, from the other direction:

# 2,000 partitions -> 2,000 files, each 400 KB. Every subsequent read pays
# 2,000 S3 round trips of ~30 ms latency. Ch. 9 §9.7.
df.write.parquet("s3://lake/output/")

# Aim for 128-512 MB files.
df.coalesce(20).write.parquet("s3://lake/output/")

Sizing the write is a separate decision from sizing the compute, and using one number for both is the usual mistake. You may want 2,000 partitions to compute and 20 files to store.

🎓 Interview Angle — "your Spark job is slow. Walk me through it."

An open diagnostic question, and it is scored on the order you do things.

The weak answer starts tuning: more executors, more memory, spark.sql.shuffle.partitions. That is step 4 of §21.3's reading order, and starting there is the tell.

The strong answer reads before it changes:

"First I'd look at the stage view sorted by duration and check whether it's one task or all of them — if one task is forty minutes and 1,999 finished in a second, it's skew and nothing about cluster size will help. Assuming it's not, I'd count the Exchange nodes in the plan, because shuffles are the cost and the number of them is the one summary of the job. Then the scan: is PushedFilters empty, and is the partition predicate actually pruning — the byte count in the UI, not the plan. Only after that would I look at partition counts and memory, and I'd check the disk spill column before changing anything, because non-zero spill on most tasks means the partitions are too big and that's a different fix from too few."

Four things that answer does. It checks skew first, because skew is immune to everything else. It reads the UI rather than the plan for the byte count (Exercise 21.18). It counts shuffles. And it names spill as the diagnostic for partition sizing rather than guessing at a number.

The follow-ups:

"How would you fix the skew?" — filter the pathological key, broadcast the other side, salt. In that order, and salting last because it is the only one that changes the code.

"What does coalesce(1) do before a write?" — the propagation question (§21.4). A candidate who knows it makes the whole upstream stage single-threaded has debugged a four-hour job.

"When would you not use Spark?" — under a few hundred gigabytes. Naming DuckDB or Polars and a measurement is the answer; "when the data is small" without a number is not.

And the question behind the question is whether you have read a Spark UI under pressure. Mentioning the stage view's task-duration distribution, or the Shuffle Spill (Disk) column, does more than any amount of configuration vocabulary.

🏭 From the Pipeline — the job that got slower every week for four months

A nightly aggregation ran in 22 minutes in January and 68 minutes in May. Data volume grew 9% over the same period.

Every investigation looked at the wrong thing, because the natural hypothesis — more data, more time — was almost true and produced a plausible answer each time somebody asked.

What was actually happening:

text January the input was 1,712 files, average 198 MB May the input was 41,206 files, average 8.2 MB

A streaming writer upstream had been switched from a five-minute to a thirty-second commit interval in February, for an unrelated latency requirement. Nothing about the data changed; the file count went up 24×.

The Spark job's time was not in compute. It was in listing 41,206 objects, opening 41,206 files, and reading 41,206 footers — and the task count exploded, because spark.sql.files.maxPartitionBytes cannot merge files smaller than itself across a listing that large without cost.

The tell was in the UI and nobody had looked: the scan stage's task count had gone from 1,712 to 41,206 while its input bytes were nearly unchanged.

The fix was not in the Spark job. A compaction step after the streaming write, producing 128 MB files, took the job back to 24 minutes — which is 22 minutes plus the 9% of genuine growth.

Three lessons, and the third is the reason this is in Chapter 21 rather than Chapter 9.

A plausible explanation with no measurement is the most expensive kind. "More data" was accepted four times.

Task count against input bytes is the diagnostic, and it takes ten seconds. A rising task count with flat bytes is always a layout change, and there is no other cause.

And the change that caused it was upstream, correct, and made by somebody else — which is why the investigation kept looking inside the job. A Spark job's performance is mostly a property of its input's layout, and the layout is owned by whoever writes it.

🔁 Idempotency Check — Spark's write modes, and the one that is a trap

A Spark job will be retried. The task will be retried by Spark, the stage will be retried, and the whole job will be retried by the orchestrator — so the write must survive all three.

```python df.write.mode("append") # NOT idempotent. Ever. df.write.mode("overwrite") # depends entirely on the next line spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic") df.write.mode("overwrite").partitionBy("event_date").parquet(path)

-> replaces ONLY the partitions present in the DataFrame. Idempotent.

```

The default for partitionOverwriteMode is static, which means overwrite deletes the entire table and writes the partitions you supplied. A job re-run for one day, with the default, deletes every other day.

That is the trap, and it is a one-word configuration difference between "replace yesterday" and "delete everything except yesterday."

Three more things that are not idempotent and look like they are:

A task retry with a non-deterministic transformation. Spark re-executes the failed task, and if the transformation involves rand(), a shuffle whose input order changed, or a monotonically_ increasing_id(), the retried task produces different rows from the ones it replaced — and the other tasks' output is already written.

foreachPartition writing to an external system. Spark's write modes do not apply; each partition writes independently and a retried partition writes twice. The idempotency has to be at the sink, keyed on something stable.

And saveAsTable on a table another job is reading. There is a window in which the table is partially replaced, and nothing in the API tells you. This is Chapter 10's argument, and it is the single strongest practical reason for a table format under a Spark job.

The test is Chapter 20 §20.12's: run the job twice over the same interval and EXCEPT the output in both directions. On a distributed writer, run it three times — the second run tests the overwrite and the third tests it after the table has already been overwritten once, which is a different code path in some sinks.

🔐 Privacy & Governance — shuffle spills write your data to disk, unencrypted by default

A shuffle writes every partition to local disk before anything reads it back, and the local disk of an ephemeral executor is a place nobody has classified.

text where a Spark job leaves data outside your tables ───────────────────────────────────────────────────────────────────── shuffle spill files every wide operation, on the executor's local storage broadcast blocks a small dimension, replicated to every node cached RDDs / DataFrames MEMORY_AND_DISK spills to the same place the event log query plans, and any literal in them the driver's stdout a `collect()` and a `print` is a table in a log aggregator checkpoint directories intentionally durable, and often forgotten

Three controls, and the first is a single setting people do not know exists:

spark.io.encryption.enabled = true encrypts shuffle and spill data at rest on the local disk. It is off by default. On a job processing personal data on ephemeral cloud instances, that default is the finding, and it costs a few percent of throughput.

Set the checkpoint and scratch directories deliberately, on storage you have classified, with a lifecycle rule. A checkpoint directory is a durable copy of intermediate state — Chapter 29 says it cannot be moved and must not be lost, which also means it must be included in Chapter 31's manifest.

And never collect() a DataFrame containing identifiers into a log. The driver's stdout goes to the orchestrator (Chapter 24's 🔐) and to the cluster's log aggregator, both of which have wider access and different retention from the warehouse.

The point worth carrying: a distributed job multiplies the number of places data exists, by the number of executors, and every one of those places is outside the storage layer that your governance tooling knows about. Encryption at rest on the compute layer is not a paranoid setting — it is the only control that covers all six rows at once.

🏭 From the Pipeline — the UDF that cost nine minutes a night

A Python UDF parsed a user-agent string into three columns. It was forty lines, well tested, and readable. It ran on 14 million rows a night and took nine minutes.

The plan showed nothing wrong. One stage, no shuffle, no spill. The cost was invisible because a Python UDF's cost is not in the plan — it appears as a slow scan-and-project stage with no explanation attached.

What actually happens per row: the JVM serialises the row to the Python worker, the worker deserialises it, runs your function, serialises the result, and the JVM deserialises it back. Four serialisation boundaries per row, 14 million times, and the parsing itself is a rounding error next to them.

Three fixes, measured on the same data:

text a row-at-a-time Python UDF 9 m 12 s a pandas / Arrow UDF (vectorised, batched) 1 m 41 s 5.5x native Spark SQL expressions (regexp_extract x3) 19 s 29.1x

The native version is 29× faster and it is also uglier — three regexp_extract calls instead of forty readable lines of Python. That is a real trade and it should be made deliberately, with the measurement in a comment above it (§21.14's form).

Two things generalise past UDFs.

A cost that does not appear in the plan is the hardest kind to find, and Spark has exactly three: Python UDFs, a collect() into the driver, and a slow external call inside a foreachPartition. All three look like a slow stage with no explanation, and all three are invisible to every tuning knob.

And the vectorised middle option is usually the right answer. 5.5× for a change that keeps the Python readable is a better trade than 29× for something nobody will maintain — which is the opposite of the conclusion a benchmark alone would produce, and it is why the comment matters more than the number.

🧭 Version Note — AQE changed which advice is still true

Adaptive Query Execution went from experimental to on-by-default in Spark 3.2, and it invalidated a great deal of tuning advice that is still widely repeated.

text advice status now ───────────────────────────────────────────────────────────────────────── "set spark.sql.shuffle.partitions carefully" MOSTLY OBSOLETE. AQE coalesces after the shuffle, from actual sizes. Set it high and let AQE reduce it. "salt every skewed join" often unnecessary; AQE's skew join handling splits large partitions "broadcast hints everywhere" AQE converts sort-merge to broadcast at RUNTIME, from real sizes "collect statistics with ANALYZE" still useful, and less critical "count the Exchange nodes" UNCHANGED, and still the highest-value habit in the chapter

Two things to take from that table.

AQE fixes the problems caused by bad estimates, not the ones caused by bad queries. It cannot remove a shuffle you asked for, cannot prune a partition your predicate hid behind a CAST, and cannot make a Python UDF fast. Every finding in §21.3 through §21.5 survives it.

And the plan you read is now the plan before adaptation. df.explain() shows the initial physical plan; the executed plan can differ. AdaptiveSparkPlan isFinalPlan=false in the output is the tell — to see what actually ran, read the SQL tab in the UI after the query finishes, not the plan before it.

The three settings worth enabling explicitly, each with a comment saying what it buys (Exercise 21.23c): adaptive.enabled, adaptive.coalescePartitions.enabled, and adaptive.skewJoin.enabled. All three default on in recent versions and all three are turned off by somebody's inherited configuration in about a third of deployments.

21.13 The Kestrel Platform

🧱 Kestrel Platform — Increment 21: the clickstream sessionizer on Spark

Chapter 18 §18.9's sessionization, at 14,000,000 events a day, as a Spark job.

text platform/spark/ sessionize.py ← the job conf/spark-defaults.conf tests/test_sessionize.py

The job, in outline:

```python events = (spark.read.parquet(f"{LAKE}/events/") # Ch. 18 CS2's overlap: read back, emit forward. The 2 hours is # measured, not chosen -- see Ch. 20 §20.7. .filter((F.col("event_ts") >= start - two_hours) & (F.col("event_ts") < end)))

w = Window.partitionBy("anonymous_id").orderBy("event_ts") sessions = (events .withColumn("gap", F.col("event_ts").cast("long") - F.lag("event_ts").over(w).cast("long")) .withColumn("is_new", F.when(F.col("gap").isNull() | (F.col("gap") > 1800), 1).otherwise(0)) .withColumn("session_seq", F.sum("is_new").over( w.rowsBetween(Window.unboundedPreceding, Window.currentRow))) .groupBy("anonymous_id", "session_seq") .agg(F.min("event_ts").alias("session_start"), ...) .filter((F.col("session_start") >= start) # emit forward only & (F.col("session_start") < end))) ```

Six things this increment must get right:

1. Both window specifications share partitionBy("anonymous_id").orderBy("event_ts"), so the job has one shuffle for the sessionization rather than three. Chapter 18 §18.10, in Spark. Verify by counting Exchange nodes in the plan.
2. spark.sql.shuffle.partitions is set from the data, not left at 200. One day of events is 11.48 GB of JSON; state the arithmetic in a comment.
3. AQE is on, all three settings, and the config file says why each one.
4. The overlap filter and the emit filter are both present. Chapter 18 Case Study 2 — without the second, the overlap creates duplicates instead of fixing phantoms.
5. The write is coalesced to a file size, separately from the compute partition count.
6. A skew check runs before the aggregation: median versus max partition size, failing the build above 10×. Bot traffic concentrated on one anonymous_id is the realistic cause and it happens.

The exercise that matters is 21.23(e): run the job with spark.sql.shuffle.partitions at 200 and at the computed value, and report elapsed time, disk spill, and cost. The gap is the chapter.

21.14 Summary

Ask whether you need Spark. A 64-core machine with DuckDB handles more than most people think, and Spark's fixed overheads do not shrink with cluster size. Kestrel's dim_product job cost $6,307 a year to avoid a pip install, with cluster startup as 74% of elapsed time.

Driver, executor, partition, stage. A partition is one task on one core; a stage boundary is a shuffle; lazy evaluation means the failing line is rarely the wrong line.

Narrow moves no data; wide shuffles. Count the Exchange nodes — that is the job's cost in one number, and reducing five stages to three removes about forty percent of it.

Read a plan in this order: exchanges, join strategies, pushed filters, and only then partition counts. Most people start at the end, because that is where the knob is.

Partitions: ~128–200 MB each, or 2–4× cores, whichever is larger. The default of 200 is the single most common cause of slow Spark jobs. Non-zero disk spill on most tasks means partitions are too large, and that column diagnoses it in five seconds.

coalesce propagates its parallelism backwards. .coalesce(1) before a write makes the whole upstream stage single-threaded.

Skew is a property of the key, not the data size, and a bigger cluster does not help. Compare median to max task duration: above ~10× is skew; near 1× is capacity.

Fix skew by filtering the pathological key, then by broadcasting, then by salting — in that order. Salting replicates the other side N times and trades a skewed shuffle for a larger one.

Skew is often a modelling signal. Kestrel's heavy key was a distributor miscategorised as a customer, distorting six models of which the Spark job was the only one that complained.

AQE handles coalescing, join-strategy switching, and skewed joins — but not skewed groupBy, not skew in the source before any shuffle, and not oversized partitions outside the skew-join path.

Four unrelated problems produce OutOfMemoryError, and where it failed, when, and how many tasks identify which. Raising executor memory is the wrong first response to three of the four, and buys literally nothing for skew.

A plain Python UDF pays a per-row round trip and blinds the optimizer — roughly 10–100× a built-in, against 2–5× for a pandas UDF. Check the built-in list first; a surprising share of production UDFs reimplement something in pyspark.sql.functions.

Caching helps only when a DataFrame is used more than once. It is lazy, it evicts shuffle memory, and it is the most over-applied optimization here.

Confirm predicate pushdown in the plan. An empty PushedFilters list means you are reading everything, and the plan says so plainly.

Size the write separately from the compute. 2,000 partitions to compute and 20 files to store is a normal answer.

Chapter 22 takes the other branch of §21.1: pandas, Polars, and DuckDB, and how to tell which of the three a given job wants.

Key terms: driver · executor · partition · task · stage · UDF · pandas UDF · unified memory · narrow transformation · wide transformation · shuffle · spill · skew · salting · broadcast join · sort-merge join · adaptive query execution · lazy evaluation · predicate pushdown · partition pruning · small files problem · coalesce · repartition · dynamic allocation