Case Study 1: Forty-Seven Minutes, One Core

"We doubled the cluster and it cost twice as much and finished at exactly the same time. That is when somebody finally opened the UI."

Executive Summary

Kestrel's nightly agg_customer_revenue Spark job took 47 minutes and had been getting slower for a year. The response, twice, had been to add executors — from 12 to 24, then to 48.

Neither helped. The cost went from $13.44 a night to $90.24 and the runtime moved from 28 minutes to 47.

The stage was skewed. 1,999 of 2,000 tasks finished in about a second; one ran for 46 minutes, because one wholesale customer holds roughly 8% of Kestrel's order lines and the job grouped by customer_id. A bigger cluster gives that task more neighbours, not more cores.

Salting cut the job to 9 minutes. A year later the salting was removed, because the real problem was that a distributor had been modelled as a peer of a retail shopper — a mistake that was distorting six models, of which the Spark job was the only one that complained.

Skills applied: reading a Spark stage (§21.2, §21.5); the median-versus-max diagnostic (§21.5); salting (§21.7); skew as a modelling signal (§21.7's 📐 callout).

Background

The job. agg_customer_revenue reads a year of silver.order_items — 6,483,117 rows — joins to silver.orders, and aggregates to one row per customer with revenue, order count, and a set of derived measures.

(order_items
   .join(orders, "order_id")
   .groupBy("customer_id")
   .agg(F.sum("net_revenue_cents").alias("revenue_cents"),
        F.countDistinct("order_id").alias("n_orders"),
        F.min("ordered_at").alias("first_order_at"),
        F.max("ordered_at").alias("last_order_at"))
   .write.mode("overwrite").parquet(f"{LAKE}/gold/agg_customer_revenue/"))

Nothing about that is wrong. It is the obvious code, it is correct, and it would be the same fourteen lines in any tutorial.

The history.

2025-03   12 executors    28 min    $13.44/night
2025-09   12 executors    44 min    $21.12/night   ← "it's getting slower"
2025-10   24 executors    51 min    $48.96/night   ← doubled. WORSE.
2026-01   48 executors    47 min    $90.24/night   ← doubled again.

Doubling twice made it slower once and barely faster the second time, and the cost went up 6.7×. Each change was made because the previous one had not worked, which is a reasonable-sounding sequence that nobody in it enjoyed.

The Problem

The job was on the critical path to the 6am SLA (Chapter 1 §1.7), and at 47 minutes it left too little margin. The escalation was a capacity request for a further doubling.

The escalation was denied, and the reason given was the useful part: the previous two doublings were in the record, and neither had produced the effect that justified them. Somebody was asked to explain the 24-to-48 result before another instance was approved.

The Analysis

Step 1: open the stage. Fifteen seconds.

Stage 4 (groupBy)   2000/2000 tasks    Duration: 46.4 min
──────────────────────────────────────────────────────────
  Task duration     min 0.3 s   25th 0.9 s   median 1.1 s
                    75th 1.4 s  max 2,784 s     ← 46.4 min
  Shuffle read      min 4 MB    median 21 MB    max 41.2 GB
  Spill (disk)      0 on 1,999 tasks            18.4 GB on 1

$$\frac{\text{max}}{\text{median}} = \frac{2{,}784}{1.1} = 2{,}531\times$$

🔎 Read the Plan — the fifteen-second check that would have saved $86 a night

Median task duration versus maximum task duration. That is the entire diagnostic, it is on the stage page of every Spark UI, and it decides whether the next thing you do is worth doing.

ratio meaning does a bigger cluster help?
~1–3× even work yes
~3–10× mild imbalance partially
>10× skew no

At 2,531×, adding executors does exactly nothing: the 1,999 healthy tasks were never the constraint, and the one that matters is a single core processing a single partition.

Two doublings were approved without anyone looking at this number. Not through negligence — "the job is slow, add capacity" is a reasonable heuristic that is right most of the time. It is wrong for precisely one failure mode, and that failure mode is the most common one in Spark.

The habit worth building: before requesting capacity for a slow distributed job, paste the median and max task duration into the request. If you cannot, you have not looked. It takes fifteen seconds and it is the difference between an $86-a-night mistake and a morning's work.

Step 2: find the key.

(order_items.groupBy("customer_id").count()
    .orderBy(F.desc("count")).show(5))

customer_id     count       share
     4417       519,142      8.01%
    91002        18,330      0.28%
    22815        14,206      0.22%
    ...
   (median)          3       0.00%

Customer 4417 has 519,142 order lines. The median customer has three.

$$\frac{519{,}142}{6{,}483{,}117} = 8.01\%\ \text{of all order lines on one hash partition}$$

That is the 41.2 GB of shuffle read and the 18.4 GB of spill, and it is why the task took 46 minutes: one core, sorting and aggregating half a million rows with 41 GB of joined width, spilling most of it to disk.

Step 3: check whether it is one key or a distribution. This matters, because the fix differs.

top 1 customer     8.01% of rows
top 10             9.94%
top 100           12.31%
top 1,000         18.02%

One key, not a heavy tail. The distribution beyond 4417 is unremarkable. That points at a specific entity rather than at Zipf, and it is the first hint that the answer might not be a Spark answer.

Step 4: ask what customer 4417 is.

customer_id  4417
name         (a regional distributor)
segment      wholesale
orders       11,847 in FY2025      (median customer: 1.3)
lines/order  43.8                  (median: 2.7)

It is not a customer. It is a distributor placing consolidated orders on behalf of dozens of retail outlets, and it has been in the customers table since 2019 because that is where the application put it.

The Decision

The immediate fix was salting, because the SLA was at risk and the modelling conversation was going to take weeks.

SALT = 16      # 519,142 / 16 = 32,446 rows per task -- comparable to the
               # 21 MB median shuffle read. Chosen from the measurement
               # above, not from a default.

salted = order_items.withColumn(
    "salt", F.when(F.col("customer_id") == HEAVY_KEYS_BROADCAST,
                   (F.rand() * SALT).cast("int")).otherwise(F.lit(0)))

partial = (salted.groupBy("customer_id", "salt")
                 .agg(F.sum("net_revenue_cents").alias("rev"),
                      F.min("ordered_at").alias("first_at"),
                      F.max("ordered_at").alias("last_at")))

final = (partial.groupBy("customer_id")
                .agg(F.sum("rev").alias("revenue_cents"),
                     F.min("first_at").alias("first_order_at"),
                     F.max("last_at").alias("last_order_at")))

Two details that are easy to get wrong, and both were caught in review:

Only the heavy key is salted. Salting every key multiplies the shuffle for 1.9 million customers who did not need it. Salting selectively costs a broadcast lookup and keeps the fix proportionate.

countDistinct cannot be two-phase aggregated this way. SUM, MIN, and MAX decompose across salts — the sum of partial sums is the sum — but a distinct count does not. It was rewritten as approx_count_distinct for the salted path with a note in the model's description, and for the small number of keys where the exact figure was needed, computed separately.

⚠️ Failure Mode — not every aggregate survives salting

Salting splits a key into N groups and then re-aggregates. That is only correct for aggregates that decompose.

decomposes?
SUM, COUNT, MIN, MAX ✅ sum the sums, min the mins
AVG ⚠️ not directly — carry sum and count, divide at the end
COUNT(DISTINCT x) ❌ a value can appear under several salts
median, percentile ❌ not decomposable at all
collect_set ✅ union of unions

AVG is the dangerous row, because averaging the partial averages runs and returns a number that is wrong by an amount proportional to how uneven the salt groups are. It will be close. Nothing will fail, and the error is small enough to survive a spot check.

The general rule: before salting, list your aggregates and classify each one. If any is in the bottom half of that table, the salted path needs a different formulation, and "it produced a number" is not evidence that it produced the right one.

Result: 47 minutes → 9 minutes, and the cluster went back to 12 executors:

$$12 \times \tfrac{9}{60}\ \text{h} \times \$2.400 = \$4.32\ \text{a night, against } \$90.24$$

$$(\$90.24 - \$4.32) \times 365 = \mathbf{\$31{,}361\ \text{a year}}$$

And the runtime mattered more than the money. 47 minutes was the largest single block in the window before the 6am SLA; 9 minutes removed the schedule risk entirely, which is what the escalation had actually been about.

What Happened

The salting worked and was removed a year later.

The modelling conversation concluded that customer 4417 and the eleven other distributors in the table are a different kind of entity. customer_type was split, and wholesale orders were processed in a separate path.

The skew disappeared without any salting, because the two populations no longer shared a partitioning key. The job settled at 7 minutes, slightly faster than the salted version and considerably simpler.

📐 Design Decision — the salting was right, and removing it was also right

It is tempting to read this as "they should have fixed the model in the first place." That is the wrong lesson and it is worth being explicit about why.

The SLA was at risk in the current week. The modelling change touched six models, two dashboards, and a definition the finance team owns; it took eleven weeks and two meetings that could not be scheduled sooner. Salting bought the eleven weeks for one day of work, and the job would otherwise have missed the SLA every night in the interim.

The mistake would have been stopping there, and it is the more common one: a tactical fix that works removes the pressure that would have produced the structural one. Kestrel avoided it by filing the modelling issue at the same time as shipping the salt, with the salt's own code comment pointing at the ticket:

```python

SALTING customer 4417 because it is a distributor modelled as a customer.

This is a workaround for DATA-1847, not a design. If that ticket closes,

delete this and re-measure -- the skew should be gone.

```

A workaround with an expiry condition written into it is a different object from a workaround. That comment is why anyone went back.

The audit found the modelling problem was not confined to Spark. Once customer_type was separated, five other things changed:

  • Average order value had been $75.83 including a distributor whose average order is 43.8 lines. The retail-only figure is meaningfully different, and every "is AOV improving?" conversation for three years had been contaminated by one entity's ordering cadence.
  • Customer cohort retention counted the distributor as one retained customer, in every cohort.
  • The "top 10 customers by revenue" report was, in effect, a list of distributors.
  • Two dashboards' median-order-size figures moved by more than 4%.
  • The dim_customer SCD2 snapshot had been creating a version for 4417 on most days, because a distributor's attributes churn — a small instance of Chapter 20's Case Study 1.

None of those had produced a complaint. The Spark job was the only consumer of customer_id whose failure mode was loud.

Lessons

  1. Compare median task duration to maximum before requesting capacity. Above ~10× is skew and a bigger cluster does nothing. Fifteen seconds, on the stage page.

  2. "The job is slow, add capacity" is right most of the time and wrong for the single most common Spark failure. That is what makes it durable.

  3. Two doublings took the cost from $13.44 to $90.24 a night and the runtime from 28 minutes to 47. Nobody was careless; nobody looked at the one number that would have stopped it.

  4. Check whether the skew is one key or a distribution. One key points at an entity; a heavy tail points at Zipf. The fixes differ, and so does whether the fix is a Spark fix at all.

  5. Salt selectively. Salting every key multiplies the shuffle for everyone who did not need it.

  6. Not every aggregate survives salting. SUM/MIN/MAX decompose; COUNT(DISTINCT) and percentiles do not; AVG runs and returns a plausibly wrong number, which is the worst case.

  7. Skew is often a modelling signal. A key whose distribution breaks your execution engine is frequently conflating two kinds of thing — and salting works either way, which is exactly why the question stops being asked.

  8. The tactical fix was correct and so was removing it. It bought eleven weeks for one day's work. The mistake would have been stopping there.

  9. Write the expiry condition into the workaround. A comment naming the ticket and saying "delete this and re-measure" is why anyone came back.

  10. The Spark job was the only consumer that complained. Five other things were wrong — AOV, cohort retention, a top-customers report, two dashboards, and an SCD2 snapshot — and all of them were silent.

Questions for Discussion

  1. Each doubling was a reasonable response to the previous one having failed. What would break that sequence in your organization, and at what cost?

  2. The capacity request was denied on the grounds that the previous two had not produced their claimed effect. Is that a good gate? What does it cost when the request is legitimate?

  3. AVG under salting produces a number that is wrong by a small amount and fails nothing. Design the test that catches it. What does it cost to run?

  4. Salting was shipped with a comment naming a ticket and an expiry condition. How would you make that pattern survive across a team, rather than depending on one engineer's habit?

  5. AOV had been contaminated by one entity for three years and produced no complaint. What class of metric is most vulnerable to this, and how would you audit for it?

  6. The modelling fix took eleven weeks because it touched six models and a finance-owned definition. Is there a way to make that faster that does not amount to skipping the agreement?

  7. §21.7 says "Zipf is not a modelling error." Where is the line between a genuine heavy tail and a miscategorised entity, and how would you tell from data alone?