Sharding, Hot Keys & Rebalancing

Part 0. Start here

The problem: 66,667 reads and 10,000 likes a second on one post

A celebrity with 60 million followers posts a photo. The push notification goes out, and about 8 million followers open the app over the next 2 minutes. Every feed read pulls her newest posts with one Query on one DynamoDB partition key. The post's like counter is a single item, and it takes about 10,000 likes a second (our example's number). The table has plenty of capacity: on average it is about 20% busy (also our example's number).

The arithmetic, with DynamoDB's documented limits:

The numbers come from step 2.2 of the news feed loop, which met this exact spike.

The big picture

Synthesizing vector architecture diagram...

What to notice: the routers find each key's shard through a cached directory. One post, p9, lives on S2, so S2 alone is far over its ceiling while the other three sit at 20%. Everything on this page either spreads that load, absorbs it in the routers, or moves the piece.

What you'll be able to do after this page

1. Explain how a request finds its shard: key, slot, directory, and the shard's own slot table (Part 1).
2. Choose a partition key and a shard count, and tell key skew from load skew (Part 2).
3. Find a hot key, and explain why adding shards doesn't help it (Part 3).
4. Tame a read-hot key with a short cache, single-flight and coalescing routed by key (Part 4).
5. Tame a write-hot key with salting or pre-aggregation, and compare every hot-key fix on equal terms (Part 5).
6. Add capacity with hash mod N, a ring with virtual nodes or fixed slots, and say what each moves (Part 6).
7. Say what DynamoDB and Kinesis split for you, and why they never split one key (Part 7).
8. Move live data to a new shard without losing a write, and roll the move back (Part 8).
9. Run queries and transactions that span shards, and say what they cost (Part 9).
10. Trace one like and one read through every layer, and size the router fleet per AZ (Part 10).
11. Map all of it to AWS services, and name the look-alikes that are not sharding (Part 11).

You may have arrived from a step that relies on this: step 2.2 of the news feed loop (the celebrity post), step 2.5 of the ad-click aggregation loop ("split the hot Kinesis shard"), step 2.1 of the digital wallet loop (1,024 buckets on 16 shards) or step 3.2 of the Discord case study (request coalescing). This page is the "why" behind all four.

---

Part 1. What a shard is

One machine can't hold or serve all the data, so we split it. Then every request has a new question to answer before it can do anything: which piece holds my key, and which machine holds that piece?

Key, slot, shard

A shard (DynamoDB and Kafka say partition) is one piece of the data, served by one machine or one small replica group. We don't map keys straight to shards. We map them in two steps:

1. Key → slot. A fixed function turns the key into one of a fixed number of slots (also called buckets, hash slots or token ranges). Valkey and Redis Cluster use slot = CRC16(key) mod 16,384. The slot count never changes.
2. Slot → shard. A directory (a small, versioned table) says which shard owns each slot. This mapping does change when we add shards or move data.

A hash tag lets several keys share a slot on purpose. If a key contains {...}, only the text between the first { and the first } after it is hashed (if there is at least one character between them). So post:{p9}, likes:{p9}:r0 and comments:{p9} all hash p9 and land in the same slot, on the same shard, where one request can read them together.

Each shard is a leader plus replicas in other Availability Zones. How those copies stay in step is the Replication, Quorums & Read-Your-Writes loop primitive's subject; on this page a shard is one unit with one ceiling.

Synthesizing vector architecture diagram...

What to notice: only the tag p9 is hashed, so every key of post p9 lands in slot 12,067. The directory names the owner; the owner's own slot table confirms it before serving.

The router, the directory and the slot table

A router is whatever computes the slot and sends the request: a proxy tier, or a "smart" client library. Routers cache the directory, because asking the directory on every request would make it the bottleneck. A cached copy can be old. So correctness can't rest on the router's copy:

• The directory is the plan: slot range → shard, plus an epoch per slot that goes up every time the slot changes owner, and a version for the whole map.
• Each shard's slot table is the truth about that shard: for every slot, "owned at epoch e", "migrating", or "moved to S4 at epoch e". A shard checks every request against it.

textCopy
ROUTE(request for key k)
  1. tag   = text inside the first {...} of k, else all of k
  2. slot  = CRC16(tag) mod 16,384
  3. shard = cached_directory[slot]
  4. send (request, slot, epoch from the cached directory) to shard
  5. the shard checks its own slot table for slot:
       owned (request's epoch older: also tell the router to reload) -> serve
       moved to S', epoch e'      -> reply MOVED slot S' e'
       migrating                  -> reply "slot migrating, retry"
  6. on MOVED: reload the directory, go to 3
     on "retry": wait (10, 20, 40 ms ...), go to 3

That one check in step 5 is what makes moving data safe later (Part 8): a router with an old map is bounced, however old its map is.

Three words for one idea

The example we follow

A photo app keeps posts, their like counters and their comment threads in a sharded key-value store, the post store. Sixteen posts, p1 to p16, live on four shards, S0 to S3, behind six routers. At 20:01 the celebrity nova promotes her post p9. We follow that one post through every Part. The example's numbers are smaller than the hook's, so each shard's arithmetic fits on a line. Every trace on this page comes from running a private reference implementation of this store, not from working by hand.

Placement uses the real slot function. Shards own contiguous slot ranges: S0 owns 0 to 4,095, S1 4,096 to 8,191, S2 8,192 to 12,287, S3 12,288 to 16,383. From the reference implementation, the 16 posts land four per shard:

Per-like records live elsewhere. Each like also writes a record "user u liked post p" to a separate table, liked, keyed by the user (liked:{user_id}, one row per user and post), with its own shards. So a like on p9 never writes a per-event record to S2; the post store counts only post, counter and comment writes. A like is acknowledged only once it is durable in liked; the like-events stream is fed from liked's change stream (Change Streams & the Transactional Outbox), so every durable like has its event (Part 5 shows why).

Snapshot S1, at 20:00:00, from the reference implementation:

The story has ten beats, each in its own Part: the key is chosen (Part 2), p9 goes viral (3), the reads are tamed (4), the writes are tamed (5), the cluster grows, on a copy (6), the managed services, on a copy (7), p9 moves to its own shard while it takes writes (8), a query and a transaction cross shards (9), one like end to end (10), and on AWS (11). The full event table, including side rows left out of the body, is in Part 12.

---

Part 2. Choosing a partition key

The key decides which queries touch one shard and which touch all of them, and whether the load spreads. Get it wrong and no amount of hardware fixes it later.

Trace: the same traffic under three layouts

Baseline traffic (800 likes a second across the 16 posts), from the reference implementation:

Range, hash, directory

Most real designs combine the last two: hash keys into many fixed buckets, then map buckets to shards through a directory. The wallet loop hashes wallets into 1,024 buckets on 16 shards (step 2.1), the hotel loop does the same with 1,024 buckets on 4 clusters (step 2.4), and Uber's Schemaless has 4,096 fixed shards placed on storage clusters (the Uber case study, step 2.3). Hashing spreads the keys; the directory lets us decide where each bucket goes.

Synthesizing vector architecture diagram...

What to notice: keys never rehash. Moving capacity means changing directory rows, and a single bucket (here 717, holding a big customer) can be carved out to a shard of its own.

User, tenant and time keys

• User or entity keys (user_id, post_id, hotel_id) are the default: most requests carry one, and one entity's data stays on one shard.
• Tenant keys keep a customer's data together, so per-customer queries touch one shard. The cost is that a whale customer is one unit: the drill One Customer, One Shard, One Outage is exactly this, and Part 5 and Part 8 answer it.
• Time never leads the key. A key that starts with the time sends every new write to one place. Two fixes, both in the loops: bucket a growing key by time inside an entity key, as Discord does with (channel_id, bucket) in 10-day buckets, which bounds each partition's size (Discord case study, step 1.3); or shard a time key into sub-keys, like HOLD#<minute>#<0-63> in the notification loop (step 2.2) and <minute>#<shard> in the job scheduler loop (steps 1.1 and 2.1), so the current minute's writes land on many shards.

The key follows the most frequent query and the transaction boundary

Anything a layout can't answer on one shard becomes a scatter-gather (Part 9). So list the top queries and the transactions first, and pick the key they already carry. The payment loop shards by merchant (step 3.1) because a merchant's ledger is what changes together.

Key skew vs load skew

• Key skew: some shards hold more keys than others. Hash by post has none here: exactly 4 posts per shard.
• Load skew: some shards take more traffic than others, because some keys are busier. Hashing does nothing about it: in Part 3, S2 holds 4 posts like everyone else and takes 12.75 times the writes of each other shard.

The loops say the same thing in their own domains: "hashing spreads symbol counts evenly, not load" in the matching engine loop (step 2.7), which assigns symbols by measured load instead; and Uber's city-sharded dispatch struggled because cities differ so much in size and spikiness (Uber case study, step 1.1).

How many shards, how many buckets

Take the larger of two numbers, and round up:

• Load: peak writes ÷ (one shard's ceiling × a target such as 0.7).
• Size: total data ÷ the most you are willing to copy or restore for one shard.

Then choose far more buckets than shards, because the bucket count caps how far you can grow without rehashing every key. Worked, from the digital wallet loop: 38,750 transfers a second at the ceiling ÷ (4,000 per shard × 0.7) = 38,750 ÷ 2,800 = 13.8, so at least 14 shards; the loop takes 16 so that 1,024 buckets divide evenly, 64 per shard.

---

Part 3. A hot key appears

At 20:02 S2 is at 255% of its write ceiling while the cluster as a whole is 79% busy. A hot key is one key whose demand is more than its shard has left. It is invisible in the average and fatal to its shard.

Trace: p9goes viral

Synthesizing vector architecture diagram...

Synthesizing vector architecture diagram...

Snapshot S2, from the reference implementation. What to notice: one bar crosses the 100% line in each chart, and it is the same shard. The other three are unchanged.

Collateral damage. S2 throttles by shard, not by key: p1, p5 and p13 are rejected too, although their own traffic never changed. Their owners did nothing wrong except share a shard with p9.

Why more shards don't help

Whatever the layout, p9 alone is 2,400 writes and 9,000 reads a second on one shard: 240% and 300% of a ceiling that no fleet size changes. The same "add nodes doesn't split one key" lesson is in the key-value store loop (step 2.7), the message queue loop (step 2.7), the rate limiter loop (step 2.3), Discord (step 2.1) and Uber (step 2.1: "an airport's S2 cell is still one key on one owner").

Finding the key

Two measurements, in this order:

1. Per-shard utilization against its ceiling tells you where: S2 at 255% and 310%.
2. Per-key counts tell you which. Routers sample requests and keep a top-K per shard. From the reference implementation, p9 is 94.1% of S2's writes (2,400 ÷ 2,550) and 96.8% of its reads (9,000 ÷ 9,300); a 1% sample over 5 seconds already put it at 118 of 127 sampled writes.

On DynamoDB, CloudWatch Contributor Insights does step 2 for you. It has two modes: throttled keys (only the most throttled items, cheap enough to leave on) and accessed and throttled keys (also the most accessed items). One blind spot to know: write throttling caused by a global secondary index that lacks capacity isn't counted in the throttled-items graph. Watch the index's own most-accessed graph for that (Part 7).

textCopy
EVERY SAMPLED REQUEST at a router (1 in 100)
  1. topk[shard][key] += 1 (a bounded top-K sketch per shard)
  2. every 5 s: for each shard over 70% of a ceiling,
       report the keys whose share of that shard is over 10%
       and whether the share is reads, writes or both

EVERY REQUEST at a router (admission)
  3. if key has a token bucket and it is empty    -> 429, Retry-After
  4. if the tenant's bucket is empty              -> 429, Retry-After
  5. else take a token from each and route (Part 1)

Reads or writes?

Read-hot and write-hot keys need different fixes, so classify first. p9 is both: reads are 300% of the leader's ceiling (Part 4 fixes them), and writes are 240% (Part 5). A fix for one does little or nothing for the other.

Little's law says why a hot shard hurts so fast: requests in flight = arrival rate × time each spends in the system. At 9,300 reads a second and 5 ms, S2 holds about 46.5 requests at once; if throttling and queuing stretch that to 500 ms, it holds 4,650, and every client waiting on p1, p5 and p13 queues behind them.

A limit per key

The first guard is to refuse load a shard can't take, before it takes the shard down. Routers enforce budgets per key and per tenant:

A limit protects the neighbours; it doesn't serve the celebrity's fans. Parts 4 and 5 do that. Isolating whole tenants in cells, and shuffle sharding, are the Multi-Region Failover loop primitive's subject (coming).

---

Part 4. Hot reads: cache, coalesce, copy

p9 gets 9,000 reads a second, and they are all the same question: "what does post p9 look like now?" Asking the shard 9,000 times a second for one answer is the waste to remove.

Cache it for a second

Each router keeps a micro-cache: p9's answer for 1 second, with two extras from the news-feed loop's step 2.2:

• Single-flight: on a miss, only one fetch goes to the shard; every other request for the key waits for that fetch.
• Stale-while-revalidate: an entry older than 1 second but younger than 3 seconds is still served, while one refresh runs in the background.

textCopy
READ(k) at a router
  1. entry fresh (age < 1 s)                   -> serve it
  2. entry stale but age < 3 s                 -> serve it; if no refresh is running, start one
  3. no usable entry:
       a fetch for k is in flight              -> wait for it
       else                                    -> start the fetch; everyone waiting gets its result
  4. a write to k seen by this router          -> drop the entry

The staleness is now a stated number, not an accident. A cache in front of a hot key must have one: a short TTL, or an invalidation path that every write takes. The news feed loop's step 2.2 met the counter-example: the DAX query cache isn't invalidated by writes to the table.

Why not cache it forever?

Because then a caption edit, a takedown or a price change is invisible until the entry is evicted: a write you can't see for a day. And a cache restart is the worst moment: every entry is gone at once, so thousands of readers miss together and each would query the shard. Single-flight is what saves that moment, the cold-cache stampede in the drill The Product Page That Melted Redis: each router sends one fetch per key and everyone else waits for it. A short TTL with single-flight bounds both the staleness and the stampede.

Coalesce identical reads

Some reads can't be cached, even for a second (a check that must see the latest write). Request coalescing still helps: while a query for p9 is in flight, identical requests join it instead of starting their own. A request may join only if no write to the key completed since the query began, or it could get an answer older than a write it already saw acknowledged.

With requests arriving at rate λ and each query taking t, the shard sees:

$$\text{queries per second} = \frac{\lambda}{1 + \lambda t}$$

The formula is exact for random (Poisson) arrivals and a fixed query time; the reference implementation's simulation matched it to within 0.2% (176.25 against 176.47, 195.61 against 195.65).

Synthesizing vector architecture diagram...

What to notice: three requests, one query. The next request after a completed write can't join an older query, so it starts a new one.

Coalescing isn't a cache. A coalesced result lives only while its query is in flight, 5 ms here. It never serves an answer older than one query, and it does nothing for requests that don't overlap. A micro-cache serves one answer for a whole second. Use the cache where a second of staleness is fine, and coalescing where it isn't.

The edge can do the same in front of the routers: a CDN that collapses identical requests for one object into one fetch to the origin (Part 11).

Replica reads

Letting the shard's two replicas serve reads too raises S2's read ceiling to 3 × 3,000 = 9,000 a second. S2 needs 9,300: still 103% (event 7a in Part 12). And replica reads are stale by the replica's lag (page 06); they raise the ceiling of the whole shard by 3 times, whatever the key; and they do nothing for writes, which every copy must still apply. Useful for a warm shard, not a cure for one hot key.

Copies of one value

The search autocomplete loop uses the same idea (step 3.2, more replicas for hot shards). Every fix for hot reads is compared, with the write fixes, in one table at the end of Part 5.

---

Part 5. Hot writes, and every fix on equal terms

p9's counter takes 2,400 likes a second, and one shard takes 1,000 writes. Caching doesn't help a write: each like must change the count. Either the writes go to several places, or fewer writes are sent.

Salting

Salting (write sharding) splits one key into S sub-keys and sends each write to one of them. A read adds them back up.

The naive sizing is S = rate ÷ ceiling = 2,400 ÷ 1,000 = 2.4, so 3. It forgets that a shard carries every key hashed to it. Three sub-keys of 800 a second each, on shards that already carry 200, put those shards at exactly 1,000: 100%. The right rule, in words (Formula 2): each shard's load is its other traffic plus, for each sub-key it holds, the key's rate ÷ S; choose S and the sub-keys' placement so every shard stays at or under 80%. Here each non-S2 shard has 800 − 200 = 600 a second to spare, so the 2,400 must be spread evenly over all four shards.

We take S = 8, with the salt inside the hash tag: likes:{p9#0} to likes:{p9#7}. Each sub-key takes 2,400 ÷ 8 = 300 likes a second.

Synthesizing vector architecture diagram...

What to notice: every shard carries two sub-keys, and each shard's total counts its other traffic, not just its share of p9. (Arrows to the second sub-key of each shard are left out to keep the picture readable.)

On four shards, S = 4 would do the same: suffixes #0 to #3 already land one per shard, and give the same 750 and 800 with half the read fan-out. Loops pick S larger (10 to 64) to leave room for growth and for collisions they can't see. Changing S later means every reader must learn the new S first, and for an index whose reader looks up one member (the news feed loop's FOLLOWED_BY#<id>#<n>, step 2.1), a backfill too.

When salts collide, and the tag trap

So: where you know the slot function (Valkey, your own directory), check where each sub-key lands and pick suffixes that spread. Where you can't see partitions (DynamoDB), leave headroom for collisions with a larger S. DynamoDB's documentation also describes calculated suffixes, which solve a different problem: the suffix is derived from an attribute of the item, so a reader of one item knows which sub-key to read, without reading all S.

Pre-aggregation

Most of those 2,400 writes are "+1". Each router can add them up and write the total. That is pre-aggregation (write sharding with local sums):

(A store with conditional writes can instead take one conditional write per window: each router adds its 500 ms sum under "apply only if this window's id is new", so a retry can't count twice. Our per-router sub-keys do the same with seq.) The switch is itself a small migration: the eight salted sub-keys stop taking writes, and a read of the count adds their frozen sum, which never changes again, to the six router sub-keys.

textCopy
PRE-AGGREGATE p9's likes, on router N
  0. (precondition) nothing else writes the hot key once per event
  on each like:
    1. write "u liked p9" to liked:{u} (durable); like-events is fed from liked's change stream
    2. acknowledge the like
    3. buffer += 1
  every 500 ms:
    4. total = total + buffer; seq = seq + 1; buffer = 0
    5. write likes:{p9}:rN = (total, seq), applied only if seq > the stored seq
  on restart:
    6. read likes:{p9}:rN back: total and seq continue from there
  a read of the count:
    7. sum likes:{p9}:r0 .. r5 (one multi-key request, one slot) plus the frozen salted sum

Nothing else may write the hot key

Pre-aggregating a counter only helps if nothing else writes the hot key per event. If each like also wrote a record "u liked p9" under p9's tag, S2 would still take 2,400 writes a second and the counter's 12 would change nothing. That is why the per-like records are keyed by the user, liked:{u}, in a separate table: they spread over that table's shards by user, and "who liked p9?" becomes a derived index or a scatter (Part 9). The same shape shows up in the mobile news feed loop (R2.5), which sums counters from the change stream instead of incrementing per like.

When a router crashes

A flush can fail, be retried, or never happen. The design above makes each case harmless:

The 211 likes are not lost: each was durable in liked:{u} before its "OK", and reaches like-events through the table's change stream. The count is 211 low (the reference implementation checked: short by exactly R4's lost buffer) until the job that rebuilds each post's count from like-events every few minutes (our design choice) corrects it: it recounts p9's likes and writes the difference from the sub-keys' sum into a correction sub-key likes:{p9}:fix, which reads also add. The count is a view, rebuildable from durable events. Why the flush must carry an absolute total and not "+211" is the Idempotency & Effectively-Once Processing loop primitive's rule (Part 5): an additive flush that is retried counts twice.

When the pre-aggregation runs inside a stream job instead of a router, it has more rules (lateness decided per event, partials flushed before a watermark and at a checkpoint barrier); those belong to the Event Time, Watermarks & Checkpoints loop primitive (Part 3). The ad-click loop (step 2.5) and the YouTube loop (step 2.5) both pre-aggregate this way.

Leasing and striping

• Token leasing is pre-aggregation for a limiter: instead of decrementing a hot tenant's counter on every request, each server leases a block of tokens and spends them locally (rate limiter loop, step 2.3).
• Striping a balance handles a hot account that has a floor (it may not go below zero), where salting's "read all S and add" would let two stripes spend the same money. Credits land on stripes; a sweeper moves stripes into the main account; debits come only from the main account, where the floor is checked under one lock. S is sized from the lock hold time (digital wallet loop, step 2.3; payment loop, step 3.1). The cost is delay: a credit is spendable only after the sweep.

Counters that must merge concurrent increments without coordination can also be CRDT counters, one sub-count per writer; that is the same idea as the per-router sub-keys.

A dedicated shard

Moving p9's slot onto a shard of its own protects p1, p5 and p13, but before the fixes that shard alone would carry 2,400 writes (240%) and 9,000 reads (300%): isolation, not a cure. A dedicated shard can't give one key more than one shard's ceiling. With the fixes on, it is worth doing, and Part 8 does it, live. For a customer you know is coming (the drill's 10× tenant), carve out its buckets before launch day.

GSI write sharding

A hot key in a secondary index is as dangerous as one in the table. In DynamoDB, a global secondary index that can't keep up throttles writes to the base table. If an index is keyed by creator_id, every write about any of nova's posts lands on one index key: Part 7 shows 2,550 a second. The fix is the same salting, on the index key: creator_id#0 to creator_id#7, with reads querying all 8 in parallel and merging (URL shortener loop, step 2.5: owner_id#0..15).

Every hot-key fix on equal terms

Each row is filled from both sides: does a read fix help writes, and does a write fix help reads? Numbers are p9's, from the reference implementation.

Read the table down the "writes" column: nothing for reads touches writes. Read it down the "reads" column: salting makes reads worse unless you cache the sum. That is why p9 needs one fix from each half.

Snapshot S3, after event 9 (micro-cache and pre-aggregation on), from the reference implementation:

---

Part 6. Adding capacity: rings, slots and splits

Day 2, on a copy of the cluster: the app is growing, and every post's baseline has doubled to 100 likes and 200 reads a second, so each shard is at 400 writes a second (40%). We add a fifth shard, S5. The question is how much data moves, and whether we get to choose which.

Trace: three ways to add S5

From the reference implementation, over 10,000 synthetic keys and our 16 posts:

hash mod N

mod N is the naive answer every loop warns against: the S3-like storage loop (step 1.1: hash % 100 → 101 moves almost everything) and Uber's dispatch (step 2.1: adding the 301st machine moves 300 of every 301 keys).

Doubling is the best mod N can do, which is why "double the partitions" is a common step. It still moves half the data, and for an ordered topic it moves half the keys' new records to a different partition (Part 7).

The ring, with one point and with many

With one token per shard, each shard owns one arc of the ring, and random arcs are very uneven. Adding S5 cuts one arc in two:

Synthesizing vector architecture diagram...

Arcs in ring order, positions as fractions of the ring, from the reference implementation. A key belongs to the next token at or after its hash. What to notice: S5's token lands inside S3's arc, so S5 takes everything between 0.037 and 0.381 from S3, and S0, S1 and S2 shed nothing.

Synthesizing vector architecture diagram...

What to notice: with many tokens, S5's points cut many different arcs short, so it takes a slice from several shards instead of all from one. With only 16 random tokens the slices are still uneven: in our run S0 gave 1,388 keys and S3 gave 28.

How many tokens is enough? Over 30 random layouts in the reference implementation, the busiest shard after adding S5 held a median of 1.26 times the average with 16 tokens, 1.11 with 100 (Uber's Ringpop default, step 2.1) and 1.07 with 256. Many small pieces are what make a move even. Cassandra defaults to 16 tokens per node, and pairs them with a token allocator that places new tokens to balance load for a given replication factor, instead of picking them at random. This answers the drill The Ring Rebalance That Crushed Node 07: without virtual nodes, a new node takes keys only from the one node after it on the ring, so the old nodes don't each shed a fair share.

Fixed slots and a directory

Nothing is rehashed: keys stay in their slots, and we decide which slots S5 takes. Because we know where the load is, we choose one block of 819 or 820 slots from each old shard, each holding one of our posts:

Snapshot S4 (day 2, on the copy), from the reference implementation: S5 takes 3,277 slots and 19.8% of the synthetic keys; with 4 posts at 100 likes a second, S5 is at 400 writes/s (40%), and S0 to S3 drop to 3 posts each, 300 writes/s (30%).

The Dynamo paper (2007) found that fixed, equal-sized partitions assigned to nodes balanced load best of the strategies it tried, and turned a move into copying whole partition files. Uber's Schemaless grows the same way: a fixed 4,096 shards, and "we typically expand by splitting each MySQL server in two" (case study, step 2.3). The wallet and hotel loops' 1,024 buckets are the same design.

Split and merge

Stores that own ranges (Bigtable-style, and DynamoDB internally) grow differently: a range that grows too big or too hot is split in two, and cold neighbours are merged. Split at the key that halves the load, not the key count. On our copy, S2's load sits in one slot, so a load split cuts at 12,067: p13, p1 and p5 (slots 10,396 to 11,951) on one side, slot 12,067 alone on the other, and no split goes finer than one slot, or one key (event K7c in Part 12). The cost is brief throttling of the range while it splits (DynamoDB: splits "usually complete in the order of minutes"). A range whose writes all go to its tail (a monotonic key) can't be helped by splitting either: the new tail takes every write. The S3-like storage loop (step 2.6) spreads such a tail over hash-prefixed sub-ranges instead.

Bounded load, briefly

Consistent hashing with bounded loads caps each node at, say, 1.25 times the average load and sends overflow keys to the next node in ring order. It evens out skewed traffic for stateless placement (which server handles a request) and for owners a controller can move (the web crawler loop, step 3.1). It is the wrong tool for a store: the key's owner now depends on current load, so a spilled key's request goes to a node that doesn't have its data, and ownership moves whenever loads shift. And like vnodes, it never splits one hot key. This is the drill's second question, and Uber's answer in step 2.1.

On equal terms

---

Part 7. DynamoDB and Kinesis: managed splitting never splits one key

Managed services split partitions for you. So does the celebrity post just work on DynamoDB and Kinesis? The same p9, replayed on copies of each.

DynamoDB: adaptive capacity and split for consumption

D2: a hot GSI key. Suppose post_likes has a GSI keyed by creator_id. Every like of any of nova's four posts writes the index key nova: 2,400 + 3 × 50 = 2,550 writes a second on one index partition key, which throttles the index and, through it, the base table. Salt the index key: nova#0 to nova#7, 318.75 a second each, and query all 8 in parallel to read it. Contributor Insights won't show the index's write throttling in its throttled graph (Part 3); the index's most-accessed graph will.

Kinesis: shards, splits and lineage

A Kinesis stream is a set of shards. Each record's partition key is hashed with MD5 into a 128-bit integer, and the shard whose hash-key range contains that integer takes the record. Each shard takes 1 MB/s or 1,000 records/s of writes, whichever comes first, and serves 2 MB/s and 5 GetRecords calls a second of reads. The limit is shared by every key hashed to the shard.

Synthesizing vector architecture diagram...

What to notice: after a split the parent stops taking records but keeps the ones it has. A reader that starts the child holding p9 before finishing the parent can see p9's newer events before its older ones.

K6: resharding the whole stream. UpdateShardCount 4 → 8 with uniform scaling: p9 ends on shard 6 of 8. The 4 parents become CLOSED and later EXPIRED; the 8 children take new records. By default UpdateShardCount can't go above double or below half the current count in one call, or run more than 10 times per rolling 24 hours per stream. Reading children only after their parents is the Change Streams & the Transactional Outbox loop primitive's subject (Part 9).

K5: packing. Putting 9 likes in one record cuts 3,150 records a second to 350, but a record has one partition key for everything inside it, so the key can no longer be post_id. Packing only works once the key doesn't have to name an entity.

Kafka, in one sentence. Kafka picks a partition with toPositive(murmur2(key)) % numPartitions, so adding partitions sends about half the keys' new records to a new partition when doubling (the reference implementation: 50.4% of 10,000 keys going 4 → 8; p9 happened to stay in partition 2), existing records stay where they are, and the partition count can't be reduced; what that does to order is the Change Streams page's Part 4.

---

Part 8. Moving live data

The fixes hold tonight, but nova posts daily and every spike still lands on p1, p5 and p13's shard. We give p9's slot a shard of its own, and we do it while it's being liked. That means copying data that keeps changing, switching the owner without losing a write, and coping with routers that haven't heard.

At 20:10 slot 12,067 holds post:{p9}, the six counter sub-keys and p9's comment thread: 200 MB (labelled). With the fixes on, the slot takes about 12 writes a second (the counter flushes), plus a caption edit and comments now and then; S2 as a whole takes about 162.

Copy and tail

Starting from a snapshot at a known position and then applying the log from the next position is the same bootstrap rule as the Change Streams & the Transactional Outbox loop primitive's Part 7. Instead of shadow reads, a mover can compare by scanning the whole slot on both sides at one position (Vitess calls its tool VDiff).

Synthesizing vector architecture diagram...

What to notice: S2 stops accepting the slot at the fence, before anyone owns it elsewhere, and S4 starts only after the directory commit. R5 never heard about v8, and S2's own slot table sent it to S4.

Fence and flip

textCopy
MOVE slot s from source to target        (directory at version v, slot epoch e)
  1. copy a snapshot of s at log position p, from a copy applied through p
  2. target tails the source's log from p + 1, applying s's entries
  3. shadow-read (or scan-compare) until source and target agree by position
  4. FENCE: source appends "s: migrating, epoch e + 1" to its log as entry f;
       from now until step 5 commits, it answers every read AND write for s
       with "slot migrating, retry"; routers retry with backoff
  5. when the target has applied through f (the fence entry itself):
       commit directory version v + 1: s -> target, epoch e + 1   (the commit point)
       target slot table: "s: owned, epoch e + 1"
       source slot table: "s: moved to target, epoch e + 1"      (kept forever)
  6. routers learn v + 1 by push or poll; any request that reaches the source
       gets MOVED, reloads the directory and retries (it must be safe to repeat)
  7. after the compare is clean, delete the source's copy of s;
       the source still answers MOVED for s from its slot table
  8. abort before step 5 = drop the target; after step 5, roll back by the same move in reverse
  9. invariant: exactly one shard accepts reads and writes for s,
       and it holds every write any shard accepted for s before

The fence is the Leases, Fencing Tokens & Distributed Locks loop primitive's rule (Part 4) applied to a slot: the epoch is the fencing token, and the check happens at the place that holds the data.

Synthesizing vector architecture diagram...

Snapshot S5, at 20:11:10.500. What to notice: the router with the old map reaches the old owner, and the old owner's own table turns it away; nothing depends on the router having heard about v8.

Without the slot table

In the hotel reservation loop's terms, that lost write is a hold taken on the old copy: an oversold room. In the digital wallet loop, it is a double spend. Both loops use this page's rule: the old shard keeps a permanent "moved" row for the bucket, checked under the same lock by every transaction, answered with a retryable error that makes the service reload the map (hotel step 2.4, wallet step 2.1 and R2.11). Uber met the same problem with ring owners during membership churn and fenced at the store with a version check (case study, step 2.7).

When the move fails

Abort before the flip = drop the target. Rollback after the flip = the same move in reverse. Vitess keeps a reverse replication stream from the new owner back to the old one after it switches writes, so that ReverseTraffic can roll back quickly; its documentation warns that turning reverse replication off means "you will not be able to rollback once you have switched write traffic".

Dual writes instead of a log tail

Some teams keep both copies current by having the application write to both during the move, and backfill the old data separately. It works only with versions (event 23 in Part 12, from the reference implementation):

• The backfill overwrites a newer write. It read the caption at v4, a live dual write then put v5 on both shards, and the backfill's late write of v4 landed on S4: S2 says v5, S4 says v4.
• Two writers race. Routers R1 and R2 write caption v5 and v6 to both shards; S2 receives v5 then v6, S4 receives v6 then v5. Without versions, S2 ends at v6 and S4 at v5.

The cutover order: dual-write, backfill, run shadow reads until they match, flip reads to the target, then stop writing the source. With a version on every write and "apply only if newer" on both shards, both cases converge (S4 keeps v5 in the first, both end at v6 in the second). The version must come from one authority per key (the source's log position or a per-key counter), never from router clocks, which disagree (page 03, Part 3). Discord's migration backfilled with the original write timestamps (case study, step 3.3), and Uber's Mezzanine backfill and mirrored writes converged because both wrote each trip version under the same ref key (step 2.3).

Moving key by key

Valkey and Redis Cluster move a slot without freezing it, one key at a time:

It avoids a slot-wide pause, but a large key moves in one piece (ElastiCache doesn't migrate keys holding a large item during online resharding, and won't delete a shard holding an item over 256 MB), and multi-key requests on a moving slot keep retrying. Our fence is the whole-slot version: one short pause, one switch.

What the move costs everyone else

The copy reads S2's replica's disk and uses the links between AZs while S2 keeps serving, so it is throttled below the spare I/O (Cassandra's stream_throughput_outbound defaults to 24 MiB/s for the same reason). When the hot slot is big and the cold ones small, it can be cheaper to move the cold slots away instead: p1, p5 and p13 (slots 11,819, 11,951 and 10,396) are three small copies that give the same isolation (event 24 in Part 12).

At scale. A 10 TB cluster going from 4 to 5 shards, as in Part 6, moves a fifth of its data: 2 TB. At 100 MB/s aggregate, throttled below the sources' spare I/O: 2,000,000 MB ÷ 100 MB/s = 20,000 s, about 5.6 hours. Buckets move in parallel, each with its own milliseconds-long pause. The move's length is set by bandwidth, and each bucket's pause by catch-up.

Engines that do it for you

---

Part 9. Across shards: queries and transactions

Everything so far touched one key and one shard. Some queries don't name a key, and some changes touch two. At 20:15 the app shows "Trending now": the top 10 posts by likes in the last hour, from every shard.

Trace: "trending now"

The router asks every shard (now five: S0 to S4) for its top 10 and merges them. Each reply is about 20 KB (labelled), and clients ask 1,000 times a second.

Synthesizing vector architecture diagram...

Snapshot S6. What to notice: the answer waits for the slowest of the five replies, and from AZ a four of the five replies cross an AZ boundary.

The slowest shard decides

A scatter-gather's latency is the slowest reply's, not the average. If each shard is slow (over 50 ms, labelled) 1% of the time, independently (a slow reply takes 50 to 200 ms, a normal one 3 to 7 ms, labelled), the chance that at least one of N is slow is:

$$P(\text{slow}) = 1 - (1 - p)^{N}$$

At 5 shards in the reference implementation, a query's 99th-percentile latency was 168.5 ms against 51.0 ms for a single shard: the rare slow shard is now in every hundredth query's path. Two standard defences: a per-shard timeout that returns partial results (for a "trending" list, 4 of 5 shards is fine), and hedged requests (after a short delay, send a second request to a replica and take the first answer).

A popular query is a hot read

The same top 10, asked 1,000 times a second, is Part 4's hot read again. Compute it once a second in one job and cache it. The scatter-gather then runs once a second, not 1,000 times, and moves 5 × 20 KB = 100 KB a second.

The bytes matter too. EC2 bills data crossing AZs at $0.01/GB in each direction. With routers spread over the three AZs, a scatter from each router crosses an AZ for 3.33 of the 5 replies on average (4 from AZ a, 3 from b, 3 from c). Uncached, that is 1,000 × 20 KB × 3.33 = 66.7 MB/s crossing AZs, 172,800 GB a month, about **$3,456 a month** at $0.02/GB (both directions). One job in AZ a, once a second: 80 KB/s, 207 GB a month, about **$4.15**. Salting multiplies the same bytes by S for every uncached read.

Local and global indexes

Reports run on a copy

A platform-wide report, such as the drill's admin report over all tenants, shouldn't scatter across production shards at all. Stream every shard's changes to a warehouse or data lake and run it there, seconds to minutes behind: the wallet loop does it with S3 and Athena (step 2.1), and the hotel and payment loops do the same.

Transactions that span shards

For a follow, the derived index is the natural fit: nobody needs alice's follower list to change in the same millisecond. "Who liked p9?" is the same shape: the liked table is keyed by user, so the per-post list is a derived index built from like-events, or a scatter.

---

Part 10. End to end through the layers

Where is each decision actually taken? Follow one like on p9 from a phone whose request lands on a router in AZ b at 20:20, after the move, and one read.

Synthesizing vector architecture diagram...

What to notice: the like is acknowledged after step 1; step 2 is the router's periodic flush, which the shard checks against its slot table and epoch before it reaches storage.

A read of p9 stops at the router's micro-cache almost every time; about once a second per router it reaches S4 as one multi-key request (the post plus six sub-keys), and on the way back it refills the cache. Hot keys are cheapest to absorb near the client.

Sizing per AZ

Routers run in every AZ, so we size them to survive losing one. Peak traffic: 3,150 likes (15 × 50 + 2,400) plus 10,500 reads (15 × 100 + 9,000) = 13,650 requests a second. Each router takes 2,500 (labelled) and should run at 80% or less, so 2,000.

Fleets are placed per AZ, so round up per AZ, and size for the loss of one.

---

Part 11. On AWS

Every AWS data service has a unit, a way keys map to it, and a ceiling per unit. Knowing those three things tells you where your hot key will hurt. Stated only from AWS's public documentation and papers.

Managed services that use it

Running it yourself

Look-alikes that are not this mechanism

---

Part 12. What you've learned

Back to the celebrity post

Our table was 20% busy while one post took 33 times a partition's read ceiling and 10 times its write ceiling. In the tiny store, p9 put S2 at 255% of its write ceiling and 310% of its read ceiling. The pieces:

• One key lives in one piece (Part 1), and hashing spreads keys, not load (Part 2). Adding shards moved other keys, never p9 (Part 3).
• Hot reads stopped at the routers: a 1-second micro-cache with single-flight cut 9,000 reads a second to 6, and coalescing routed by key cut uncacheable reads to 196 (Part 4).
• Hot writes were summed at the routers: 2,400 likes a second became 12 absolute-total flushes, safe to retry and rebuildable from durable events, because nothing else wrote p9's slot per like (Part 5). Salting was the alternative, at 8 reads per count.
• Growth moved exactly the slots we chose (Part 6); DynamoDB and Kinesis split for us, but never below one key (Part 7).
• p9 got its own shard while it was being liked: copy, tail, compare, fence, flip, with a 13 ms pause and a stale router bounced by the old owner's slot table (Part 8).
• Queries across shards paid the slowest shard's latency, so the popular one was computed once a second (Part 9), and the routers were sized per AZ (Part 10).
• The costs: a second of staleness, 500 ms of count delay, a rebuild job, a directory, and slot tables that remember every move forever.

The whole story, event by event

From the reference implementation:

The cheat card

Failure checklist

• Does the key follow the most frequent query and the transaction boundary, and does no key lead with time?
• Is every shard's utilization alarmed against its own ceiling, not the cluster average?
• Can you name the hottest keys per shard within seconds (top-K at the routers, or Contributor Insights including the GSIs' most-accessed graphs)?
• Is there a per-key and per-tenant limit that answers 429 before a shard melts?
• Does every cache in front of a hot key have a stated staleness bound and single-flight on misses?
• Are salts inside the hash tag, placed or checked so they spread, and is the merged read cached?
• Does anything still write a pre-aggregated hot key once per event?
• Are flushes absolute totals with a sequence, and are the events durable before the "OK"?
• Does every shard check every request against its own slot table, and does the old owner refuse a moved slot for good?
• Is every live move throttled below spare I/O, copied from a caught-up source, and able to abort or roll back?
• Do readers of Kinesis or DynamoDB Streams finish a parent shard before its children, and is any Kafka partition increase planned as an operation?
• Are routers, coalescers and shards sized per AZ for the loss of one?

Think-first drills

Interview questions

Where to go next

• Replication, Quorums & Read-Your-Writes: the copies inside each shard, and which copy may be a copy source.
• Leases, Fencing Tokens & Distributed Locks: epochs and fencing tokens, the rule behind the slot table.
• Idempotency & Effectively-Once Processing: why flushes and retries after MOVED must be absolute or deduplicated.
• Change Streams & the Transactional Outbox: the log tail that moves a slot, derived indexes, Kafka partition increases (Part 4) and stream lineage (Part 9).
• Event Time, Watermarks & Checkpoints: pre-aggregation inside a stream job (Part 3), and Flink's key groups when a job rescales (Part 9).
• Write-Ahead Log, fsync & Group Commit: what a shard's log position is.
• The Multi-Region Failover loop primitive (coming): cells, shuffle sharding and home Regions.
• Background: Primitive #08: Database sharding and partition keys, Primitive #01: Consistent hashing, Primitive #04: Distributed cache patterns and Primitive #10: Two-phase commit and sagas.
• Drills: The Product Page That Melted Redis (answered in Part 4), One Customer, One Shard, One Outage (Parts 5 and 8 for the tenant that grows 10×, Part 9 for the cross-tenant report) and The Ring Rebalance That Crushed Node 07 (Part 6).
• Loops: the key-value store (steps 1.3, 2.7, R2.8), the message queue (steps 1.4, 2.1, 2.7), the rate limiter (steps 2.1, 2.3, R2.6), the URL shortener (steps 2.1, 2.2, 2.5), the news feed (steps 2.1, 2.2), chat (steps 2.1, 2.2), the gaming leaderboard (steps 2.6, 3.1), notifications (step 2.2), S3-like storage (steps 1.1, 2.6), search autocomplete (steps 2.2, 3.2), the web crawler (steps 2.3, 3.1), metrics (steps 2.2, 3.1), the job scheduler (steps 1.1, 2.1), payments (step 3.1), the digital wallet (steps 2.1, 2.2, 2.3, R2.11, R3.2), the matching engine (steps 2.7, 3.6), hotel reservations (steps 2.1, 2.4, R3.9), the transactional outbox ledger (R1.11), the proximity service (steps 2.1, 2.5), maps and routing (steps 2.1, 3.1), nearby friends (steps 1.1, 2.3), ride-sharing dispatch (steps 1.1, 1.4, 2.6), YouTube (step 2.5), Google Drive (step 2.2), ad-click aggregation (steps 2.5, R2.6, R2.8), email (steps 2.4, 2.6), the mobile news feed (R2.5), mobile stock trading (steps 2.1, 2.3), and the Discord (steps 1.3, 2.1, 3.2, 3.3) and Uber (steps 1.1, 2.1, 2.3, 2.7) case studies.