Partitioning & sharding
Caching (lesson 06) bought you read scale by keeping a copy of hot data close. But a cache cannot hold a dataset that no longer fits on one machine, and it does nothing for write throughput. When the authoritative state outgrows a single box, you have no choice left: you split it. This is where distributed systems stop being convenient and start being hard.
1. The first principle: when one box is no longer enough
A single node has three hard ceilings. You partition when you hit any of them:
- Storage. A 4 TB dataset will not fit on a node with 512 GB of SSD, full stop.
- Write throughput. Reads scale by adding replicas (lesson 08), but every write must hit the single primary. One primary doing fsync-durable writes tops out around 10–50k writes/s; past that, the only lever is more primaries, each owning a slice of the keyspace.
- Working set / index in RAM. A B-tree index (what an index physically is, and B-tree vs LSM, is 04) that no longer fits in memory turns every lookup into a disk seek — a 100× latency cliff (100 ns RAM vs ~100 µs NVMe). Splitting the index across nodes keeps each shard’s index resident.
Partitioning (a.k.a. sharding) splits the dataset into disjoint partitions (shards), each owned by a different node. It buys near-linear scaling of storage and write throughput: N shards → ~N× capacity. It is the orthogonal axis to replication — partitioning splits the data; replication copies each piece. Real systems do both: shard into 64 partitions, then keep 3 replicas of each.
- Cross-shard queries become scatter-gather. A query that can’t be answered by one shard must fan out to all of them and merge results — tail latency now equals the slowest shard (recall lesson 02’s tail-amplification math).
- Cross-shard transactions are hard. Atomicity across two primaries needs 2PC or sagas (forward-ref lesson 12). Single-shard transactions stay cheap; the design goal is to keep them single-shard.
- You must choose a partition key — and live with its skew. The key decides everything below. Choose wrong and you get a hot shard you cannot rebalance away.
2. Partitioning schemes
Three families, each a different answer to “given a key, which shard?”
| Scheme | Mapping | Range scans | Load spread | Main failure mode |
|---|---|---|---|---|
| Range | keys sorted into contiguous ranges (A–F, G–M, …) | Excellent — adjacent keys co-located | Poor — depends on key distribution | Hot spot: a monotonic key (timestamp, auto-increment id) sends all new writes to the last shard |
| Hash | partition = hash(key) mod N (or via a ring) | Dead — adjacent keys scattered, range = scatter-gather over all shards | Excellent — uniform by construction | Kills locality; cannot do efficient ORDER BY / time ranges |
| Directory | a lookup service stores key → shard explicitly | Whatever you encode | Tunable — you place data deliberately | Extra hop on every request; the directory is a SPOF / bottleneck if not itself replicated |
Composite strategies combine them: hash a high-level prefix to pick a shard, then range-sort within it (e.g. Cassandra’s (partition_key, clustering_key) — hash for spread across nodes, sort for range scans inside a partition). This recovers in-partition range scans without re-introducing global hot spots.
The resize problem with naive hashing
The obvious hash scheme is shard = hash(key) mod N. It distributes beautifully — until N changes. Add or remove one node and the modulus changes, so almost every key maps somewhere new. When N goes from 4→5, a key’s shard is unchanged only if h mod 4 == h mod 5, which holds for a vanishing fraction. In general changing N reshuffles roughly (N−1)/N of all keys — ~80% at N=5, approaching 100% as N grows.
Make it felt. Take ten keys with hashes 20…29 and recompute their shard when you grow from N=4 to N=5:
Six of these ten keys land on a different shard. The four that stayed (20–23) are exactly the keys where h mod 4 and h mod 5 happen to coincide — which, because 4 and 5 share no common factor, recurs only once every lcm(4,5)=20 keys (whenever h mod 20 is 0, 1, 2 or 3). Walk a full cycle of 20 keys (0…19) and the count is exactly 16 moved, i.e. 16/20 = 80% = (N−1)/N. So adding one node doesn’t nudge a few keys — it relocates four keys in five. A beginner should picture almost the whole room standing up and swapping seats.
3. Consistent hashing — the centerpiece
The fix: stop tying placement to N. Picture a clock face. The servers stand at a few of the hour-marks, and each data item is also a point on the rim — an item belongs to the next server you reach walking clockwise. Add a server and it drops onto the rim at one spot: only the items between it and the previous server change hands; everyone else keeps the same owner. That is why a circle localises the disruption to ~1/N of keys, where mod-N renumbered the whole room. Formally: map both keys and nodes onto a fixed circular hash space (a ring, e.g. 0…232−1). A key is owned by the first node clockwise from its position. Adding or removing a node only re-homes the keys in the arc between that node and its predecessor — about 1/N of keys, not all of them.
keys moved when node added ≈ 1/N vs mod-N reshuffle ≈ (N−1)/N
Virtual nodes (vnodes): fixing the lumpiness
Plain consistent hashing has two problems. First, with few nodes the random arc lengths are uneven — one node may own a 2×-wide arc and thus 2× the load. Second, removing a node dumps all its load onto its single clockwise neighbour, doubling that node’s burden and often cascading.
The fix is virtual nodes: each physical node is hashed onto the ring at V points (e.g. V=128 or 256). Now a physical node owns 128 scattered small arcs. When it leaves, its load is sprinkled across many neighbours, not dumped on one. And because you’re averaging over V·N random arcs, the load skew shrinks. With more independent random placements landing on each physical node, the lumps average out — the same law-of-large-numbers reason a coin flipped 1000× lands nearer 50% than one flipped 10× — so the relative spread shrinks. The standard deviation of per-node load falls like 1/√V, so:
load skew = max load / avg load ≈ 1 + c / √V (c ≈ 2–3)
At V=1, skew can be 2× or worse; at V=256, skew is within a few percent of perfectly even. Vnodes also enable heterogeneous hardware: give a node with 2× the RAM 2× the vnodes and it carries proportionally more load automatically.
- Naive mod-N (4→5): fraction remapped = (5−1)/5 = 80% → 0.80 × 800 GB = 640 GB moves, and ~80% of cache entries go cold at once.
- Consistent hashing: fraction remapped ≈ 1/5 = 20% → 0.20 × 800 GB = 160 GB moves — the new node simply claims its ~1/5 slice from existing nodes. 4× less data, no herd.
- Skew, V=128: 1 + 2.5/√128 ≈ 1 + 0.22 = 1.22, so the hottest shard runs ~22% above average — provision for that, not for the average.
4. Choosing the shard key
The shard key is the most consequential decision in the design — it’s effectively irreversible at scale (changing it = full re-shard). Three requirements:
- High cardinality. Many distinct values, so partitions can be split finely.
countryhas ~200 values → you can never have more than 200 partitions, and the US shard dwarfs the rest.user_idhas millions. - Even distribution. No single value should dominate write/read volume.
- Query alignment. The dominant query should be answerable from one shard. If you shard by the field your hot path filters on, that read hits a single shard; otherwise it’s scatter-gather.
| App: social network | Shard by user_id | Shard by hashtag |
|---|---|---|
| Load a user’s timeline | Single shard ✓ | Scatter-gather ✗ |
| “all posts tagged #X” | Scatter-gather ✗ | Single shard ✓ |
| Distribution | Even (millions of users) ✓ | Skewed — a few tags are huge ✗ |
There is no key that makes every query single-shard — you optimize for the dominant access pattern and pay scatter-gather (or a secondary index) for the rest.
5. The hot-shard / celebrity problem
Here is the catch a good hash function cannot save you from: skew at the key level. Hashing spreads distinct keys evenly, but if a single key is disproportionately hot — a celebrity with 100M followers, a viral product on launch day — all of its traffic lands on whatever shard owns that one key. No partition scheme distributes the load of one key, because that key lives on one shard by definition.
- Split the hot key. Append a random suffix (
celebId#0…celebId#15) so writes spread across 16 sub-keys / shards; reads fan out and merge. Trades single-key locality for spread. - Cache it hard. A hot key is, by definition, highly cacheable — pin it in the cache tier (lesson 06) so the shard rarely sees the read traffic.
- Special-case it. Detect celebrities and route them through a different path — e.g. fan-out-on-read instead of fan-out-on-write (the hybrid timeline design, fully developed in the capstone (19)).
6. Secondary indexes & rebalancing
You sharded by the primary key, but queries also filter on other fields. Two ways to index them (for what an index physically is — B-tree vs LSM — see 04):
| Local / document-partitioned | Global / term-partitioned | |
|---|---|---|
| Index location | Co-located with the data on each shard | Index itself partitioned by the indexed term |
| Write cost | Cheap — one shard, local update | Expensive — write may touch a different index shard (cross-shard) |
| Read cost | Scatter-gather — query every shard’s local index | Single shard — go straight to the term’s index partition |
| Used by | Elasticsearch, MongoDB (default), Cassandra SI | DynamoDB GSI, search systems wanting fast term lookups |
It’s the same write-vs-read trade as everywhere: local indexes make writes cheap and reads fan out; global indexes make reads pinpoint and writes fan out.
Rebalancing without full reshuffles
Even consistent hashing benefits from a deliberate rebalancing strategy. The most robust trick: pick a fixed, large number of partitions up front (e.g. 1024), far more than nodes. Each node owns many partitions; adding a node means moving whole partitions to it, never re-hashing keys. (This is how Riak, Elasticsearch shards, and Kafka-style fixed partition counts work.) Avoid: hash-mod-N (full reshuffle), and unbounded dynamic splitting that thrashes. Rebalancing should move data in the background, throttled, and ideally be operator- or automation-triggered, not a surprise.
Interview prompts you should be ready for
- Why not just
hash(key) mod N? (senior answer) Because changing N re-mods nearly every key — ~(N−1)/N remap on a single add/remove — forcing a full data migration and a simultaneous cache-cold thundering herd. Consistent hashing pins placement to a ring so only ~1/N of keys move; or use a fixed large partition count and move whole partitions. - What do virtual nodes buy you? (senior answer) Two things. They smooth load skew (std-dev of per-node load falls like 1/√V, so skew ≈ 1 + c/√V), and on node removal they spread the departing load across many neighbours instead of dumping it all on one. They also let heterogeneous nodes carry proportional load by owning more vnodes.
- How do you pick a shard key? (senior answer) High cardinality, even distribution, and alignment with the dominant query so it hits one shard. I’d name the top query, shard to make it single-shard, and explicitly accept scatter-gather (or a secondary index) for the rest. Avoid low-cardinality keys like country — they cap partition count and create giant uneven shards.
- A celebrity user is overloading one shard. Now what? (senior answer) No partition scheme fixes a single hot key — it lives on one shard. Mitigate: split the key with a random suffix to spread writes across sub-keys; cache it aggressively since it’s highly cacheable; or special-case the path (fan-out-on-read for celebrities). I’d combine caching + special-casing.
- Local vs global secondary indexes? (senior answer) Local (document-partitioned) indexes sit with the data: cheap writes, but reads scatter-gather across all shards. Global (term-partitioned) indexes are partitioned by the indexed term: reads hit one shard, but writes may touch a different index shard. Pick by read/write ratio of the secondary query.
- How is partitioning different from replication? (senior answer) Orthogonal axes. Partitioning splits the dataset into disjoint shards (scales storage + write throughput); replication copies each shard (scales reads + provides fault tolerance). Production systems do both: shard into K partitions, then keep R replicas of each. They’re composed, not alternatives.
- How do you rebalance when adding capacity without an outage? (senior answer) Use a fixed large partition count (≫ nodes) and move whole partitions to the new node — no key re-hashing. Throttle the background data copy, keep serving from the old owner until the new replica is caught up, then cut over. Never hash-mod-N.
- Estimate the data moved when going from 4 to 5 nodes on an 800 GB dataset. (senior answer) Mod-N: (5−1)/5 = 80% → 640 GB moves plus a cache-cold herd. Consistent hashing: ~1/5 = 20% → 160 GB, the new node claiming its slice. 4× less data and no herd — that gap is exactly why we use the ring.