ComputeGENERALSIMULATEDENGINE-SPECIFIC

Data Skew

Real key distributions are not uniform. When one value holds most of the rows, the partitioner faithfully sends them all to one task — and that task becomes the job.

What actually happens How to build it Can I trust it?▶ Run the lab

Who needs this, what one row is, and why the obvious build breaks

Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.

The question

Every task in this stage finished in seconds except one, which has been running for forty minutes. Why, and what does the cluster size have to do with it?

Who needs this

The schedule downstream of this job, and the on-call engineer watching a stage sit at 999 of 1000. Skew does not usually make data wrong; it makes it late, and late data is a broken promise with a green pipeline behind it (The Freshness SLO).

What one row is

The unit is the partition after a shuffle, measured in rows or bytes. Skew is a property of the distribution across those partitions — specifically the ratio of the largest to the mean, which is the number that predicts the stage duration.

The obvious build

Group by the natural business key — country, tenant, customer, product — because that is what the question asks for. The distribution of that key is a property of the business, not something the pipeline chose, and for most keys it is fine.

Why it breaks

One country, one tenant or one merchant holds the large majority of rows. Its partition is orders of magnitude bigger than the others, and the stage takes as long as that one task (Hot Keys: When Aggregate Metrics Hide a Saturated Node).

How it breaks with real data

One country, one tenant or one merchant holds the large majority of rows. Its partition is orders of magnitude bigger than the others, and the stage takes as long as that one task (Hot Keys: When Aggregate Metrics Hide a Saturated Node).
The join key is nullable. Every null hashes to the same destination, so a single task receives every unmatched row in the dataset — and the more broken the upstream join, the worse the skew gets.
A migration set a default of -1 or unknown on a foreign key. That default is now the largest key in the table, and it is not a real entity at all (Nullability & Defaults).
A backfill re-processes a year into one job. Partitioning by date is fine day to day, and the day the outage was replayed holds twelve months of arrivals (What Backfills Break).
Adding executors changes nothing, because the work is not divisible. The cluster is bigger and the longest task is exactly as long (Amdahl's Law).

Source→Ingestion→Raw→Transformation→Validation→Storage Model→Serving→Consumer→Observability

What is actually happening

A shuffle assigns each row a destination by hashing its key. The assignment is deterministic and even *across keys*, which is not the same as even across rows: if one key holds most of the rows, its destination holds most of the rows (The Shuffle).
The stage is a barrier, so its duration is the duration of its slowest task. A distribution with one dominant key converts a thousand-way parallel stage into a one-way stage with 999 idle observers (Stages and Tasks).
The oversized task is not merely slower in proportion. Once its data exceeds the memory its slot was given it starts spilling, so its cost grows faster than its share of the rows (Memory Pressure, Swap and the OOM Killer).
Skew is a property of the *key you chose*, not of the dataset. The same rows grouped by (country, day) and by country have completely different distributions, and one of them may be perfectly even (Partition Cardinality).
Skew is also a moving target. Distributions shift as a business grows, as a large customer signs, as a default value is introduced by a migration. A job that has been balanced for a year can become skewed overnight with no code change (Volume Anomalies).

What an uneven key looks like

SIMULATEDBoth the bar chart and the ratios are computed from the declared mixes in src/de/sim/pipeline.ts and asserted by scripts/de-sim.test.ts, which pins that the skewed run has a straggler ratio above three and produces revenue identical to the truth. They are a model's outputs, not a benchmark.

The picture below is the whole lesson. Eight countries, the same total number of orders in both cases, and one difference: in the second, a single key holds the large majority of the rows.

The numbers come from the model in src/de/sim/pipeline.ts, which generates a deterministic set of orders under a declared country mix. Its normal mix has a largest share of 30%; its skewed mix has a largest share of 80%. Nothing here was measured on a cluster — it is arithmetic on a distribution the model states openly, which is the only kind of magnitude this domain publishes.

The number that matters is the ratio of the largest partition to the mean, because that ratio is what the stage duration follows. With eight even-ish keys the largest partition is a small multiple of the mean. With one key at 80% of eight, the largest partition carries roughly six times the mean, and the stage takes correspondingly longer than the work it contains would suggest.

Notice what is *not* wrong in the skewed case. Every order is present exactly once, every amount is correct, and reconciliation against the source matches to the cent. The model asserts this in its tests: the consequence of skew is a straggler, not a wrong number.

What skew actually costs, relative to a balanced run of the same work

Idle executors waiting at the barrier

The whole cluster is billed while one task runs. This is the largest term and the one that no configuration change addresses (Compute Waste).

Spill on the oversized task

The dominant partition exceeds its slot's memory, so it writes and re-reads local disk that the balanced case never touches.

Retry cost when the big task fails

A task that dies on memory after twenty minutes is retried, and each attempt pays the same twenty minutes before failing again.

Extra fetch concentration

One reducing task pulls a disproportionate share of every map output, so its network and merge work is concentrated rather than spread.

Total bytes processed

Unchanged. Skew moves the same rows — this is why "the job got slower but the data did not grow" is the characteristic report.

Relative weights, not measurements. The ordering is what transfers between platforms; the magnitudes do not, and this domain never publishes a price.

Relative weights for a stage with one dominant key, to establish an ordering rather than to predict a runtime. The bottom row is the teaching: the work did not change, only its distribution.

Rows per shuffle partition, 4,000 orders over 8 country keys   [SIMULATED: src/de/sim/pipeline.ts]

NORMAL mix                              SKEWED mix
US  ████████████████████  30%           US  ████████████████████████████████████████████████████  80%
DE  ███████████           17%           DE  ███                                                    5%
GB  ████████              13%           GB  ██                                                     4%
FR  ███████               11%           FR  ██                                                     3%
PL  ██████                 9%           PL  ██                                                     3%
ES  █████                  8%           ES  █                                                      2%
IT  ████                   7%           IT  █                                                      2%
NL  ███                    5%           NL  ·                                                      1%

largest / mean  =  2.4x                 largest / mean  =  6.4x
stage duration follows the largest bar, not the total.

Where skew comes from, and which ones are bugs

It is worth separating skew that reflects reality from skew that reflects a defect, because the remedies are different and only one of them is a data quality problem.

Natural skew is a property of the business. A marketplace has a biggest seller; a SaaS product has a biggest tenant; a European retailer has a biggest country. Nothing is wrong and nothing upstream can be fixed. The remedy is a processing one — a composite key, a broadcast, salting.

Degenerate skew is a defect. Nulls, empty strings and sentinel defaults are not entities, and grouping millions of rows under unknown produces a partition that is large and meaningless at the same time. The remedy is upstream: fix the join, filter the sentinel, or replace it with a randomised value so at least it spreads (The Dimensions of Data Quality).

The third category is self-inflicted and the easiest to miss: a backfill or a replay that concentrates a long period into a single partition of a job that is otherwise perfectly balanced day to day.

Diagnosing a skewed stage

Trigger	Symptom	Cause	Response
One task with far more input bytes than its peers.	Stage stuck at n-1 of n tasks; cluster idle.	A genuinely dominant business key on the shuffle key.	Composite key if the grain allows, broadcast if the other side is small, salting if neither (Salting a Skewed Key).
One task holding rows whose key is null, empty or a sentinel.	The largest partition's key is `null`, `-1` or `unknown`.	A broken upstream join or a migration default, not a business reality.	Fix it upstream and add a test that fails on unexpected sentinel volume. Filtering it in this job hides a data defect (Data Tests).
Every task takes the same input bytes; one is still slow.	Even distribution, uneven durations.	Not skew. A degraded machine, a noisy neighbour, or a garbage-collection pause (Straggler Tasks).	Look at executor-level metrics rather than at the data. Speculative execution helps this case and not the one above.
A backfill run over a long range.	A stage that is balanced in daily runs and wildly skewed in the backfill.	The date key that spreads work across days concentrates when a year is processed as one job.	Bound the backfill by range and run it as several jobs, which also makes it restartable (Planning a Backfill).
A distribution alert fires during a genuine business surge.	A shape alarm on data that is entirely correct.	The check compares against historical shape, which a real change also violates.	Treat the alert as "look at this", never as "the data is wrong", and pair it with a completeness check that can distinguish the two (Quality Alerting).

Why more workers cannot help

This is the point at which the intuition built on web services fails, and it is worth being precise about why. A web service under load has many independent requests, so more capacity serves more of them at once. A skewed stage has one indivisible task that is bigger than the others, and no amount of capacity divides it.

Formally this is the same observation Amdahl made about the serial portion of a program: the parallel part shrinks with more processors and the serial part does not, so the total time approaches the serial part as a floor (Amdahl's Law). A dominant partition is the serial part of a data job, and adding executors moves the ceiling on the parallel part that was never the constraint.

The practical consequence is a checklist rather than a config change. Either make the work divisible — by changing the key, by salting, by splitting the hot key into its own job — or remove the redistribution that concentrated it, by broadcasting the other side. Everything else is spending money on idle capacity.

Two responses to a stage stuck at 999 of 1000

Double the cluster

Add executors, resubmit, and observe that the stage still takes as long. The 999 fast tasks now finish even faster, which changes nothing, because they were never what the stage was waiting for. The bill doubles and the schedule does not move.

Make the work divisible or unnecessary

Compare max and median task input bytes to confirm the cause is data rather than a machine. Then pick the remedy the data allows: a composite key if the grain permits, a broadcast if the other side is small enough after filtering, salting on the identified hot key if neither, or a separate job for the dominant key if it is a single well-known tenant.

A task is the smallest unit the scheduler can place, so a single oversized task sets a floor on the stage duration that capacity cannot lower. Every effective remedy either splits that unit into several, or removes the redistribution that created it — those are the only two mechanisms available (Straggler Tasks).

Find the skew before the job finds it for you

1-- Run this on the source before choosing a shuffle key.
2SELECT country,
3       COUNT(*)                                        AS rows,
4       COUNT(*) * 1.0 / SUM(COUNT(*)) OVER ()          AS share,
5       COUNT(*) * 1.0 / AVG(COUNT(*)) OVER ()          AS times_the_mean
6FROM   orders
7WHERE  dt = DATE '2026-08-25'
8GROUP  BY country
9ORDER  BY rows DESC
10LIMIT  20;
11 
12-- Read two things:
13--   share          -- is one key most of the data?
14--   times_the_mean -- this is the straggler ratio your stage will have
15--
16-- Then check the degenerate cases explicitly, because they hide in the
17-- same column and mean something completely different:
18SELECT COUNT(*) FILTER (WHERE country IS NULL)              AS null_key,
19       COUNT(*) FILTER (WHERE country IN ('', 'unknown'))   AS sentinel_key
20FROM   orders
21WHERE  dt = DATE '2026-08-25';

The second query is the one people skip. A dominant real key is a processing problem; a dominant null or sentinel is a data defect that happens to look identical in the first query (Data Tests).

How to build it

Most important first.

Measure the distribution before choosing the key. A count per key on the source, ordered descending, answers in one query what a week of tuning will not (Aggregation: COUNT, SUM, AVG, GROUP BY, HAVING).
Handle the degenerate keys explicitly. Nulls and sentinel defaults should be filtered, replaced with a random distribution, or routed down their own path, because they are not entities and grouping by them means nothing (Data Tests).
Prefer a composite key when the natural one is skewed and the composite is still meaningful — (country, day) instead of country spreads a dominant country over many partitions for free.
Where one side of a skewed join is small, broadcast it. The shuffle disappears and with it the skew, which makes this the first remedy to consider rather than the last (Broadcast Joins).
Salt the hot key when nothing else applies, and salt only the hot key (Salting a Skewed Key).
Monitor the shape, not just the runtime. A per-key distribution check catches the migration that turned a foreign key into a constant before it becomes an incident (Volume Anomalies).

What this actually promises

Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.

The shuffle guarantees that every row of a key is co-located. It guarantees nothing at all about how many rows that is, and it never has.
Skew does not compromise correctness. In the in-repo model, a run with the skew fault produces reported revenue exactly equal to the true revenue: every row is present, unique and correctly typed (Reconciliation).
What skew does compromise is the freshness promise. A job whose duration is set by one task cannot commit to a completion time based on average throughput (Pipeline SLOs).
No engine guarantees an even distribution. Adaptive execution can split an oversized partition after observing the shuffle statistics, which mitigates the symptom for some operations and is not a guarantee (Query Optimizers).

Can I trust it?

A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.

The check that would catch this

Run a distribution check: compare each key's share of today's rows with its historical share, and alert on a material shift. It catches a nullable key becoming dominant, a migration introducing a default, and a tenant onboarding that changed the shape of the data.
It has a specific and instructive blind spot, visible in the in-repo model: the check compares today's mix with the platform's historical mix, so it fires on skew even when nothing was lost, duplicated or mistyped. It is a *shape* alarm, not a correctness alarm, and treating it as the latter trains people to ignore it (Quality Alerting).
It also misses skew that has always been there. A key that has held 80% of the rows since the beginning looks perfectly normal to a check that compares against history.

Freshness

Skew turns the job's duration from a function of total work into a function of the largest key's share. A pipeline that met its window at an even distribution misses it at a skewed one with the same total volume.
Because the effect is proportional to the dominant key's share rather than to data growth, it can appear as a step change rather than a trend — which defeats capacity planning built on averages (Capacity Planning: Traffic to Machines).
The remedy that preserves freshness best is the one that removes the shuffle, not the one that redistributes it more cleverly (Broadcast Joins).

When the schema or meaning changes

A key's distribution changes without any schema change. This is the domain's recurring theme in its physical form: the types are identical, the meaning is identical, and the physics changed (Semantic Changes).
A schema change that adds a nullable foreign key creates a future skew source the moment something starts failing to populate it.
Adding a partition or bucket column to a table changes which operations can avoid the shuffle entirely, which changes where skew can hurt (Bucketing).

How to re-run this safely

There is nothing to repair. A skewed run produces correct output slowly, so recovery means re-running with a better key, a broadcast, or salting — not repairing data (Salting a Skewed Key).
When a skewed job has already missed its window, the fastest recovery is usually to split the work: run the dominant key as its own job and the remainder as another, then union the results (Reprocessing vs Retrying).
If the skew came from a backfill concentrating a year into one partition, the fix is to bound the backfill by range and run it as several jobs (Planning a Backfill).

What can go wrong

Failure modes

One task running for many multiples of the median while the rest of the cluster idles.
That task exceeding its memory budget and either spilling heavily or failing outright, taking the stage down with it after several retries.
A distribution alert firing on a legitimate business change — a genuine surge in one country — and being dismissed as noise, which is how the same alert gets ignored during the incident that matters.
Adding capacity as the response, which raises the bill and leaves the duration where it was (Compute Waste).
The mitigation failing: salting applied uniformly to every key, which multiplies partitions and the mean by the same factor and moves the ratio not at all (Salting a Skewed Key).

Misreads

"The job is slow, add workers." The longest task is indivisible. Extra workers idle alongside the existing ones (Straggler Tasks).
"Increase the partition count." Cutting an uneven distribution into more pieces leaves the largest piece the same size relative to the mean. It helps only when partitions were too large across the board.
"Skew means the data is wrong." Skew is usually a faithful reflection of a real business. The wrongness case — nulls and sentinel defaults dominating a key — is a different problem wearing the same symptom.
"The distribution check caught a data quality issue." It caught a change in shape. Whether that is a quality issue or a good quarter is a question only a human with context can answer (Quality Alerting).

Operating it

How you see it in production

Maximum versus median shuffle-read bytes per task in the stage. This is the direct measurement, and it separates data skew from a slow machine in one glance (Tail Latency: Why p50 Being Fine Does Not Help).
Top keys by row count on the shuffle key, computed on the source. It names the culprit rather than describing the symptom (Hot Keys: When Aggregate Metrics Hide a Saturated Node).
Per-key share over time, so the day the distribution changed is visible rather than inferred (Volume Anomalies).
Stage duration against total rows processed. When the two stop tracking each other, the distribution is what changed (Pipeline Metrics).

What changes at 10x and 100x

At 10x volume with the same distribution, the skewed partition grows 10x while the mean does too — the ratio is unchanged and the absolute pain is worse, because the largest task now exceeds memory where before it merely exceeded patience.
At 100x, skew usually decides whether the job is possible at all, and the remedies stop being optimisations and become requirements.
Higher cardinality generally helps: more distinct keys means the dominant key holds a smaller share, unless the dominant key is a sentinel that grows with the data.

What drives cost here

The dominant cost is idle capacity: every executor waiting for one task is billed at full rate for doing nothing (Compute Waste).
Spill on the oversized task adds local disk I/O that the balanced case never pays (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
Retries multiply everything. A skewed task that fails on memory after twenty minutes and is retried three times costs an hour before it fails the stage.

What this approach costs

Every remedy costs something. A composite key changes the grain of the output. Broadcasting adds a size assumption. Salting adds a stage and worsens the output layout. There is no free fix (Salting a Skewed Key).
Monitoring the distribution creates an alert that fires on legitimate business change, and tuning it to fire less means it fires later.
Splitting the hot key into its own job doubles the pipeline's surface area — two runs, two commits, one union — in exchange for a predictable duration.

Skew lab — partitions, workers, and the straggler

Change an input and watch which number moves — and which one does not. Everything here comes from a model in this repository, not from a measurement.

Skew lab — partitions, workers, and the straggler

One country grows to hold most of the day. Drag the worker count and watch what does not happen.

Workers8

Partitions

8sim

Largest / mean

6.31xsim

Rows on the busiest worker

3,157sim

Parallel efficiency

16%sim

Rows per partition

US3,157 · 78.9%

DE210 · 5.3%

GB183 · 4.6%

FR110 · 2.8%

IT109 · 2.7%

PL109 · 2.7%

ES89 · 2.2%

NL33 · 0.8%

Rows each worker must process

worker 13,157

worker 2210

worker 3183

worker 4110

worker 5109

worker 6109

worker 789

worker 833

The job finishes when the worker holding 3,157sim rows finishes. Every other worker is waiting. Adding another worker gives the new worker a small partition and leaves the largest one exactly where it was.

One partition holds 79% of the day. The job finishes when that task does, and adding workers changes nothing.

SIMULATEDRows per partition come from the same pipeline model the flagship lab uses. Work is measured in rows, never in seconds — a latency claim here would contradict the domain that owns latency.

Where this applies

Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.

GENERALAny system that partitions by a hash of a key inherits the key's distribution — distributed engines, sharded databases, and partitioned logs alike. What differs is only whether the system can rebalance at runtime or whether you must pick a better key.
SIMULATEDThe distributions and the straggler ratios quoted in this lesson come from the model in src/de/sim/pipeline.ts, whose skewed mix places 80% of orders in one of eight countries. They are arithmetic on a declared model, not measurements from a cluster, and a real dataset will differ in every magnitude while behaving the same way.
ENGINE-SPECIFICAdaptive execution in recent Spark versions can detect an oversized shuffle partition from runtime statistics and split it for some join types; Flink rebalances differently and a distributed warehouse may hide the problem behind its own scheduler. None of them removes skew from a plain aggregation on a dominant key.

Where the depth lives

This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.

Databaseaggregation-and-grouping partitioning-and-sharding join-algorithms-internals

Architectureconsistent-hashing load-balancing

Observabilityhot-keys tail-latency saturation capacity-planning disk-io-performance

Concurrencyamdahls-law work-and-span parallel-scaling-is-not-linear fairness

Distributed Systemspartial-failure

DSAhash-table

Domains that do not exist yet

— Distributed Systems owns why a hash partitioner cannot balance rows without knowing the distribution, and what rebalancing costs a system that has already placed state by key.