Case Study 2: The Optimization That Serialized the Cluster

"One line. It fixed the thing it was meant to fix, and it made twenty-three of twenty-four machines stop doing anything."

Executive Summary

silver.events had accumulated 730,000 Parquet files averaging 467 KB, and a full-year read had degraded to 51 minutes, most of it spent opening files rather than reading data. The small-files problem (Chapter 9 §9.7) had arrived exactly as advertised, one nightly job at a time.

The fix applied was .coalesce(1) before the write. It worked — one file a day instead of 2,000 — and it made the nightly conversion job go from 9 minutes to 38, because coalesce propagates its parallelism backwards through every narrow transformation to the previous shuffle boundary. There is no shuffle in that job. The entire read-and-parse of 11.48 GB of JSON ran on one core, with 23 of 24 executors idle for 38 minutes.

Cost went from $8.74 to $36.77 a night — $10,232 a year — to fix a problem whose actual solution was a repartition and a compaction job, both of which were already in Chapter 10.

Skills applied: coalesce versus repartition (§21.4); sizing the write separately from the compute (§21.12); the small-files problem (Chapter 9 §9.7); Delta OPTIMIZE (Chapter 10 §10.5).

Background

The job. json_to_parquet reads a day of raw clickstream from the landing zone — 11.48 GB of JSON, 14,000,000 events (Chapter 1 §1.5) — parses it with an explicit schema, applies four regexp_extract calls to normalize user agents, filters bot traffic, and writes Parquet.

Every transformation in it is narrow. There is no groupBy, no join, no orderBy. That fact is the entire case study and nobody noticed it until afterwards.

(spark.read.schema(EVENT_SCHEMA).json(f"{RAW}/events/dt={day}/")
   .withColumn("browser", F.regexp_extract("user_agent", BROWSER_RE, 1))
   .withColumn("os",      F.regexp_extract("user_agent", OS_RE, 1))
   .filter(~F.col("user_agent").rlike(BOT_RE))
   .write.mode("overwrite").parquet(f"{LAKE}/silver/events/dt={day}/"))

9 minutes on 24 executors × 8 cores. Roughly 2,000 input splits, roughly 2,000 output files.

The problem it created, accumulating for a year:

silver/events/            730,000 files      467 KB average
                          341 GB total
full-year read            51 min   (theoretical, on bytes alone: ~4 min)
  of which listing        6 min

The money was never the issue. 730,000 GET requests at the frozen $0.0004/1,000 is $0.29. The time was. The 51 minutes decomposes, and the decomposition is the useful part:

$$\underbrace{6\ \text{min}}_{\text{listing}} + \underbrace{\frac{730{,}000 \times 0.63\ \text{s}}{192\ \text{cores}} = 39.9\ \text{min}} _{\text{per-file overhead}} + \underbrace{\sim\!4\ \text{min}}_{\text{actually reading 341 GB}} \approx 50\ \text{min}$$

0.63 seconds of overhead per 467 KB file, measured — two S3 round trips (Parquet needs the footer before the data), task scheduling, and, past a few hundred thousand requests against one prefix, S3 throttling with exponential backoff. The last term is the one that makes small files superlinear rather than merely wasteful: the more files you have, the more the retries cost per file.

The Problem

An engineer was asked to fix the small-files problem. They read §21.4, understood that coalesce avoids a shuffle and is therefore cheaper than repartition, and applied it:

   .coalesce(1)                    # ← one file per day instead of 2,000
   .write.mode("overwrite").parquet(...)

The reasoning was sound at every step. coalesce does avoid a shuffle. One file is better than 2,000. The change was one line, reviewed, and merged.

The next morning:

json_to_parquet     2026-05-14   09:12  →  succeeded
json_to_parquet     2026-05-15   38:04  →  succeeded

Succeeded. No error, no warning, no failed test. The output was correct — one 934 MB Parquet file containing exactly the right rows.

The Analysis

Step 1: the stage page, which is unmistakable.

Stage 0 (json → parquet)   1/1 tasks      Duration: 38.3 min
──────────────────────────────────────────────────────────────
  Tasks                    1
  Input                    11.5 GB
  Executors used           1 of 24
  Cluster cores busy       1 of 192          (0.52%)

One task. Not one slow task among two thousand — Case Study 1's shape — but one task, total.

$$\frac{11{,}480\ \text{MB}}{2{,}296\ \text{s}} = 5.0\ \text{MB/s}$$

Which is about what one core does when it is decompressing, parsing JSON against a 41-field schema, and running four regular expressions per row. The core was not idle. It was working perfectly, alone.

Step 2: understand why, which requires knowing one thing about coalesce.

BEFORE                                AFTER  .coalesce(1)
──────────────────────                ──────────────────────
Stage 0: 2,000 tasks                  Stage 0: 1 task
  read  ─┐                              read  ─┐
  parse  ├─ 2,000-way                   parse  ├─ ONE-way
  filter─┘                              filter─┘
                                        coalesce
write: 2,000 files                    write: 1 file

coalesce does not insert a shuffle. That is its whole selling point, and it is why this happens.

Without a shuffle there is no stage boundary, so the reduced partition count is not a property of the write — it is a property of the entire stage, and the stage extends backwards through every narrow transformation until it hits a shuffle or the source. This job has no shuffle. So the stage is the whole job.

⚠️ Failure Mode — coalesce sets the parallelism of everything upstream of it

This is the most consequential thing in §21.4 and it is not what the API name suggests. coalesce reads like a formatting operation applied at the end. It is a constraint on the stage.

```python df.filter(...).withColumn(...).coalesce(1).write.parquet(...)

^^^^^^^^^^^^^^^^^^^^^^^^^^ all of this now runs with one task

```

The rule: coalesce(n) limits the parallelism of every narrow transformation between it and the nearest upstream shuffle.

Which means its danger is a property of the pipeline's shape, not of the number you pass:

Pipeline coalesce(1) before the write
read → filter → write catastrophic — the whole job is one task
read → groupBy → write harmless — the pre-shuffle stage keeps its parallelism
read → join → filter → write harmless for the join, serial for the filter

So the same line is fine in one job and disastrous in the next, and the difference is whether there is a shuffle between the source and the coalesce. That is a thing you have to go and look at, which is why this keeps happening.

The tell: a job whose stage count did not change but whose task count collapsed. Both numbers are on the job page, next to each other, and the second is the one nobody reads.

Step 3: confirm the alternative. repartition(4) was measured against the same day:

                       tasks   duration   files   file size   cost/night
coalesce(1)                1     38.3 min      1     934 MB      $36.77
repartition(4)         2,000      9.6 min      4     234 MB       $9.22
(original)             2,000      9.1 min  2,000     467 KB       $8.74

repartition(4) costs 33 seconds more than the original — that is the shuffle — and produces four files instead of two thousand.

$$(\$36.77 - \$9.22) \times 365 = \$10{,}056\ \text{a year, for the wrong line}$$

The Decision

Three changes, and only the first is about this job.

One: repartition(4), sized from the data rather than chosen.

# 341 GB/year / 365 = 934 MB/day. Four files of ~234 MB sits in the
# 128-512 MB band (Ch. 9 §9.7) with room for a heavy day. Re-derive this
# if daily volume moves more than 50%.
TARGET_FILE_MB = 256
n_files = max(1, round(daily_mb / TARGET_FILE_MB))
df.repartition(n_files).write.parquet(...)

Computed, not constant. A hardcoded 4 is correct until Black Friday, when the day is 6.28× normal and four files become 1.5 GB each.

Two: compact the 730,000 historical files. This is the part the coalesce was never going to solve, because it only affected new writes.

silver.events is a Delta table (Chapter 10), so the fix is one statement:

OPTIMIZE silver.events WHERE dt >= '2025-05-01';

files before   730,412
files after      1,341        (average 254 MB)
full-year read    51 min  →  5.8 min
compaction cost   24 executors x 1.4 h x $2.400 = $80.64,  once

Three: a file-count assertion, so the next accumulation is caught in a week rather than a year:

# Fails the build if a day's write produces more than 12 files. Normal is
# 4; Black Friday is 6.28x volume and would be ~23 -- so the threshold is
# per-GB, not per-day.
assert n_files <= max(12, round(daily_mb / 128)), (
    "write produced %d files for %.0f MB -- check the repartition" % ...)

📐 Design Decision — why not let AQE handle it?

spark.sql.adaptive.coalescePartitions.enabled merges small shuffle partitions automatically (§21.8), and the obvious question is why any of this is manual.

Because AQE coalesces shuffle partitions, and this job has no shuffle. AQE works from statistics produced at a shuffle boundary; a narrow read-filter-write pipeline never produces any, so there is nothing for AQE to act on. It is not that AQE was disabled. It is that AQE had no opinion.

This is the second of the three gaps §21.8's 🧭 callout lists, met in production, and it is worth generalizing: AQE is a runtime optimizer over shuffle statistics. A job with no shuffle is a job AQE cannot help.

The corollary is useful in both directions. Adding repartition(4) does introduce a shuffle — and once there is a shuffle, AQE's other behaviours become available for the rest of the job. The 33 seconds the shuffle costs buys more than four files.

What Happened

Original coalesce(1) repartition(n)
Nightly duration 9.1 min 38.3 min 9.6 min
Tasks 2,000 1 2,000
Cores busy 192 1 (0.52%) 192
Nightly cost $8.74 $36.77 $9.22
Files/day 2,000 1 4
Full-year read 51 min 51 min* 5.8 min†

* the coalesce only affected new writes; the 730,000 historical files were untouched. † after OPTIMIZE.

The coalesce was live for eleven days, costing $308 and, more importantly, pushing the conversion job's finish time from 03:14 to 03:43 — which was still inside the window, which is why nothing alerted.

🏭 From the Pipeline — a regression that stays inside the SLA is a regression nobody reports

This job got 4.2× slower and the only automated signal was that it still succeeded.

Duration is the most under-monitored property of a data pipeline. Every team alerts on failure, most alert on SLA breach, and almost none alert on change. A job that goes from 9 minutes to 38 and finishes at 03:43 against an 06:00 deadline has consumed most of its margin and reported nothing.

The margin is the thing you are actually protecting, and it is invisible in a pass/fail check.

Two cheap controls:

  • Alert on duration change, not duration. A run more than 2× its trailing 14-day median is worth a look, at any absolute duration. This is Chapter 25 §25.4's material and it costs nothing to add.
  • Publish the margin. "This DAG finished 2h17m before its deadline" as a tracked number. When it trends down over a quarter, you find out in the quarter rather than on the night it goes negative.

The version of this that hurts: eleven days of 4.2× regression that nobody reported, followed by Black Friday at 6.28× volume, which would have taken this job past 06:00 on its own.

The audit found coalesce in six other jobs. Four were harmless — a shuffle sat between the source and the coalesce, so only the trivial post-shuffle stage was serialized. Two were the same bug, one of them costing 22 idle executor-minutes a night since the previous September.

And one job had the opposite problem: repartition(2000) before a write on a job producing 40 MB of output, creating 2,000 files of 20 KB every hour. Someone had copied the partition count from the clickstream job, where it is correct, into a job where the data is four orders of magnitude smaller. That directory held 1.7 million files.

Lessons

  1. coalesce(n) limits the parallelism of every narrow transformation between it and the nearest upstream shuffle. It is a constraint on the stage, not a formatting step at the end.

  2. Whether that is catastrophic depends on the pipeline's shape, not on the number. With a shuffle in between it is harmless; in a read → filter → write job it serializes everything.

  3. The tell is a job whose stage count did not change but whose task count collapsed. Both numbers are on the job page, next to each other.

  4. repartition costs a shuffle and is usually the right answer — here, 33 seconds for four files instead of two thousand.

  5. Size the write from the data, every run. A hardcoded file count is correct until a 6.28× day.

  6. coalesce only affects new writes. The 730,000 accumulated files needed OPTIMIZE, which cost $80.64 once and took the full-year read from 51 minutes to 5.8.

  7. The small-files problem is about the accumulated directory, not one day's write. $0.29 of GET requests, and 39.9 minutes of per-file overhead — 0.63 s per 467 KB file, which grows per file as throttling retries kick in, so the cost is superlinear rather than merely wasteful.

  8. AQE cannot help a job with no shuffle, because it works from shuffle statistics. Adding the repartition creates the boundary that makes AQE's other behaviours available.

  9. 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.

  10. The margin is what you are protecting, and a pass/fail check cannot see it.

  11. Copying a partition count between jobs is the same mistake in the other direction. repartition(2000) on 40 MB produced 1.7 million files.

Questions for Discussion

  1. The engineer read §21.4, understood coalesce correctly, and applied it. What would the section have to say to prevent this — and is that achievable without making every optimization note unreadable?

  2. The job succeeded for eleven days at 4.2× its normal duration. What is the cheapest change that would have surfaced it on day one?

  3. "Alert on duration change, not duration" will fire on legitimate changes — a backfill, a schema addition, a busy day. How would you make it useful rather than noisy?

  4. The OPTIMIZE cost $80.64 once and saved 45 minutes on every full-year read. How would you decide how often to run compaction, and what would you measure?

  5. Four of the six other coalesce uses were harmless. Would you remove them anyway? Argue both sides.

  6. Someone copied repartition(2000) into a job with 40 MB of data. What makes a number like that dangerous to copy, and how should the original have been written to discourage it?

  7. Case Study 1's fix was right and later removed; this one's fix was wrong and looked right. What distinguishes a plausible-but-wrong optimization from a correct one, before you measure?