The previous lesson treated a shuffle as if its partitions were the same size. They never are, and the difference between the shape a shuffle is supposed to have and the shape it actually has is the single most common reason a distributed job fails for no visible reason.

A definition that is actually useful

Skew is not a property of a dataset. It is a property of a dataset and a partition key together.

The same table is perfectly even when partitioned by a surrogate identifier and catastrophically uneven when partitioned by country. Nothing about the rows changed. What changed is which column the partition function was asked to hash, and therefore which real world distribution the engine inherited.

This matters because it tells you where to look. People hunt for skew in the data, and the data is usually fine. The skew lives in the choice of key, which is a line in a query rather than a property of a table, and which a planner may have chosen for you.

So the useful definition is this. A shuffle is skewed when the largest partition is substantially larger than the average partition. Substantially means enough that the task reading the largest partition takes materially longer than the rest of the stage, which in practice means a factor of about three.

Why the average is a lie and the maximum is the truth

A stage of a thousand tasks finishes when its slowest task finishes. This is the fact that makes skew expensive and it is worth stating in its starkest form.

Suppose a shuffle moves a terabyte into a thousand partitions. The average partition is a gigabyte. If nine hundred and ninety nine partitions are nine hundred megabytes and one is a hundred gigabytes, the average is still almost exactly a gigabyte and the stage takes as long as the hundred gigabyte task.

Every dashboard that reports average partition size, average task duration or total bytes shuffled will describe this stage as healthy. Every one of them is averaging over the thing that does not matter.

The measurement that matters is the ratio of the maximum to the median. Not the mean, because the mean is dragged upward by the outlier you are trying to detect, which is exactly backwards. The median tells you what a normal partition looks like, and the maximum tells you what the stage will actually cost.

A ratio under two is fine. A ratio between two and five is worth knowing about. A ratio over ten is why the job failed.

What the partition function can and cannot do

The default partition function is a hash of the key modulo the partition count. It has one guarantee and one non guarantee, and conflating them causes most of the confusion in this area.

The guarantee is that identical keys land in the same partition. This is the whole point. A group by cannot work unless every row with a given key reaches the same place, and a join cannot work unless both sides agree on where a key goes.

The non guarantee is anything at all about size. The function distributes keys evenly, which is a completely different statement from distributing rows evenly. If one key has ten million rows and the others have ten each, the hash will place that key somewhere, and wherever it lands becomes the largest partition in the shuffle.

This is why a better hash function does not help, and why people occasionally waste a week on one. The hash is already excellent at spreading keys. The problem is that the rows are not spread across the keys.

The three shapes skew takes

In practice skew arrives in three distinguishable forms, and they need different fixes.

The first is a single hot key. One value has orders of magnitude more rows than any other. Null is the archetype, because a nullable join column puts every unmatched row into one partition, and nulls are frequently a tenth of a table. Sentinel values do the same: a default identifier of zero, an unknown country code, a placeholder customer.

The second is a small set of heavy keys. The top ten values hold half the rows. This is what any real popularity distribution looks like, and it is the normal shape for users, products, merchants and cities. There is no bug, the world is simply like this.

The third is a structural imbalance in the partition count. The key is fine and the distribution is fine, but the partition count shares a factor with something in the key space, so hashing maps many keys onto a few partitions. This is rare and usually self inflicted, by a custom partitioner or by a key that is already a hash.

What skew costs, in the three places it costs something

The first and most obvious cost is duration. The stage lasts as long as the largest task, so a ten to one ratio turns a five minute stage into a fifty minute one, and the cluster is idle for forty five of those minutes.

The second cost is failure. A task that must hold a hundred gigabyte partition in memory to join or aggregate it will not succeed, regardless of duration. It will spill, then spill more, then run out of disk or heap and fail. The stage retries, the same partition goes to the same place, and it fails again. Four times and the job is dead. This is the cost that produces a pager alert rather than a slow dashboard.

The third cost is the one nobody attributes correctly: everything downstream inherits the shape. A skewed shuffle produces a skewed output, so the files it writes are uneven, and the next job to read those files inherits task sizes it did not choose. Skew propagates through a pipeline the way a bad schema does, quietly, until someone three stages away is debugging a symptom with no local cause.

Measuring it before it bites

Every engine computes the information you need and none of them put it anywhere obvious. The numbers to take for each shuffle are small and cheap.

Take the maximum partition size and the median partition size, and record their ratio. This is the one number that predicts everything in this lesson.

Take the maximum task duration and the median task duration for the consuming stage. The ratio should be close to the size ratio, and if it is much larger then the problem is not size but something that scales worse than linearly, such as a join that degraded into a nested loop over one partition.

Take the spilled bytes for the largest task specifically rather than for the stage. A stage total hides the fact that one task did all of it.

Those three, per shuffle per run, turn skew from a thing that is diagnosed after a failure into a thing that is visible weeks before one.

What this lesson is going to do

The rest of this lesson works through the fixes in the order you should try them, which is not the order they are usually taught.

First, remove the keys that should not be there, because a filter is cheaper than every other technique and nulls and sentinels account for most real cases. Second, let the engine handle it, because adaptive execution now splits large partitions automatically and a surprising number of skew problems were solved three releases ago by somebody else. Third, change the aggregation so the heavy keys are reduced before they are gathered, which costs one extra shuffle stage and is almost always worth it. Fourth, salt, which is the technique everybody reaches for first and which should be last, because it is intrusive, it changes the query, and it has to be undone afterwards.

That ordering is the real content of this lesson. The techniques themselves are mechanical once you know which one the situation calls for.