Hot Keys, Skew, and Load Imbalance in Hash Partitioning

A hash partitioner distributes keys, not work. When keys are numerous and roughly uniform, those two things coincide closely enough that the difference is invisible. When a small number of keys carries a large share of the records, the partition holding them becomes a serial bottleneck inside a parallel job, and the stage finishes when that one task finishes rather than when the cluster’s capacity is spent.

Balanced by key is not balanced by work

The partitioner’s obligation is that equal keys meet. That obligation cannot be relaxed to spread load, because splitting a key’s records across two tasks breaks every reduce-side operation that assumes a complete group. The only freedom available is which keys share a partition.

Under a well-mixed hash and uniform keys, the record count per partition has a relative standard deviation of about $1/\sqrt{n/R}$, so with a million records in a partition the deviation is a fraction of a percent. Skew changes the picture entirely. If one key accounts for 30 percent of a trillion records and the job runs 200 partitions, the partition containing that key holds 30 percent of the data while the average partition holds half a percent — roughly sixty times the mean. The other 199 tasks finish and their slots sit idle while one task reads, merges, and aggregates sixty times its share. The wall-clock cost is not the total work but that maximum, so a job with a skewed key ends up using about $1/60$ of the parallelism it was scheduled with.

Two other shapes of skew are easy to conflate with this one. Byte skew occurs when a key has few records but very large values, so the partition is expensive in bytes moved rather than in record count. Per-record cost skew occurs when one key’s values trigger expensive work — a nested structure to expand, or a join probe that misses every time. A partitioner that balances record counts does nothing for either.

Two-stage aggregation with salted keys

The standard static remedy for key skew in an aggregation is to make the hot key pass through several partitions in a first stage and converge in a second. The map stage emits the key with an additional salt drawn from a small fixed range, $s \in {0, \dots S-1}$, typically chosen by a hash of something incidental to the record so the choice is stable per record rather than random. The first stage aggregates over the composite $(k, s)$, so the hot key now contributes $S$ groups spread across $S$ partitions. The second stage aggregates those $S$ partial results per key down to one.

The salt bounds the damage. Before salting, one partition held 30 percent of the records; after salting with $S = 64$, the hot key’s records spread over 64 partitions, and the worst partition holds about 0.5 percent of the hot key’s share plus its ordinary load, a few times the mean rather than sixty times it. The total shuffle volume is largely unchanged for the hot key’s own records — they are still written and fetched once each — but the first stage’s output adds at most $K \times S$ rows, where $K$ is the distinct key count, because each group collapses to one row before the second stage.

Choosing $S$ is a balance. Too small and the salt’s partitions stay hot; too large and the salted intermediate grows toward $K \times S$ rows regardless of how much data comes in, which can exceed the input for low-selectivity aggregations. Salting also adds a stage boundary, and therefore one more barrier in the sense described in the shuffle data path, so it is applied to the stage that actually needs it rather than to every stage.

The technique is limited to operations that decompose. Sums, counts, maxima, and set unions all survive being computed in pieces, which is exactly the associativity precondition that combiners rely on. A join cannot be salted on one side alone: if a hot key’s rows are split across 64 salted partitions, the other side must be replicated to all 64 so that each partition can match locally, multiplying the smaller side’s shuffle volume by 64. Salting therefore trades skew for duplication, and the trade is only favourable when the duplicated side is genuinely small.

Sampled partitioners that target equal volume

A hash partitioner is blind to how much data each key carries. A custom partitioner can instead be built from a sample: collect approximate per-key record counts, sort keys by count, and cut the key space into $R$ intervals with equal cumulative volume rather than equal key counts. A key with a thousand times the median volume can be given its own partition or share one with very few others, which removes the sixty-times-the-mean problem at its source.

The cost is the sample itself. An exact count is a full aggregation over the dataset, which is the work the job was going to do anyway, so it does not pay for itself in a single run. Sampling a small share of the input splits gives estimates accurate to a few percent for the heavy tail but can miss a heavy key entirely if it appears in few splits, so the sample size has to be justified rather than assumed. Persisted statistics — a histogram maintained alongside the table, or the boundary points stored with a bucketed table — avoid the resample cost and are the reason persistent partitioning is attractive for repeated jobs, as co-partitioned joins also depend on.

Whatever the boundaries, the resulting partitioner must remain a deterministic function of the key. A partitioner whose assignment depends on ambient state, on wall-clock time, or on the order in which records arrive will place equal keys in different partitions on a retried map attempt, producing the failure described in shuffle fault tolerance.

Splitting a partition at run time

Static fixes have to be chosen before the data is read. A runtime fix instead observes the sizes of completed shuffle outputs and splits an oversized partition across several reduce tasks. Each of those tasks reads a subset of the map-output segments for that partition; the union of their inputs is exactly the original partition, so no record is lost and none is processed twice.

The benefit is real but bounded by what the reduce function can do with an incomplete group. A split partition works naturally for a commutative, associative aggregation, because partial results from the sub-tasks can be merged; it does not work for operations that need to see the whole group at once, such as the outer side of a sorted merge join or a top-$k$ that is not re-mergeable. Split tasks also pay the reducer’s fixed costs — building a merge heap, opening fetched segments, warm-up — once per split, so splitting a partition into a hundred fragments is worse than splitting it into eight. Frameworks that do this expose it alongside other decisions made at the stage boundary, as described in adaptive partitioning at runtime.

A related but distinct problem is the straggler caused by hardware rather than data: one attempt is slow because its machine is, not because its partition is large. That is handled by speculative execution in the scheduler — running a duplicate attempt and taking whichever finishes first — and it does nothing for skew, because a duplicate of an oversized partition is also oversized. Mistaking one for the other leads to duplicate work with no improvement.