The autumn tax of Riverbend depended on one sorting room and ten counting houses.

The rule was not complicated. Salt and grain merchants bought from farmers every year, and the moment goods left a hand, a receipt was written with exactly three things on it: household number, quantity, year. After the harvest, every one of those receipts had to be stamped with the householder’s name, the village, and the tax owed — none of which appeared on the receipt. They lived in another register called the household roll. Three million households, one sheet each.

Receipt against roll, matched on household number, stamped, and only then was the tax settled.

Turning pages one at a time is the stupidest method available: three million roll sheets against ten million receipts is more than one lifetime of page-turning. So the town built ten counting houses, each responsible for a slice.

The slicing was clever: look at the last digit of the household number. A trailing zero went to East One, a trailing one to East Two, and a trailing nine to West Five. Household numbers had been issued haphazardly over the years, so the trailing digit was naturally even — each house received three hundred thousand roll sheets, no more, no less, not one page off.

Receipts were divided by the same digit. And in past years it truly was even: a household generates a few dozen receipts a year, a pile of receipts flowed into ten rooms, and every room held a thin stack.

All ten rooms started together and finished together. Whichever room stamped its last seal first still had to wait for the slowest room to close its ledger. The autumn tax was always settled in seven days.

Then came the year of the Chen Compound.

The Chen family was not a household. On paper it was one household — a single number, because three hundred years ago nobody had bothered to split the family registry, and three centuries of land sat under that one number. But it did not produce a few dozen receipts. It produced more than four million. Half the town’s grain and forty percent of its salt moved under that single number.

The sorting room followed the old rule and went by the trailing digit. The Chen number ended in seven. All four million receipts, not one misplaced, went into West Two.

That night West Two lit forty lamps.

The other nine rooms cleared their thin stacks on the first night, and from the second morning on they sat in the corridor drinking tea, waiting for West Two.

They waited eleven days.

Sitting in West Two’s doorway that night, the old accountant understood something. Evenness was never about how many sheets of paper there were. It was about whose names were written on them. The trailing digit guaranteed that households were even. It guaranteed nothing about receipts. And when ten rooms could close had nothing to do with how much the ten rooms did together — only with when the slowest one stopped.

His first thought was more hands. West Two went from four people to thirty. For one day it was genuinely three times faster. By the third day the room was too crowded to turn around in: there was one set of ledgers, one person who could turn its pages, one table with room for one seal. Thirty pairs of hands were all waiting on that one register.

You cannot add hands. So change how the work is divided.

The next autumn he added a room that did nothing but weigh.

Receipts came in and went first to the sorting room: thrown into ten wooden bins by trailing digit. Then, before loading the carts, weigh each bin.

The scale spoke, and the truth was plain. Bin seven: four thousand catties. The other nine: eighty each.

This was the step the old accountant had never taken in forty years — weigh before you divide. Before, he divided, got it wrong, finished dividing, and only then learned it had gone wrong. By then the tax was eleven days in arrears.

The four-thousand-catty bin had to be split.

But it could not be split. A receipt carries one household number, the number ends in seven, and no matter how many times you throw it, it lands in bin seven. To separate those receipts you must add something to the paper that was never there.

On the back of every Chen receipt the old accountant wrote one small character in vermilion: one, two, three, four, five, six, seven, eight — cycling, then cycling again. With that character, the Chen household acquired eight identities: Chen-seven-one, Chen-seven-two, and so on through Chen-seven-eight.

The roll side followed suit: the single Chen sheet was copied eight times, and each copy carried one of the same characters on its back.

So the four thousand catties in bin seven became eight parts of five hundred, landing in eight different counting houses. Eight rooms started at once; eight “Chen-seven-something” entries each matched against their own receipts.

A day and a half later the eight rooms had each closed their portion and produced a subtotal, which went to one small reconciliation table. One addition, and the total existed.

The old accountant stood at that table for a long time. Forty years he had tried to divide the work evenly. It had never occurred to him that you could take one household and, for a while, pretend it was eight.

In later years, that number became a figure he had to revise annually. In the third year the salt price doubled and the Chen receipts reached eight million; eight characters were not enough, so he used sixteen. A few years on he wrote twenty-four. Every autumn, before the harvest, the sorting room’s scale spoke and told him how many characters this year required. More characters meant more even rooms; more characters also meant more subtotals on the reconciliation table — eight becoming sixteen, sixteen becoming twenty-four.

One year the houses got greedy and wrote sixty-four characters. The sorting room collapsed first: one bin had to be carved into sixty-four parcels, and every parcel had to be tagged, weighed, loaded, and carted. The packing took longer than the stamping. That year the forty lamps burned in the sorting room.

The hardest year was the fifth. Three new trading houses opened beside the Chens, each taking fifteen percent.

The sorting room weighed: bin seven was still four thousand catties. Opened up, it was not one lump but three — thirteen hundred each — plus a dozen small lumps of a few dozen catties apiece.

The old accountant applied the old method and gave each lump eight characters. Three hot households, eight subtotals each, twenty-four subtotals, all crowded onto one reconciliation table.

The reconciliation table became the new West Two.

And he finally understood the last layer: “hot” is not a fixed roster. It differs every year. Sometimes one household, sometimes three. Sometimes one lump as heavy as a mountain, sometimes a dozen lumps at ten percent apiece. Fix the roster in advance, fix the number of characters in advance, and one year it will not match — and that year’s accounts rot in the sorting room.

That winter he rewrote the rule as three steps, and kept to them thereafter.

First: sort, then weigh. The bins were still divided by trailing digit; the only addition was measurement. This step changes no division.

Second: carve only what is heavy. A bin more than twice the average share gets the knife. And the carving stays inside the bin — the heavy household is split into parts, while the light households are untouched, not one character written.

Third: reconcile. Every split household gathers its own subtotals onto one table; households that were never split have exactly one part and enter the ledger directly, never touching the table. From then on the table’s size tracked the number of hot households, not the number of parts.

In the sixth autumn: three million roll sheets, eight million receipts. In the ten bins, the heavy ones scattered into thirty parts and the light ones stayed as they were. The ten rooms closed their ledgers one after another in a single evening, and that year’s autumn tax was settled on the fifth day.

The old accountant sat in the corridor with the scale still hanging from the sorting-room beam. He remembered his master’s words from forty years before: whether work divides evenly is not about how many sheets there are — it is about whose names are on them. His master had not gone on. The next sentence was his own: if the name on the paper is too concentrated, then for that paper, give it a few more names.

By now you have probably recognized it: this is data skew and hot-key salting.

What it is

A distributed join rests on a precondition: both sides must be shuffled into the same set of partitions by the join key. Rows sharing a key must land in the same partition, or they cannot meet. The partitioning function is usually hash(key) % N.

That precondition assumes one thing: keys are evenly distributed. When they are not, hash partition crushes a mass of same-key data into a single partition — data skew. Common sources: one enormous customer, a catch-all “other” value, and a whole batch of NULL keys (which all hash to the same side).

Skew’s damage is not in the total data volume but in max task time: a stage finishes when its slowest task finishes. If 31 of 32 executors take 30 seconds and one takes 30 minutes, the stage takes 30 minutes — effective parallelism collapses to single digits.

Hot-key salting is three moves: append a random suffix to the hot key’s value (key#0 … key#S-1); copy the matching few rows of the small side S times, each carrying one suffix; and the hot key’s N rows now spread across S partitions to be processed in parallel. Each partition computes a partial result, and a small two-stage aggregation adds the S subtotals back together.

The essential part is doing it only for hot keys. Salt every key and you multiply every key’s work by S, the partition count by S, and the scheduling overhead by S.

Why it matters

Sizing resources by average data volume is the most common self-deception in distributed systems. Skew tears “average” away from “slowest”: you buy machines by the mean and wait for the maximum.

Three things matter in practice:

• Measure first. Partitioning without statistics is gambling blind on a distribution. Modern engines move this into the runtime: Spark’s AQE collects map statistics per partition after the shuffle write and then decides which partitions to split — a re-optimization between two stages, treated as normal.
• Split only the heavy ones. Skew lives in a tiny number of keys. Salting everything makes 99% of keys pay for nothing. The threshold is usually “more than twice the average share” or an absolute byte limit.
• Reconciliation is a bottleneck too. The finer you split, the more concentrated the merge. Two-stage aggregation must scale with the number of hot keys, not with the number of parts.

Know the costs: the result set grows (the small side’s rows are replicated S times), an extra stage appears, and the hot keys must be identifiable on both sides. The better route is not to shuffle at all: if the small side fits in memory, a broadcast join (map-side join) replicates it to every task and nobody repartitions. But once the small side genuinely is not small, that road is closed and salting is all that is left.

Metaphor mapping

• One roll sheet per household → the small side (dimension table), the side that gets widened
• A few dozen receipts a year → the large side (fact table)
• Ten counting houses by trailing digit → hash(join_key) % N shuffle partitioning
• Naturally even trailing digits, three hundred thousand sheets each → the small side is naturally uniform; skew almost never comes from it
• The Chen Compound: one household number, four million receipts → a hot key: one key carrying a large slice of the actual data
• The digit guarantees even households, not even receipts → hashing evens the key space, not the data volume
• Ten rooms start and finish together, waiting on the slowest to close → stage time = max task time; parallelism eaten by the longest task
• West Two going from four people to thirty → vertical scaling (more executors/cores), which has a ceiling
• One set of ledgers, one page-turner, one sealing table → serial bottleneck inside a single task; more resources do not help
• Receipts entering the sorting room and its ten bins → shuffle write
• Weighing the bins: seven at four thousand catties, the rest at eighty → runtime per-partition byte sizes reveal the skewed partitions
• Carving only bins over twice the average → AQE skew thresholds (skewedPartitionFactor, skewedPartitionThresholdInBytes)
• Light bins untouched, not one vermilion character → skew join processes only over-threshold partitions; zero cost for the rest
• Writing one through eight on the receipt backs → appending a random suffix key#0 … key#S-1 to the hot key value
• Copying the Chen roll sheet eight times, each with one character → replicating the small side’s hot-key row S times, aligned to the suffixes
• Eight “Chen-seven-something” entries sealed in eight rooms at once → the hot key’s data split across S partitions in parallel
• Eight subtotals delivered to one reconciliation table → two-stage aggregation: partial results, then a final merge
• More characters mean even rooms and a more crowded table → the salt-degree S trade-off: diminishing balance gains, linearly rising replication and merge costs
• Sixty-four characters and the sorting room collapses → over-salting turns shuffle and scheduling overhead into the new bottleneck
• Three lumps of thirteen hundred, twenty-four subtotals on one table → with multiple hot keys, the merge side becomes a single point again
• Table size tracking the number of hot households → aggregation scales with hot-key count, not part count
• “Hot” is not a fixed roster → skew drifts with the data; a hardcoded salt table will fail in some year
• All the Chen land under one number → the real-world sources of hot keys: a single large customer, a default value, a catch-all, a NULL key