System designCore11 min

Hot partitions & skew

Sharding spreads data across machines on the assumption that load spreads with it. Real load doesn't: one celebrity, one product or one timestamp can send most of the traffic to one shard while the others sit idle. This page shows where skew comes from and how to break a hot key apart.

1 · The problem

More shards don't help when the load is on one key

A TV talent show's final has four contestants and a 10-minute voting window. At its peak, viewers cast 120,000 votes a second. Each vote adds one to its contestant's counter in a database spread over 16 shards, keyed by contestant. Each shard can take about 10,000 writes a second, so on paper the cluster can take 160,000.

But the votes aren't even. Contestant A gets 60% of them, and every vote for A updates the same key, which lives on one shard:

Shard 3 (contestant A)
72,000 writes/s
Shard 9 (contestant B)
30,000 writes/s
Shard 12 (contestant C)
12,000 writes/s
Shard 5 (contestant D)
6,000 writes/s
Each of the other 12 shards
0 writes/s

Votes split 60%, 25%, 10% and 5% of 120,000 a second. Three shards are past their 10,000 limit and 12 do nothing. The cluster average is 120,000 ÷ 16 = 7,500 a second, which looks healthy.

This is a hot partition: one shard getting far more than its share. The general condition behind it is skew, load or data spread unevenly across keys. Two lessons are already visible:

  • Adding shards changes nothing. With 32 shards, contestant A's key still lives on one of them.
  • Averages hide it. The cluster is asked for 75% of its capacity, while A's shard receives over 7 times what it can handle. Only per-shard numbers show the problem.

2 · The causes

Skew comes from popular keys, too few keys, or keys that arrive together

The voting app has all the common causes at once:

Cause

In the voting app

Result

A hot key

Contestant A gets 60% of the votes

One shard takes 72,000 writes a second

Too few keys

Only 4 contestants, so only 4 counter keys

At most 4 of 16 shards can ever be busy

Keys that arrive together

The vote log is keyed by timestamp

Every new vote lands in the same, newest range

A key that holds too much

The vote log keyed by contestant

One partition stores 60% of the rows

The third cause catches teams that did everything else right. Suppose the vote log is stored in a database that splits keys into ranges (Sharding & partitioning), and each row's key starts with the time of the vote. Time only moves forward, so every insert falls at the end of the key space:

New voteskeys 18:04:31.002, …Range up to 18:00no writes18:00 to 18:04no writes18:04 onwardevery insert0/s0/s120,000/sNew voteskeys 18:04:31.002, …Range up to 18:00no writes18:00 to 18:04no writes18:04 onwardevery insert0/s0/s120,000/s

With timestamps as the leading key, the newest range takes every insert. When it is split, the new upper half becomes the hot one.

Any key that grows steadily does this: timestamps, auto-increment IDs, sequence numbers. Splitting the range doesn't help, because the new upper half immediately becomes the hot spot.

3 · The trap

Hashing spreads keys, not requests

The usual advice against hot spots is to hash the key, and for the timestamp log it works. Put a hash in front of the time, or lead with the voter's ID, and inserts scatter over every shard. The price is range scans: "all votes between 18:04 and 18:05" now has to ask every shard.

For contestant A, hashing does nothing. A hash sends each key to one shard, deterministically; that is the whole point of it. Every vote for A carries the same key, so every one of them hashes to the same shard. Consistent hashing and virtual nodes (Consistent hashing) balance the number of keys per shard, not the traffic per key:

Cause

Does hashing the key fix it?

Keys that arrive together (timestamps, sequences)

Yes: consecutive keys scatter across shards

Too few keys

No: 4 keys can occupy at most 4 shards

One hot key

No: one key always hashes to one shard

One key that holds too much data

No: the key's data stays in one place

Finding a hot partition needs the right numbers:

  • Load per partition, not per cluster: requests, CPU and latency for each shard.
  • Throttling or errors per partition: a managed database often reports when one partition hits its limit.
  • The busiest keys: many databases and caches can sample the top keys by traffic. One key at 60% of writes is the whole diagnosis.

4 · Write fix

Split a hot key into sub-keys and add them up on read

If one key is too hot for one shard, make it several keys. Instead of one counter votes:A, keep 16: votes:A#0 to votes:A#15. Each vote picks one at random and adds 1 to it. The 16 sub-keys hash to different shards, so A's 72,000 writes a second spread out. The total is the sum of the 16. B, at 30,000 a second, and C, at 12,000, are past the limit too, so they get sub-keys of their own, 8 and 4; D stays a single key.

How many sub-keys? At least the hot key's load divided by what one shard can take, plus headroom, because hashing 16 sub-keys onto 16 shards won't place them perfectly evenly:

Step

Contestant A

Load on the key

72,000 writes/s

One shard's limit

10,000 writes/s

Minimum sub-keys

72,000 ÷ 10,000 = 7.2, so 8

Chosen, with headroom

16 sub-keys, about 4,500 writes/s each

Writing and reading a split counter

javascript
const N = 16;
// write: one random sub-key per vote
await db.increment(`votes:A#${randomInt(N)}`, 1);

// read: fetch all N and add them up
const keys = [...Array(N).keys()].map(i => `votes:A#${i}`);
const total = (await db.getMany(keys)).reduce((s, v) => s + v, 0);

The cost moves to the reader: one read of the total became 16. So split only the keys that are actually hot, not every key in the table. That needs some bookkeeping, a list of which keys are split and into how many, which readers must consult.

Two variants are worth knowing. A random suffix, as above, spreads writes best but means nobody knows which sub-key holds a given item, which is fine for counters and bad for records you need to look up. A calculated suffix, such as a hash of the voter's ID modulo 16, spreads writes almost as well and lets you find one voter's row directly.

Splitting works for counters and for logs, where order between sub-keys doesn't matter. It doesn't work for a value that needs one serialized history, such as an account balance with an overdraft rule.

5 · Read fix

Copy a read-hot key instead of splitting it

The results page has the opposite problem. Two million viewers keep it open, and the app refreshes it every 5 seconds: 2,000,000 ÷ 5 = 400,000 reads a second of the same totals. Splitting the counters made this worse, since one read of the four totals now needs 29 sub-keys (16 + 8 + 4 + 1).

Reads of the same data don't conflict, so a read-hot key isn't split; it is copied closer to the readers. One small job sums the sub-counters once a second and writes the result. Every app server caches that result for one second:

2 million viewers400,000 reads/s50 app serverseach caches 1 secondResults rowupdated every second400,000/s50/s2 million viewers400,000 reads/s50 app serverseach caches 1 secondResults rowupdated every second400,000/s50/s

Each server asks the database once a second, whatever its traffic: 50 reads a second instead of 400,000. Viewers see totals up to about 2 seconds old.

The same fix scales further. A CDN can cache the results for every viewer in a city (Content delivery networks (CDNs)), and read replicas multiply the copies inside the database (Database replication). The one thing to guard is the moment a hot cached item expires, when every server misses at once (Cache failure modes).

Hot for

Conflicts?

Fix

Writes

Yes: each write changes the value

Split into sub-keys, combine on read

Reads

No: readers can share a copy

Cache and replicate; refresh on a schedule

6 · The limits

Databases rebalance hot partitions, but no database can split one key

Managed databases try to absorb skew on their own, and it helps. It stops at the same place every time, because the unit they can move is a key:

System

What it does on its own

Where it stops

DynamoDB

Each partition serves up to 3,000 reads and 1,000 writes a second (small items); adaptive capacity moves busy items apart

At best one partition holds only the hot item, which still gets 1,000 writes a second

Range-split databases, such as Spanner

Split ranges that grow or get busy

A split can't divide one key, and a timestamp key moves the heat to the new range

Kafka

Spreads keys over partitions

One key's messages stay in one partition, read by one consumer, to keep their order

Stream jobs

Spread keys over instances

One key's state lives on one instance (Stream processing)

So in DynamoDB, contestant A's single counter tops out at about 1,000 writes a second no matter how the table is provisioned. The 72,000-a-second key needs sub-keys there too. The pattern repeats in every system: automatic balancing spreads many warm keys well, and only a change to the keys themselves fixes one hot key.

7 · In practice

Plan for the celebrity before the show starts

Hot keys are rarely a surprise. The show knows who the finalists are, a shop knows which product is on sale, a social app knows who has millions of followers (Fan-out on write vs read). Designing for them in advance is far easier than splitting a key under load.

Hot key

Example

Prepare by

Read-hot

A finalist's profile page

Caching and replicating it, refreshed on a schedule

Write-hot

Votes for one finalist

Splitting the key in advance

Runaway

A key nobody expected to be hot

A per-key rate limit, and per-partition monitoring

When designing:

  • Choose keys with many values that are used evenly, and avoid a timestamp or sequence as the leading part of a key.
  • Expect a few keys to dominate. Real traffic is skewed, and a load test that spreads requests evenly over all keys tests a workload you'll never see.
  • Split known write-hot keys in advance, and only those, with the list of split keys kept where readers can find it.

When running:

  • Cache and replicate read-hot keys, and refresh them on a schedule instead of on every request.
  • Watch per-partition load and the top keys, not cluster averages.
  • Keep a last line of defense: a per-key rate limit stops one runaway key from taking its shard, and everything on it, down (Rate limiting algorithms).

Check yourself

5 questions

1. Votes go to 4 contestant keys on a 16-shard cluster, and contestant A's shard is overloaded. What happens if the cluster grows to 32 shards?
2. Which hot spot does hashing the key fix?
3. A key takes 45,000 writes a second, and one shard handles 10,000. What is the minimum number of sub-keys, before headroom?
4. After splitting A's counter into 16 sub-keys, what does reading A's total cost?
5. 400,000 reads a second ask for the same results row. What is the usual fix?

Takeaways

Remember this

  • Skew sends most of the load to a few keys, and a key always lives on one partition, so adding shards doesn't help.
  • Cluster averages hide hot partitions; watch per-partition load and the busiest keys.
  • Hashing fixes keys that arrive together, such as timestamps, but not one hot key or too few keys.
  • Split a write-hot key into sub-keys and sum them on read; split only the keys that are hot.
  • Copy read-hot keys with caches and replicas; managed databases rebalance partitions but can't split a single key.