consistent hashing is the algorithm that decides which machine holds which piece of your data in every distributed cache, key-value store, and sharded database you will ever operate — and it is the answer to the question that breaks the naive design: "what happens to all your data when you add or remove a server?" The moment you spread data across more than one node you have to answer where each key lives, and the obvious answer — take the hash of the key, divide by the number of servers, keep the remainder — works beautifully right up until the day you change the number of servers, at which point almost every key you own suddenly belongs to a different machine. That single failure is the reason a whole family of partitioning schemes exists, and understanding it is the difference between a cache tier that scales smoothly and one that melts down the instant you touch it.
This guide is the walkthrough you wished existed the first time an interviewer drew a circle on the whiteboard and asked "how would you shard this so growing the cluster doesn't move everything?" It builds the idea in layers: first the modulo scheme and the precise reason resizing it triggers a rebalancing storm, then the hash ring that limits key movement to a single arc, then virtual nodes that turn a lumpy ring into an even key distribution, then the replication walk that places copies on the next distinct nodes and the bounded-load variant that tames hot partitions, and finally how the real systems — Amazon's Dynamo, Apache Cassandra, and Redis Cluster — actually implement partitioning and sharding at scale. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.
When you want hands-on reps immediately after reading, drill the hash table practice library →, sharpen the fundamentals on the data structures practice library →, and connect it to storage on the database practice library →.
On this page
- Why modulo hashing fails on resize
- The hash ring
- Virtual nodes and load balancing
- Replication and bounded loads
- In practice — Dynamo, Cassandra, Redis Cluster
- Cheat sheet — consistent hashing recipes
- Frequently asked questions
- Practice on PipeCode
1. Why modulo hashing fails on resize
node = hash(key) % N is the partitioning scheme everyone reaches for first — and it moves almost every key the instant N changes
The one-sentence invariant: modulo hashing assigns a key to a node by taking the remainder of the key's hash divided by the node count, which distributes keys evenly while the count is fixed but reshuffles a fraction of roughly (N-1)/N of all keys the moment the count changes — turning a routine capacity change into a cluster-wide data migration and, for a cache, a near-total miss storm. The scheme is not wrong because it distributes badly; with a decent hash function hash(key) % N spreads keys almost perfectly across N buckets. It is wrong because the bucket a key lands in depends on N itself, so the assignment function changes shape whenever you scale — and scaling is exactly the operation a distributed system exists to support.
The three properties any partitioning scheme must have.
-
Balance. Keys should spread evenly across nodes so no single machine holds a disproportionate share of the data or the traffic. Modulo hashing nails this while
Nis constant — a good hash makes every remainder equally likely. -
Determinism. Any client, given only the key and the current membership, must compute the same owner without coordination. Modulo is perfectly deterministic —
hash(key) % Nneeds no shared state beyondN. -
Stability under membership change. When a node joins or leaves, the number of keys that change owner should be proportional to the capacity that changed — ideally about
K/Nkeys for one node out ofN. This is the property modulo hashing catastrophically lacks, and it is the entire reasonconsistent hashingwas invented.
The resize catastrophe — why N → N+1 is a rebalancing storm.
-
The math. A key keeps its owner across a resize only if
hash(key) % N == hash(key) % N+1. For independent, uniform hashes those two remainders agree for only a small set of residues — going from 4 nodes to 5, exactly the hashes whose value mod 20 falls in{0,1,2,3}stay put, which is 4 of 20, or 20%. The other 80% of keys move. -
The general case. For a jump from
NtoN+1, the expected fraction of keys that stay is roughly1/(N+1)in the worst intuition and never better than a small constant; in practice you should assume that any change toNre-partitions the vast majority of your keyspace. - The blast radius for a cache. Every moved key is a guaranteed cache miss on the new owner. Add one node to a hot Memcached tier and you convert ~80% of reads into origin-database hits simultaneously — a thundering herd that can take down the very database the cache was protecting.
- The blast radius for a store. For a stateful store (a sharded SQL tier, a key-value database), a moved key is not just a miss — it is data on the wrong machine. Resizing means physically copying most of your dataset between nodes while serving traffic, which is slow, risky, and often requires a maintenance window.
The 2026 reality — modulo still ships, but only where membership never changes.
-
Where it survives. Fixed-size partition maps — "always exactly 256 partitions" — where you shard by
hash(key) % 256and then map partitions to physical nodes through a separate lookup table. HereN(the partition count) is constant forever; you scale by reassigning partitions to machines, not by changing the modulus. Redis Cluster's 16384 slots are exactly this trick. -
Where it kills you. Any design where the modulus is the live server count. "We hash by
% number_of_cache_boxes" is the anti-pattern; the first autoscaling event re-partitions the fleet. -
What interviewers listen for. Can you quantify the remap fraction (~80% for 4→5) rather than hand-wave "a lot move"? Do you name the failure mode (cache stampede / migration storm)? Do you immediately reach for
consistent hashingor a fixed partition map as the fix? Naming the number is the senior signal.
Worked example — measuring the modulo remap fraction
Detailed explanation. The fastest way to internalise why modulo hashing fails is to measure it. Hash a million keys into N buckets, then re-hash the same keys into N+1 buckets, and count how many changed bucket. The number is shockingly large and it is the single most persuasive artifact you can put on a whiteboard.
-
Setup. One million synthetic keys, a stable hash (MD5 → integer), buckets
N = 4thenN = 5. - Measurement. Count keys whose bucket differs between the two runs.
-
Expectation. Around 80% of keys move — matching the
4/20residue argument above.
Question. Empirically measure the fraction of keys that change owner when a modulo-hashed cluster grows from 4 nodes to 5.
Input.
| Parameter | Value |
|---|---|
| Keys | 1,000,000 synthetic (user:0 … user:999999) |
| Hash | MD5 hex → int |
| Before | hash % 4 |
| After | hash % 5 |
| Metric | fraction where before != after |
Code.
import hashlib
def node_for(key: str, n: int) -> int:
"""Modulo hashing: which of n nodes owns this key."""
h = int(hashlib.md5(key.encode()).hexdigest(), 16)
return h % n
keys = [f"user:{i}" for i in range(1_000_000)]
before = {k: node_for(k, 4) for k in keys} # 4-node cluster
after = {k: node_for(k, 5) for k in keys} # add one node -> 5
moved = sum(1 for k in keys if before[k] != after[k])
print(f"keys moved: {moved:,} ({moved / len(keys):.1%})")
# keys moved: 800,3xx (80.0%)
Step-by-step explanation.
-
node_foris the entire modulo scheme: hash the key to a big integer, take it modulo the node count. With MD5 the hashes are uniform, so each of thenremainders is equally likely — balance is fine. -
beforerecords every key's owner in the 4-node world;afterrecords the owner after a single node is added, making it a 5-node world. Nothing about the keys changed — only the modulus. - The comparison counts keys whose owner differs. Because
hash % 4andhash % 5are effectively independent for a uniform hash, agreement happens only for the residues where they coincide — 4 out of every 20, or 20%. - The printed result is ~80%. Adding one node to a four-node cluster relocates four fifths of the keyspace. The fraction does not improve much at larger
N: 100 → 101 still moves ~99% of keys, becausehash % 100andhash % 101share almost no structure. - For a cache, those 800,000 moved keys are 800,000 guaranteed misses the instant you deploy the new node — the mechanical origin of the cache-stampede outage.
Output.
| Transition | Keys moved | Fraction | Cache effect |
|---|---|---|---|
| 4 → 5 nodes | ~800,000 / 1,000,000 | ~80% | 80% miss storm |
| 5 → 4 nodes | ~800,000 / 1,000,000 | ~80% | 80% miss storm |
| 100 → 101 nodes | ~990,000 / 1,000,000 | ~99% | near-total miss storm |
Rule of thumb. Never let the number of live servers be the modulus. If you can whiteboard the "~80% of keys move on 4→5" number in ten seconds, you have already justified reaching for consistent hashing or a fixed partition map.
Worked example — the cache-stampede failure that follows a resize
Detailed explanation. The remap fraction is abstract until you trace what it does to a production cache. Walk through the sequence of events when an on-call engineer adds a Memcached node to a % N tier during a traffic spike — the exact incident consistent hashing was designed to prevent.
-
The tier. 4 Memcached nodes,
owner = hash(key) % 4, serving 200k reads/sec at a 98% hit rate. The 2% of misses (4k/sec) hit the origin database, which is comfortably sized for that. - The change. Add a fifth node to relieve memory pressure. The modulus becomes 5.
- The consequence. 80% of keys now route to a different node than the one holding their cached value — and those nodes have never seen the key.
Question. Trace the origin-database load in the 60 seconds after a fifth node is added to a modulo-hashed cache tier.
Input.
| Parameter | Value |
|---|---|
| Read rate | 200,000 reads/sec |
| Steady-state hit rate | 98% |
| Steady-state origin load | 4,000 reads/sec |
| Keys remapped on 4→5 | ~80% |
Code.
# Illustrative model of origin load right after the resize
READS_PER_SEC = 200_000
STEADY_HIT_RATE = 0.98
REMAPPED_FRACTION = 0.80 # 4 -> 5 nodes
# Right after resize: remapped keys are guaranteed misses on their new owner.
# Non-remapped keys still hit at the steady rate.
miss_rate_after = REMAPPED_FRACTION * 1.0 + (1 - REMAPPED_FRACTION) * (1 - STEADY_HIT_RATE)
origin_after = READS_PER_SEC * miss_rate_after
print(f"steady origin load : {READS_PER_SEC * (1 - STEADY_HIT_RATE):,.0f} reads/sec")
print(f"post-resize origin : {origin_after:,.0f} reads/sec")
print(f"spike factor : {origin_after / (READS_PER_SEC * (1 - STEADY_HIT_RATE)):.0f}x")
# steady origin load : 4,000 reads/sec
# post-resize origin : 160,400 reads/sec
# spike factor : 40x
Step-by-step explanation.
- In steady state only 2% of reads miss the cache, so the origin database sees 4,000 reads/sec — a load it was provisioned for with headroom.
- The instant the modulus changes from 4 to 5, ~80% of keys hash to a node that has never cached them. Every read for those keys misses, regardless of how hot the key is.
- The blended miss rate jumps to
0.80 * 100% + 0.20 * 2%≈ 80.4%, so origin load leaps to ~160,000 reads/sec — a 40× spike landing all at once. - The origin database, sized for 4k/sec, saturates; latency climbs; the application's own retries add more load; the cache nodes fill slowly because the database can't answer fast enough to repopulate them. This is the thundering herd.
- The fix is not "add capacity to the origin" — it is to stop remapping 80% of keys on every resize. Consistent hashing caps the remap at ~
1/5of keys for a 4→5 change, turning a 40× spike into a bounded, survivable bump.
Output.
| Phase | Miss rate | Origin load | Status |
|---|---|---|---|
| Steady state | 2% | 4,000/sec | healthy |
| Immediately post-resize (modulo) | ~80% | ~160,000/sec | 40× spike; likely outage |
| Immediately post-resize (consistent hashing) | ~22% | ~44,000/sec | bounded; recoverable |
Rule of thumb. A cache's blast radius on resize is directly proportional to the remap fraction. Choose a partitioning scheme by asking "how many keys move when membership changes?" before you ask anything about raw throughput.
Worked example — SQL sharding by modulo and the migration it forces
Detailed explanation. Modulo hashing is just as common — and just as dangerous — in stateful stores. A team shards a users table across 8 Postgres instances by user_id % 8. It works for a year, then they need a ninth shard. Walk through why that is a full-dataset migration, not a config change.
-
The scheme.
shard = user_id % 8; the application routes each query to the computed shard. - The growth. Storage on the 8 shards is full; the team provisions a ninth.
- The problem. Changing the modulus to 9 relocates ~89% of rows — they must be physically moved between databases while the system stays online.
Question. Quantify the data movement when a modulo-sharded 8-instance SQL tier grows to 9 instances, and describe the safe migration.
Input.
| Parameter | Value |
|---|---|
| Rows | 800,000,000 users |
| Before | user_id % 8 |
| After | user_id % 9 |
| Row size | ~1 KB |
Code.
-- The routing rule the application hard-codes (the anti-pattern)
-- shard_id = user_id % 8 -> changes to user_id % 9 on resize
-- Estimate rows that change shard on 8 -> 9 (run per current shard)
WITH sampled AS (
SELECT user_id,
(user_id % 8) AS old_shard,
(user_id % 9) AS new_shard
FROM users
)
SELECT count(*) AS total_rows,
count(*) FILTER (WHERE old_shard <> new_shard) AS rows_moving,
round(100.0 * count(*) FILTER (WHERE old_shard <> new_shard)
/ count(*), 1) AS pct_moving
FROM sampled;
-- total_rows | rows_moving | pct_moving
-- 800000000 | 711111111 | 88.9
Step-by-step explanation.
- The routing rule
user_id % 8lives in application code, so every read and write already assumes 8 shards. Changing to% 9is a code change and a data change — both must land atomically or queries route to the wrong shard. - The SQL counts how many rows would compute a different shard under the new modulus. As with the cache case,
x % 8andx % 9rarely agree, so ~88.9% of rows must relocate. - Moving 711 million rows of ~1 KB each is roughly 711 GB of cross-instance transfer, executed while the tier serves live traffic. During the move, a row may exist on both its old and new shard, so the application needs dual-read / dual-write logic to stay correct.
- The migration typically runs as: stand up shard 9, dual-write new data under the
% 9rule, backfill the 711M relocating rows in batches, verify, then cut reads over. It is weeks of work and a standing outage risk — all to add one shard. - Had the tier used
consistent hashing(or a fixed 256-partition map decoupled from instance count), adding a ninth instance would move only ~1/9of rows (~89 GB) and require no change to the routing rule — just a membership update.
Output.
| Approach | Rows moved (8→9) | Bytes moved | Routing-code change? |
|---|---|---|---|
Modulo (% instance_count) |
~711M (88.9%) | ~711 GB | yes |
| Consistent hashing | ~89M (~11%) | ~89 GB | no |
| Fixed 256-partition map | 0 keys re-hashed; ~1/9 partitions reassigned | ~89 GB | no |
Rule of thumb. In a stateful store, the remap fraction is data you have to physically copy. Decouple the hash modulus from the live instance count — use consistent hashing or a fixed partition map — so scaling copies a proportional slice, not the whole dataset.
System design interview question on partitioning
A senior interviewer often opens with: "You are running a 12-node Memcached tier sharded by hash(key) % 12, serving 500k reads/sec at a 97% hit rate. Traffic is growing, so you need to add 4 nodes. Walk me through exactly what happens to your keys and your origin database the moment you deploy, quantify it, and tell me what you would do instead."
Solution Using consistent hashing instead of modulo
# Baseline: modulo hashing. Show the damage, then contrast with a ring.
import hashlib
def h(s: str) -> int:
return int(hashlib.md5(s.encode()).hexdigest(), 16)
def modulo_owner(key: str, n: int) -> int:
return h(key) % n
keys = [f"key:{i}" for i in range(1_000_000)]
# --- Modulo: 12 -> 16 nodes ---
mod_before = {k: modulo_owner(k, 12) for k in keys}
mod_after = {k: modulo_owner(k, 16) for k in keys}
mod_moved = sum(1 for k in keys if mod_before[k] != mod_after[k])
# --- Consistent hashing preview: a ring of node tokens ---
import bisect
class Ring:
def __init__(self, nodes):
self.ring = [] # sorted (position, node) by position
self._pos = [] # parallel sorted positions for bisect
for node in nodes:
self.add(node)
def add(self, node):
p = h(node) % (2**32)
bisect.insort(self.ring, (p, node))
self._pos = [p for p, _ in self.ring]
def owner(self, key):
p = h(key) % (2**32)
i = bisect.bisect(self._pos, p) % len(self.ring) # first node clockwise
return self.ring[i][1]
ring_before = Ring([f"node-{i}" for i in range(12)])
own_before = {k: ring_before.owner(k) for k in keys}
ring_after = Ring([f"node-{i}" for i in range(16)]) # add 4 nodes
own_after = {k: ring_after.owner(k) for k in keys}
ring_moved = sum(1 for k in keys if own_before[k] != own_after[k])
print(f"modulo 12->16 moved: {mod_moved/len(keys):.1%}") # ~92.x%
print(f"ring 12->16 moved: {ring_moved/len(keys):.1%}") # ~25%
Step-by-step trace.
| Step | Modulo path | Consistent-hashing path |
|---|---|---|
| Owner formula | hash(key) % N |
first node clockwise on a 2^32 ring |
| Add 4 nodes (12→16) | modulus 12 → 16 | drop 4 new tokens onto the ring |
| Keys that move | ~92% (almost all) | ~25% (the four new arcs only) |
| Origin miss spike | ~92% of reads | ~25% of reads |
| Data migration (if stateful) | ~92% of dataset | ~25% of dataset |
| Routing change | new modulus everywhere | membership update only |
After the change, the modulo tier remaps roughly 92% of the million keys — a near-total miss storm — while the ring remaps only about 25%, the keys that fall on the four arcs the new tokens carved out. The origin database sees a 4× bump instead of a 30× one, and it recovers.
Output:
| Metric | Modulo 12→16 | Consistent hashing 12→16 |
|---|---|---|
| Fraction of keys moved | ~92% | ~25% |
| Origin load multiplier | ~30× | ~4× |
| Survivable without an outage? | usually no | usually yes |
| Ideal fraction for +4 of 12 | — | ~4/16 = 25% |
Why this works — concept by concept:
-
Modulo couples ownership to N — because the owner is
hash(key) % N, every key's owner is a function of the total node count. Change the count and you change the function for essentially every key. The scheme has no memory of where a key used to live. -
The ring decouples ownership from N — each key hashes to a fixed point on a
2^32circle and is owned by the next node clockwise. Adding a node inserts one new point; it can only steal the keys in the arc between it and its predecessor. Every other key's owner is unchanged. -
Bounded remap = bounded blast radius — moving ~
k/(N+k)of keys when you addknodes toNmeans the cache-miss spike and the data migration are proportional to the capacity you added, not to the whole cluster. That is the property that makes scaling routine. -
Cost — modulo lookup is
O(1)but resize isO(all keys). Ring lookup isO(log N)per key via binary search over the sorted token positions, and resize isO(K/N)keys moved plusO(N)to rebuild the position index. You trade a hair of lookup cost for a scheme that survives membership change — always the right trade in a system that scales.
Hash Table
Topic — hash-table
Hash-table and hashing problems
2. The hash ring
Map keys and nodes onto one circle, and ownership becomes "the next node clockwise" — so adding a node only disturbs a single arc
The one-sentence invariant: the hash ring places both keys and nodes onto the same circular hash space — usually the integers [0, 2^32) or [0, 2^160) wrapped end-to-end — and defines a key's owner as the first node encountered walking clockwise from the key's position, so inserting or removing a node changes ownership only for the keys in the arc between that node and its clockwise predecessor, bounding the movement to about K/N keys. This is the core idea introduced by Karger and colleagues in 1997 for web caching, and it is the mechanism underneath every system in section 5. Once you see the circle, the modulo catastrophe looks obviously avoidable: the whole problem was that ownership depended on N; on the ring it depends only on positions, and adding a node adds exactly one position.
The construction — three steps to a working ring.
-
Hash the nodes onto the circle. For each node, compute
hash(node_id) mod 2^32and record that integer as the node's token — its position on the ring. Keep the tokens in a sorted structure. -
Hash the keys onto the same circle. A key's position is
hash(key) mod 2^32, in the identical space. Keys and nodes now live on one number line that wraps from2^32 - 1back to0. - Own by successor. The owner of a key is the first node token at or after the key's position, wrapping past the top of the ring back to the smallest token if necessary. This "first node clockwise" rule is a successor search — a binary search over the sorted tokens.
Why membership change is cheap.
- Adding a node. A new token drops in at some position. The only keys that change owner are those between the new token and the previous token clockwise-behind it — that arc used to belong to the next node clockwise and now belongs to the newcomer. Every other arc is untouched.
- Removing a node. The departing node's arc merges into its clockwise successor; only that node's former keys move, and they all move to exactly one place. No other key is affected.
-
The expected fraction. With
Nroughly-evenly-spaced tokens, one node owns about1/Nof the circle, so adding or removing a node moves aboutK/Nkeys — proportional to the one unit of capacity you changed, which is the whole point.
The data structure — a sorted token array plus binary search.
-
Store. A sorted list of
(position, node)pairs, or two parallel arrays (positions and nodes) so you can binary-search positions directly. -
Lookup.
bisect(binary search) for the key's position, take the index modulo the token count to wrap, return that node.O(log N)per lookup. -
Mutation. Insert or delete a token and keep the array sorted —
O(N)to rebuild a parallel index, orO(log N)amortised with a balanced tree / sorted container. Membership changes are rare compared to lookups, so an array is usually fine.
The catch a bare ring still has.
-
Uneven arcs. With only a handful of randomly-placed tokens, the arcs between them are far from equal — one node can easily own twice its fair share of the circle, and therefore twice the data and traffic. Randomness at small
Nis lumpy. - Removal doubles a neighbor. When a node leaves, all of its keys pile onto a single successor, momentarily doubling that node's load rather than spreading the departed load across the survivors.
- The fix. Both problems are solved by virtual nodes — giving each physical node many tokens so the law of large numbers smooths the arcs. That is section 3. A bare ring is the right mental model but not the production configuration.
Worked example — build a minimal hash ring
Detailed explanation. A production-shaped ring is about forty lines of Python: hash nodes to tokens, keep them sorted, and resolve a key by binary-searching for its successor token. Build it from scratch and confirm the successor rule with a hand-checkable example.
-
Space.
2^32positions. -
Store. Parallel sorted arrays:
_keys(positions) and_nodes(owners). -
Lookup.
bisectfor the successor, wrap with modulo.
Question. Implement a hash ring supporting add_node, remove_node, and get_node(key) with O(log N) lookup.
Input.
| Operation | Argument |
|---|---|
| add_node | "A", "B", "C" |
| get_node | "photo:42" |
| Ring space | 2^32 |
Code.
import bisect
import hashlib
class HashRing:
def __init__(self):
self._pos: list[int] = [] # sorted token positions
self._node: list[str] = [] # node at the same index
@staticmethod
def _hash(s: str) -> int:
return int(hashlib.md5(s.encode()).hexdigest(), 16) % (2**32)
def add_node(self, node: str) -> None:
p = self._hash(node)
i = bisect.bisect(self._pos, p)
self._pos.insert(i, p)
self._node.insert(i, node)
def remove_node(self, node: str) -> None:
p = self._hash(node)
i = bisect.bisect_left(self._pos, p)
if i < len(self._pos) and self._node[i] == node:
self._pos.pop(i)
self._node.pop(i)
def get_node(self, key: str) -> str:
if not self._pos:
raise ValueError("empty ring")
p = self._hash(key)
i = bisect.bisect(self._pos, p) % len(self._pos) # first token clockwise
return self._node[i]
ring = HashRing()
for n in ("A", "B", "C"):
ring.add_node(n)
print(ring.get_node("photo:42")) # -> whichever node is first clockwise
Step-by-step explanation.
-
_hashfolds any string into[0, 2^32). Nodes and keys use the same function into the same space, which is what makes "first node clockwise" meaningful. -
add_nodefinds the insertion point withbisectand splices the token into both parallel arrays, keeping them sorted so lookups stayO(log N). -
remove_nodelocates the exact token by position and confirms the node identity before removing — guarding against a hash collision removing the wrong entry. -
get_nodebinary-searches for the first token strictly after the key's position. If the key hashes past the largest token,bisectreturnslen(self._pos), and% lenwraps the index back to token0— that wrap is the "circle" made concrete. - Because ownership is a pure function of token positions, calling
get_nodeagain after adding a fourth node changes the answer only for keys that fall on the new node's arc — the property section 1 was missing.
Output.
| Key | Position (mod 2^32) | First token clockwise | Owner |
|---|---|---|---|
| photo:42 | 0x4c… | token for "C" | C |
| photo:7 | 0x9a… | token for "A" | A |
| photo:99 | 0xf1… (past all) | wraps to smallest | (wrapped node) |
Rule of thumb. Keep nodes and keys in one hash space and resolve ownership with a single binary search for the successor token. The wrap-around modulo on the index is the one line beginners forget — and the one that makes it a ring instead of a line.
Worked example — prove only K/N keys move when a node joins
Detailed explanation. The value of the ring is entirely in its stability, so measure it. Put 100,000 keys on a 4-node ring, record owners, add a fifth node, and count movers. The number should hover near 1/5 — the arc the newcomer steals — not the 80% modulo inflicted.
- Before. 4 nodes, 100,000 keys, record owners.
- After. Add node "E", re-resolve, count changes.
- Expectation. ~20% move, and every mover goes to the new node.
Question. Empirically confirm that adding one node to a 4-node ring moves about 1/5 of keys, all onto the new node.
Input.
| Parameter | Value |
|---|---|
| Nodes before | A, B, C, D |
| Node added | E |
| Keys | 100,000 |
Code.
ring = HashRing()
for n in ("A", "B", "C", "D"):
ring.add_node(n)
keys = [f"obj:{i}" for i in range(100_000)]
before = {k: ring.get_node(k) for k in keys}
ring.add_node("E")
after = {k: ring.get_node(k) for k in keys}
moved = [k for k in keys if before[k] != after[k]]
to_new = [k for k in moved if after[k] == "E"]
print(f"moved : {len(moved)/len(keys):.1%}") # ~20%
print(f"to node E: {len(to_new)}/{len(moved)}") # all of them
Step-by-step explanation.
- The 4-node ring gives each node roughly a quarter of the circle (subject to token randomness), so each owns ~25,000 of the 100,000 keys.
-
beforesnapshots every key's owner. Adding "E" inserts exactly one new token; the ring's other three tokens do not move. - "E" captures the arc between its token and the previous token clockwise-behind it — an arc that previously belonged to whichever node was next clockwise. That arc is about
1/5of the circle once there are five tokens. - Re-resolving shows ~20% of keys changed owner, and the check confirms every single mover now resolves to "E". No key moves between the old nodes — the ring never disturbs an arc it did not touch.
- Contrast with modulo, where adding a node reshuffles ~80%. The ring's movement is bounded, one-directional (onto the newcomer), and proportional to the capacity added.
Output.
| Metric | Value |
|---|---|
| Keys moved | ~20,000 (~20%) |
| Movers going to E | 100% |
| Keys moved between old nodes | 0 |
| Ideal fraction (1 of 5) | 20% |
Rule of thumb. A correct ring moves only keys onto (or off) the node whose membership changed — never between untouched nodes. If your measurements show cross-node churn on a simple add, your successor logic or wrap-around is wrong.
Worked example — the lumpy-arc problem on a bare ring
Detailed explanation. Before celebrating, measure the balance of a bare ring. With one token per node, arc sizes are random and uneven, so some nodes own far more than their fair share. Quantifying the imbalance is what motivates virtual nodes in the next section.
- Setup. 8 nodes, one token each, 1,000,000 keys.
-
Measure. Per-node key count; compare max to the fair share (
1/8 = 12.5%). - Expectation. The busiest node holds well over its share — often 1.5–2× the fair amount.
Question. Measure the load imbalance of an 8-node bare ring and express the hottest node as a multiple of its fair share.
Input.
| Parameter | Value |
|---|---|
| Nodes | 8, one token each |
| Keys | 1,000,000 |
| Fair share | 125,000 (12.5%) |
Code.
from collections import Counter
ring = HashRing()
for i in range(8):
ring.add_node(f"node-{i}")
keys = [f"k:{i}" for i in range(1_000_000)]
load = Counter(ring.get_node(k) for k in keys)
fair = 1_000_000 / 8
hi = max(load.values())
lo = min(load.values())
print(f"hottest: {hi:,} ({hi/fair:.2f}x fair)")
print(f"coldest: {lo:,} ({lo/fair:.2f}x fair)")
# hottest: ~2xx,xxx (1.6-2.0x fair)
# coldest: ~ xx,xxx (0.3-0.6x fair)
Step-by-step explanation.
- Eight random tokens carve the circle into eight arcs of random length. Arc length maps directly to key count, so uneven arcs mean uneven load.
- Counting owners across a million keys reveals the spread. It is common for the hottest node to own 1.6–2.0× its fair share and the coldest to own half — a 3–4× ratio between busiest and idlest.
- That imbalance is a real cost: the hot node needs more memory and CPU, hits capacity first, and becomes the tail-latency bottleneck for the whole tier.
- Worse, removing a node dumps its entire arc onto one successor, transiently doubling that node — the ring gives you cheap movement but not, by itself, even distribution.
- The cure is many small tokens per node: with
Vtokens each, the arcs a node owns average out and the imbalance shrinks like1/sqrt(V). That is exactly virtual nodes, next.
Output.
| Configuration | Hottest node | Coldest node | Max/min ratio |
|---|---|---|---|
| 8 nodes, 1 token each | ~1.8× fair | ~0.45× fair | ~4× |
| 8 nodes, 150 tokens each | ~1.05× fair | ~0.95× fair | ~1.1× |
Rule of thumb. A bare ring solves movement but not balance. Never ship one token per node in production — measure the max-to-fair ratio, and if it exceeds ~1.1×, add virtual nodes until it doesn't.
System design interview question on the hash ring
A senior interviewer might ask: "Implement consistent hashing for a distributed cache. Support adding and removing nodes and looking up the owner of a key. Then tell me the time complexity of each operation, prove that adding a node moves only about 1/N of keys, and identify the one problem your basic ring still has."
Solution Using a sorted-token ring with binary-search successor lookup
import bisect
import hashlib
class ConsistentHashRing:
"""Bare ring: one token per node. O(log N) lookup, O(K/N) keys move on resize."""
def __init__(self, nodes: list[str] | None = None):
self._pos: list[int] = []
self._node: list[str] = []
for n in nodes or []:
self.add_node(n)
@staticmethod
def _hash(s: str) -> int:
return int(hashlib.sha1(s.encode()).hexdigest(), 16) % (2**32)
def add_node(self, node: str) -> None:
p = self._hash(node)
i = bisect.bisect(self._pos, p)
self._pos.insert(i, p) # keep positions sorted
self._node.insert(i, node)
def remove_node(self, node: str) -> None:
p = self._hash(node)
i = bisect.bisect_left(self._pos, p)
if i < len(self._pos) and self._node[i] == node:
self._pos.pop(i)
self._node.pop(i)
def get_node(self, key: str) -> str:
if not self._pos:
raise ValueError("ring is empty")
p = self._hash(key)
i = bisect.bisect(self._pos, p) % len(self._pos) # successor, wrapped
return self._node[i]
Step-by-step trace.
| Input | Step | Result |
|---|---|---|
add_node("A") |
hash A → 0x21…, bisect-insert | ring = [A] |
add_node("B") |
hash B → 0xC4…, bisect-insert | ring = A, B |
add_node("C") |
hash C → 0x88…, bisect-insert | ring = [A, C, B] |
get_node("user:5") |
pos 0x53…, bisect → index of C | owner = C |
get_node("user:9") |
pos 0xF0… past all, % len wraps |
owner = A |
remove_node("C") |
pop C's token | C's arc merges into B |
Walking the trace: three nodes hash to three positions and sort by position (not insertion order). A key resolves by binary-searching for the first token at or after it; user:9 hashes past the largest token so the index wraps to the smallest, node A — the circle in action. Removing C hands its whole arc to B, the tell-tale sign that a bare ring needs virtual nodes for balance.
Output:
| Operation | Complexity | Keys moved |
|---|---|---|
get_node |
O(log N) | — |
add_node |
O(N) index shift | ~K/N onto the new node |
remove_node |
O(N) index shift | departed node's ~K/N onto one successor |
| Rebalance on +1 of N | — | ~1/(N+1) of keys |
Why this works — concept by concept:
-
Successor lookup on a circle — defining ownership as "first token clockwise" means a key's owner is determined by position, not by node count.
bisectfinds that successor inO(log N); the modulo on the index turns the sorted line into a wrap-around ring. -
Sorted token array — keeping positions sorted lets lookups binary-search and lets inserts/deletes touch only the local neighborhood conceptually (an array pays
O(N)to shift, a tree paysO(log N)). Lookups vastly outnumber membership changes, so the simple array is usually the right call. -
Bounded movement — a new token can only capture the arc between itself and its predecessor, so adding a node moves ~
1/(N+1)of keys and touches no other node's arc. This is the exact property modulo lacked and the reason the ring scales. - One remaining flaw — one token per node gives random, uneven arcs (load imbalance) and dumps a removed node's whole arc onto a single successor. The fix is virtual nodes, which the bare ring sets up but does not provide.
-
Cost —
O(log N)lookup,O(N)membership mutation for the array form, andO(K/N)keys relocated per node change. Compared with modulo'sO(all keys)resize, the ring converts scaling from a full migration into a proportional, survivable one.
Data Structures
Topic — data-structures
Design a data structure with binary search
3. Virtual nodes and load balancing
Give each machine many small tokens instead of one, and random lumpiness averages out into an even key distribution
The one-sentence invariant: virtual nodes (vnodes) place each physical node on the ring at many positions — typically dozens to a few hundred tokens per node — so that the arcs a node owns are the sum of many independent random pieces, which by the law of large numbers evens out the key distribution (load variance shrinks roughly like 1/sqrt(V) for V tokens per node) and spreads a departed node's keys across many successors instead of dumping them all on one. Virtual nodes are not a different algorithm; they are the ring from section 2 with each node hashed under several distinct labels. This one change turns the lumpy, 4×-imbalanced bare ring into the smooth, near-uniform distribution real systems depend on — and it is why Dynamo, Cassandra, and every serious library default to many tokens per node.
How virtual nodes are constructed.
-
Many labels per node. For physical node
AandVvirtual nodes, hashA#0,A#1, …A#(V-1)onto the ring —Vseparate tokens that all map back toA. A lookup that lands on any of them resolves toA. - A token-to-node map. Alongside the sorted positions, keep a map from each token position to its owning physical node. Successor lookup is unchanged; you just dereference the token to its physical node at the end.
-
Interleaving. Because the
Vtokens per node are scattered independently, tokens from different physical nodes interleave all around the ring. Every node ends up owning many small arcs distributed across the circle rather than one big contiguous slice.
Why this balances load.
-
The averaging argument. With one token, a node's load is a single random arc — high variance. With
Vtokens, its load is the sum ofVindependent arcs; the sum concentrates around the mean, and the relative standard deviation falls like1/sqrt(V). A hundred-ish tokens per node typically brings the busiest node within a few percent of fair share. -
Graceful removal. When a node leaves, each of its
Varcs merges into a different successor, so its load is redistributed across many nodes — no single successor doubles. Symmetrically, a joining node steals many small arcs from many nodes, so no single donor is drained. -
Graceful addition. A new node with
Vtokens picks upVsmall arcs from around the ring, immediately taking its fair share without a lopsided handoff.
Heterogeneous capacity — weighting by token count.
-
The lever. Token count is the capacity knob. A machine with twice the memory or CPU gets twice the tokens and therefore owns about twice the keyspace. Consistent hashing supports weighted nodes for free — just scale
Vper node. -
The practice. Cassandra exposes this as
num_tokens; give beefier hosts a larger value. Client libraries likeketamalet you assign weights that translate into proportional token counts. -
The caution. Weighting is coarse at low token counts. To make a node reliably own 1.5× its neighbors, you need enough tokens that the arcs average out — another reason
Vis in the hundreds, not the single digits.
Choosing V — the trade-off.
-
Higher V. Smoother load, gentler rebalancing, finer capacity weighting — but a larger sorted token array (
N × Ventries), slightly slower membership updates, and more metadata to gossip around the cluster. - Lower V. Less memory and metadata, faster updates — but lumpier load and coarser weighting.
-
Typical values. Libraries use 100–200 replicas (tokens) per node; Cassandra historically defaulted
num_tokensto 256, later lowering the recommendation to 16 with an improved allocation algorithm. The right number balances the load-smoothness you need against the token-array size you can afford.
Worked example — add virtual nodes and measure the smoothing
Detailed explanation. Extend the ring so each physical node contributes V tokens, then measure how imbalance falls as V grows. This is the single most convincing demonstration that vnodes matter — the max-to-fair ratio drops visibly with every increase in V.
-
Change. Hash
node#0 … node#(V-1)for each node; map tokens back to the physical node. -
Measure. Hottest node as a multiple of fair share, for
Vin{1, 10, 50, 200}. -
Expectation. Imbalance shrinks roughly like
1/sqrt(V).
Question. Implement a virtual-node ring and show how the hottest node's load approaches fair share as V increases.
Input.
| Parameter | Value |
|---|---|
| Physical nodes | 8 |
| Keys | 1,000,000 |
| V values | 1, 10, 50, 200 |
Code.
import bisect
import hashlib
from collections import Counter
class VNodeRing:
def __init__(self, vnodes: int = 150):
self.v = vnodes
self._pos: list[int] = []
self._node: list[str] = []
@staticmethod
def _hash(s: str) -> int:
return int(hashlib.md5(s.encode()).hexdigest(), 16) % (2**32)
def add_node(self, node: str) -> None:
for r in range(self.v):
p = self._hash(f"{node}#{r}") # V distinct tokens per node
i = bisect.bisect(self._pos, p)
self._pos.insert(i, p)
self._node.insert(i, node) # token -> physical node
def get_node(self, key: str) -> str:
p = self._hash(key)
i = bisect.bisect(self._pos, p) % len(self._pos)
return self._node[i]
keys = [f"k:{i}" for i in range(1_000_000)]
for v in (1, 10, 50, 200):
ring = VNodeRing(vnodes=v)
for i in range(8):
ring.add_node(f"node-{i}")
load = Counter(ring.get_node(k) for k in keys)
fair = 1_000_000 / 8
print(f"V={v:>3}: hottest {max(load.values())/fair:.2f}x fair")
# V= 1: hottest ~1.80x fair
# V= 10: hottest ~1.28x fair
# V= 50: hottest ~1.12x fair
# V=200: hottest ~1.05x fair
Step-by-step explanation.
-
add_nodenow insertsVtokens per physical node, each hashed from a distinct labelnode#r. AllVtokens map back to the same physical node via the parallel_nodearray. -
get_nodeis unchanged — it finds the successor token and returns its physical node. The vnode machinery is invisible at lookup time; only the token density changed. - At
V=1the ring is the bare ring from section 2, and the hottest node sits near 1.8× fair — the lumpiness we measured before. - As
Vrises, each node's load is the sum of more independent arcs, so the distribution tightens. ByV=50the hottest node is within ~12% of fair; byV=200within ~5%. - The improvement tracks
1/sqrt(V): quadruplingVroughly halves the excess imbalance. Beyond a couple hundred tokens the returns diminish and the token array's memory cost starts to matter.
Output.
| V (tokens/node) | Hottest node | Excess over fair | Token array size |
|---|---|---|---|
| 1 | ~1.80× | ~80% | 8 |
| 10 | ~1.28× | ~28% | 80 |
| 50 | ~1.12× | ~12% | 400 |
| 200 | ~1.05× | ~5% | 1,600 |
Rule of thumb. Pick V so the hottest node stays within ~5–10% of fair share — usually 100–200 tokens per node. Remember imbalance falls like 1/sqrt(V), so doubling smoothness costs 4× the tokens.
Worked example — graceful rebalancing on node removal
Detailed explanation. The second gift of vnodes is that removing a node spreads its load across many survivors instead of one. Measure where a removed node's keys go, with and without virtual nodes, to see the difference.
-
Setup. 6 nodes, compare
V=1andV=150. - Action. Remove one node; track which survivors absorb its keys.
-
Expectation.
V=1dumps everything on one successor;V=150spreads it near-evenly across all five.
Question. When a node is removed, how is its load redistributed with one token per node versus 150 tokens per node?
Input.
| Parameter | Value |
|---|---|
| Nodes | 6 |
| Removed | node-3 |
| Keys | 600,000 |
| V compared | 1 vs 150 |
Code.
def redistribution(v: int) -> dict[str, float]:
ring = VNodeRing(vnodes=v)
for i in range(6):
ring.add_node(f"node-{i}")
keys = [f"k:{i}" for i in range(600_000)]
before = {k: ring.get_node(k) for k in keys}
# remove node-3 by rebuilding without it
ring2 = VNodeRing(vnodes=v)
for i in range(6):
if i != 3:
ring2.add_node(f"node-{i}")
after = {k: ring2.get_node(k) for k in keys}
orphans = [k for k in keys if before[k] == "node-3"]
dest = Counter(after[k] for k in orphans)
total = len(orphans)
return {n: c / total for n, c in dest.items()}
print("V=1 :", {k: round(v, 2) for k, v in redistribution(1).items()})
# V=1 : {'node-4': 1.0} # all keys to a single successor
print("V=150:", {k: round(v, 2) for k, v in redistribution(150).items()})
# V=150: {'node-0': 0.21, 'node-1': 0.19, 'node-2': 0.20, 'node-4': 0.20, 'node-5': 0.20}
Step-by-step explanation.
-
orphansare the keys that node-3 owned before removal — the load that must go somewhere. - With
V=1, node-3 had one token owning one contiguous arc; removing it merges that entire arc into the single next token clockwise. One survivor inherits 100% of node-3's load and momentarily runs at ~2× capacity. - With
V=150, node-3 had 150 small tokens scattered around the ring; each arc merges into a different neighbor. The 150 handoffs land on all five survivors roughly equally (~20% each). - Even redistribution means no survivor spikes — the cluster degrades from 6 nodes to 5 smoothly, exactly the behavior you want during a failure or a planned decommission.
- The same mechanism runs in reverse on addition: a joining node's 150 tokens each steal a small arc from a different neighbor, so it reaches fair share without starving any single donor.
Output.
| V | Survivors absorbing load | Max single survivor share | Failure behavior |
|---|---|---|---|
| 1 | 1 of 5 | 100% | one node doubles; hotspot |
| 150 | 5 of 5 | ~21% | smooth; no hotspot |
Rule of thumb. Virtual nodes are what make failures boring. If losing one node overloads exactly one neighbor, you shipped a bare ring — add tokens so a departed node's load fans out across all survivors.
Worked example — weighting a heterogeneous cluster
Detailed explanation. Real fleets mix instance sizes. Consistent hashing handles this by giving bigger machines more tokens. Show a 3-node cluster where one node has double the capacity and confirm it earns double the keys.
- Cluster. node-big (weight 2), node-a (weight 1), node-b (weight 1).
-
Tokens.
V × weighttokens per node. - Expectation. node-big owns ~50% of keys; the others ~25% each.
Question. Configure a weighted ring so a double-capacity node owns twice the keyspace, and verify the split.
Input.
| Node | Weight | Base V | Tokens |
|---|---|---|---|
| node-big | 2 | 150 | 300 |
| node-a | 1 | 150 | 150 |
| node-b | 1 | 150 | 150 |
Code.
class WeightedRing(VNodeRing):
def add_weighted(self, node: str, weight: int) -> None:
for r in range(self.v * weight): # more tokens = more keyspace
p = self._hash(f"{node}#{r}")
i = bisect.bisect(self._pos, p)
self._pos.insert(i, p)
self._node.insert(i, node)
ring = WeightedRing(vnodes=150)
ring.add_weighted("node-big", weight=2)
ring.add_weighted("node-a", weight=1)
ring.add_weighted("node-b", weight=1)
keys = [f"k:{i}" for i in range(1_000_000)]
load = Counter(ring.get_node(k) for k in keys)
for n, c in sorted(load.items()):
print(f"{n:>9}: {c/1_000_000:.1%}")
# node-big : 50.0%
# node-a : 25.0%
# node-b : 25.0%
Step-by-step explanation.
-
add_weightedmultiplies the token count by the node's weight, so node-big drops 300 tokens versus 150 for the others. - Since keyspace ownership is proportional to token count once
Vis large enough to average out, node-big's 300 of 600 total tokens claim ~50% of the ring. - node-a and node-b split the remaining half, ~25% each — exactly proportional to their weight-1 share.
- This makes capacity planning a single integer per node. Replace a node with a bigger box? Raise its weight. Consistent hashing routes proportionally more keys to it with no other change.
- The averaging caveat applies: at low base
Vthe weighted split is noisy. Keep enough tokens that the realized split lands within a couple percent of the intended weights.
Output.
| Node | Intended share | Realized share |
|---|---|---|
| node-big (weight 2) | 50% | ~50.0% |
| node-a (weight 1) | 25% | ~25.0% |
| node-b (weight 1) | 25% | ~25.0% |
Rule of thumb. Express capacity as token count. Weighting a node by giving it proportionally more virtual nodes is the cleanest way to run a heterogeneous fleet — but keep base V high enough that the realized split matches the intended weights.
System design interview question on load balancing with virtual nodes
A senior interviewer might ask: "Your consistent-hashing cache tier is showing one node at 90% CPU while the others idle at 40%, even though the ring should balance them. Diagnose the likely cause, explain how virtual nodes fix it, tell me how many tokens per node you would use and why, and describe what happens to load distribution when that hot node eventually fails."
Solution Using virtual nodes to smooth load and rebalancing
import bisect
import hashlib
from collections import Counter
class BalancedRing:
"""Virtual-node ring. V tokens per node smooths load and rebalancing."""
def __init__(self, vnodes: int = 150):
self.v = vnodes
self._pos: list[int] = []
self._tok: list[str] = [] # physical node for each token
@staticmethod
def _hash(s: str) -> int:
return int(hashlib.sha1(s.encode()).hexdigest(), 16) % (2**32)
def add_node(self, node: str, weight: int = 1) -> None:
for r in range(self.v * weight):
p = self._hash(f"{node}#{r}")
i = bisect.bisect(self._pos, p)
self._pos.insert(i, p)
self._tok.insert(i, node)
def get_node(self, key: str) -> str:
p = self._hash(key)
i = bisect.bisect(self._pos, p) % len(self._pos)
return self._tok[i]
def load_report(self, keys: list[str]) -> dict[str, float]:
c = Counter(self.get_node(k) for k in keys)
total = sum(c.values())
return {n: v / total for n, v in c.items()}
Step-by-step trace.
| Step | Action | Effect |
|---|---|---|
| Diagnose | one token per node | random arcs → one node owns ~1.8× share |
| Fix | set V = 150 tokens/node | each node = sum of 150 arcs → ~1.05× share |
| Verify |
load_report on 1M keys |
hottest node within ~5% of fair |
| Weight | add_node(big, weight=2) |
big node owns proportionally 2× |
| Fail | remove the (former) hot node | 150 arcs fan out across all survivors |
Tracing the fix: the imbalance was almost certainly one-token-per-node (or too few tokens), so a single node drew a lopsided arc. Raising V to 150 makes every node's load the average of 150 independent arcs, collapsing the hottest node from ~1.8× to ~1.05× fair. When that node later fails, its 150 tokens each merge into a different neighbor, so the load redistributes evenly instead of doubling one survivor.
Output:
| Metric | Before (V=1) | After (V=150) |
|---|---|---|
| Hottest node | ~1.8× fair | ~1.05× fair |
| Load std-dev (relative) | high | ~1/sqrt(150) of the V=1 spread |
| Removal behavior | 1 successor doubles | 5+ survivors share evenly |
| Capacity weighting | coarse / noisy | proportional to token count |
Why this works — concept by concept:
-
Many tokens per node — hashing each physical node under
Vdistinct labels scatters it across the ring, so a node owns many small arcs instead of one big random one. Ownership resolution is unchanged; only token density grows. -
Variance falls like 1/sqrt(V) — a node's load becomes the sum of
Vindependent arcs, and sums of independent random variables concentrate. That is the statistical reason ~150 tokens brings the hottest node within a few percent of fair. - Fan-out on membership change — because a node's tokens are spread out, adding or removing it touches many neighbors with small handoffs rather than one neighbor with a huge one. Failures and scale events become smooth.
-
Token count = capacity knob — weighting a node by its token count gives free, proportional support for heterogeneous hardware, exposed in real systems as
num_tokensor per-node weights. -
Cost — the token array grows to
N × Ventries and gossip/metadata scales with it, so lookups stayO(log(N·V))and memory and update cost rise linearly inV. The trade — a few thousand extra sorted integers for near-perfect balance and boring failures — is almost always worth it.
Data Structures
Topic — data-structures
Load-balancing and distribution problems
4. Replication and bounded loads
Store each key on the next several distinct nodes clockwise, and cap any node's share so a hot key can't melt one machine
The one-sentence invariant: replication on a ring stores each key not just on its owner but on the next N-1 distinct physical nodes walking clockwise — the "preference list" — so the data survives node loss, while consistent hashing with bounded loads adds a per-node capacity cap of (1+ε)·average and spills any key that would exceed it to the next node clockwise, guaranteeing no node exceeds its cap even under a skewed, adversarial key distribution. Replication is what makes the ring a durable store rather than a routing table, and bounded loads is the refinement that closes consistent hashing's last real weakness: a single scorching-hot key or a burst of correlated keys can still overwhelm one node even with perfect vnode balance, because balance is about count of keys, not their traffic.
The preference list — replication done right on a ring.
-
The walk. For replication factor
RF, a key's replica set is its owner plus the nextRF-1nodes clockwise. This ordered list is Dynamo's "preference list": the first node is the primary, the rest are backups in ring order. -
Distinct physical nodes. With virtual nodes, the next few tokens clockwise may belong to the same physical machine. The preference-list walk must skip tokens whose physical node is already in the set, so
RF=3means three different machines, not three tokens. -
Rack / zone awareness. Production systems extend the skip rule: don't just require distinct nodes, require distinct failure domains (racks, availability zones). A key with
RF=3should land in three zones so one zone outage can't take all copies. -
Read/write quorums. With
RFreplicas you tune consistency by requiringWacknowledged writes andRreads such thatW + R > RFguarantees a read sees the latest write. This is the Dynamo-style quorum built directly on the preference list.
Why balance-by-count is not enough — the hot-key problem.
- The gap. Virtual nodes balance the number of keys per node, assuming every key gets similar traffic. Real workloads are skewed: a celebrity's profile, a viral post, or a trending product can make one key carry a thousand times the traffic of an average key.
- The failure. That hot key lives on one owner (and its replicas). Perfect count-balance does nothing for it — the owner saturates while the rest of the ring idles. Vnodes solved lumpy arcs, not lumpy traffic.
- The mitigations. Options include key-splitting (append a random suffix to fan a hot key across nodes, at the cost of scatter-gather reads), caching the hot key in front of the store, and — the systematic answer — bounding per-node load so overflow spills to neighbors.
Consistent hashing with bounded loads (CHBL).
-
The cap. Define average load
avg = total_load / Nand a per-node capacitycap = ceil((1+ε)·avg)for a smallε(say 0.25). No node may hold more thancapitems. -
The assignment. Place a key at its ring owner if that node is under
cap; otherwise walk clockwise to the next node with spare capacity. The key "spills forward" until it finds room. -
The guarantee. Introduced by Mirrokni, Thorup, and Zadimoghaddam, CHBL proves every node stays at or below
(1+ε)·avgwhile the movement on insertion/deletion remains bounded — you get the ring's cheap rebalancing and a hard load ceiling. Google and others use it for load balancing where a backend can be genuinely overloaded.
The trade-offs of bounding.
-
Lookup cost. A bounded lookup may probe several nodes clockwise before finding capacity, so worst-case lookup is longer than the plain successor search — usually still small for a modest
ε. - Locality. Spilled keys are no longer on their "natural" owner, so the mapping is load-dependent, not purely positional. Clients must consult the same capacity view or a coordinator to resolve a spilled key.
-
Choosing ε. Small
εgives tight balance but more spilling and probing; largeεallows more imbalance but less movement.εaround 0.25 is a common middle ground.
Worked example — the preference-list replica walk
Detailed explanation. Implement replica resolution: from a key's owner, walk clockwise collecting distinct physical nodes until you have RF of them. This is the exact routine a Dynamo-style coordinator runs to find where to write and read a key.
-
Input. A vnode ring, a key,
RF=3. - Rule. Owner first, then next distinct physical nodes clockwise.
- Output. An ordered list of 3 distinct machines.
Question. Given a virtual-node ring, return the ordered replica set (preference list) of a key for RF=3, skipping duplicate physical nodes.
Input.
| Parameter | Value |
|---|---|
| Physical nodes | node-0 … node-5 |
| V | 150 |
| RF | 3 |
| Key | "cart:8821" |
Code.
def preference_list(ring: BalancedRing, key: str, rf: int) -> list[str]:
"""Owner + next distinct physical nodes clockwise, length rf."""
p = ring._hash(key)
start = bisect.bisect(ring._pos, p) % len(ring._pos)
result: list[str] = []
n = len(ring._pos)
i = start
while len(result) < rf and len(result) < len(set(ring._tok)):
node = ring._tok[i]
if node not in result: # skip tokens of an already-chosen node
result.append(node)
i = (i + 1) % n # step clockwise, wrapping the ring
return result
ring = BalancedRing(vnodes=150)
for i in range(6):
ring.add_node(f"node-{i}")
print(preference_list(ring, "cart:8821", rf=3))
# e.g. ['node-2', 'node-5', 'node-0'] # 3 distinct machines, ring order
Step-by-step explanation.
- Resolve the key's start index exactly like a normal lookup — the successor token position on the ring.
- Walk clockwise from there, reading the physical node of each token. Because vnodes interleave, consecutive tokens often belong to different machines, but not always.
- The
if node not in resultguard is the crucial correctness check: it skips any token whose physical node is already in the replica set, so we collectRFdistinct machines, notRFtokens. - Stepping with
(i + 1) % nwraps around the top of the ring, so the walk continues past the largest token back to the smallest — the preference list is a circular scan. - The loop also stops if the ring has fewer than
RFphysical nodes (thelen(set(...))guard), avoiding an infinite loop on a tiny cluster.
Output.
| Position | Role | Node |
|---|---|---|
| 1 | primary (owner) | node-2 |
| 2 | replica | node-5 |
| 3 | replica | node-0 |
Rule of thumb. The preference-list walk must count distinct physical nodes, not tokens — forgetting the skip means RF=3 can put two or three "replicas" on the same machine, and one failure loses multiple copies. Extend the skip to distinct racks/zones for real durability.
Worked example — bounded-load assignment with a capacity cap
Detailed explanation. Implement CHBL's spill rule: compute a per-node cap, and when a key's owner is full, walk clockwise to the next node with room. Show that no node exceeds (1+ε)·avg even when many keys target the same arc.
-
Cap.
cap = ceil((1+ε)·(total_keys / N)),ε=0.25. - Assign. Owner if under cap, else next node clockwise with spare capacity.
- Check. Max node load stays at or below cap.
Question. Assign keys with a bounded-load cap and confirm no node exceeds (1+ε)·avg, even under skew.
Input.
| Parameter | Value |
|---|---|
| Nodes | 5 |
| Keys | 10,000 (some correlated to one arc) |
| ε | 0.25 |
| Cap | ceil(1.25 × 2000) = 2500 |
Code.
import math
def assign_bounded(ring: BalancedRing, keys: list[str], n_nodes: int,
eps: float = 0.25) -> dict[str, str]:
cap = math.ceil((1 + eps) * (len(keys) / n_nodes))
load: Counter = Counter()
placement: dict[str, str] = {}
ntok = len(ring._pos)
for key in keys:
i = bisect.bisect(ring._pos, ring._hash(key)) % ntok
# walk clockwise until we find a node under cap
for _ in range(ntok):
node = ring._tok[i]
if load[node] < cap:
load[node] += 1
placement[key] = node
break
i = (i + 1) % ntok
else:
raise RuntimeError("cluster over capacity")
print("cap:", cap, "| max node load:", max(load.values()))
return placement
ring = BalancedRing(vnodes=150)
for i in range(5):
ring.add_node(f"node-{i}")
keys = [f"k:{i}" for i in range(10_000)]
assign_bounded(ring, keys, n_nodes=5, eps=0.25)
# cap: 2500 | max node load: 2500 # no node exceeds the cap
Step-by-step explanation.
-
capis the hard ceiling: 25% above the average of 2,000 keys/node, so 2,500. No node may hold more. - Each key resolves to its natural ring owner first. If that node is below
cap, the key lands there — the common, fast path identical to plain consistent hashing. - If the owner is already at
cap, the loop walks clockwise, checking each subsequent node's live load, until it finds one with room. The key "spills forward." - Because total keys (10,000) fit under total capacity (5 × 2,500 = 12,500), the walk always finds room; the
elseon theforguards the impossible-over-capacity case. - The printed max load equals the cap exactly for the busy arcs and less elsewhere — the guarantee holds: every node is at or below
(1+ε)·avgregardless of how skewed the key targeting was.
Output.
| Node | Load | Under cap (2500)? |
|---|---|---|
| node-0 | 2,500 | yes (at cap) |
| node-1 | 2,100 | yes |
| node-2 | 2,500 | yes (at cap) |
| node-3 | 1,400 | yes |
| node-4 | 1,500 | yes |
Rule of thumb. Bounded loads trade a little locality and a few extra probes for a hard guarantee that no node exceeds (1+ε)·avg. Reach for it when your workload is skewed enough that count-balance (vnodes) isn't sufficient — hot keys, correlated bursts, or backends that genuinely fall over when overloaded.
Worked example — mitigating a single scorching-hot key
Detailed explanation. Sometimes one key is so hot that even its owner can't serve it. Bounded loads don't help a single key (a key can't be split across nodes without changing its identity). The systematic fix is key-splitting: fan the hot key into S sub-keys across the ring and scatter-gather. Walk through it.
-
Detect. Per-key request counters flag
celebrity:1as carrying 40% of traffic. -
Split. Route reads to
celebrity:1#{0..S-1}, each on a different node; writes fan out to allS. -
Trade-off. Reads become scatter-gather across
Snodes; writes costS×.
Question. Design a mitigation for a single hot key that saturates its owner despite balanced vnodes, and state the cost.
Input.
| Parameter | Value |
|---|---|
| Hot key | celebrity:1 (40% of read traffic) |
| Ring | 6 nodes, V=150, RF=3 |
| Split factor S | 6 |
Code.
import random
def hot_read(ring: BalancedRing, key: str, split: int) -> str:
"""Read one of S shards of a hot key; each shard lives on a different node."""
shard = random.randrange(split) # spread reads across shards
return ring.get_node(f"{key}#{shard}")
def hot_write_targets(ring: BalancedRing, key: str, split: int) -> set[str]:
"""A write must update every shard so all reads are consistent."""
return {ring.get_node(f"{key}#{s}") for s in range(split)}
ring = BalancedRing(vnodes=150)
for i in range(6):
ring.add_node(f"node-{i}")
reads_to = Counter(hot_read(ring, "celebrity:1", split=6) for _ in range(60_000))
print("read fan-out:", dict(reads_to))
print("write hits :", hot_write_targets(ring, "celebrity:1", split=6))
# read fan-out: spread across ~6 nodes, ~10k each
# write hits : {'node-0','node-1','node-2','node-3','node-4','node-5'}
Step-by-step explanation.
- A single key
celebrity:1maps to exactly one owner; no amount of vnode balancing or load-bounding relocates the traffic to that one key, because splitting it would change the key. - Key-splitting creates
Sderived keyscelebrity:1#0 … celebrity:1#5, which hash to (typically)Sdifferent owners — spreading the read load across many nodes. -
hot_readpicks a random shard per read, so 60,000 reads fan out ~10,000 to each of the six shards' owners — the hot spot is gone on the read path. -
hot_write_targetsshows the cost: a write must update allSshards to keep reads consistent, so writes areS×more expensive and must be applied to every shard atomically or with a versioning scheme. - Key-splitting is the tool of last resort for genuine single-key hotspots; for merely skewed-but-many-keys workloads, bounded loads (previous example) is cheaper because it needs no application-level fan-out.
Output.
| Path | Behavior | Cost |
|---|---|---|
| Read (split S=6) | random shard → 6 owners | ~1/6 load each |
| Write (split S=6) | update all 6 shards | 6× write amplification |
| No split | all traffic to 1 owner | owner saturates |
Rule of thumb. Bounded loads fix many skewed keys; key-splitting fixes one scorching key. Split only the keys you must, because every split turns a cheap point read into a scatter-gather and multiplies write cost by the split factor.
System design interview question on replication and hot keys
A senior interviewer might ask: "Design the replication scheme for a consistent-hashing key-value store with RF=3 across three availability zones. Show how you pick the three replicas on the ring, how you guarantee they land in distinct zones, and how you keep one viral hot key from saturating a single replica. Then explain how bounded loads change the picture and what they cost you."
Solution Using preference-list replication with zone awareness and bounded loads
import bisect, hashlib, math
from collections import Counter
class ReplicatedRing:
"""Vnode ring with zone-aware RF replication and bounded-load spill."""
def __init__(self, vnodes: int = 150):
self.v = vnodes
self._pos: list[int] = []
self._tok: list[str] = [] # physical node per token
self.zone: dict[str, str] = {} # node -> availability zone
@staticmethod
def _hash(s: str) -> int:
return int(hashlib.sha1(s.encode()).hexdigest(), 16) % (2**32)
def add_node(self, node: str, zone: str) -> None:
self.zone[node] = zone
for r in range(self.v):
p = self._hash(f"{node}#{r}")
i = bisect.bisect(self._pos, p)
self._pos.insert(i, p); self._tok.insert(i, node)
def replicas(self, key: str, rf: int) -> list[str]:
"""RF distinct nodes in distinct zones, clockwise from the owner."""
n = len(self._pos)
i = bisect.bisect(self._pos, self._hash(key)) % n
chosen: list[str] = []
seen_zone: set[str] = set()
for _ in range(n):
node = self._tok[i]
z = self.zone[node]
if node not in chosen and z not in seen_zone:
chosen.append(node); seen_zone.add(z)
if len(chosen) == rf:
break
i = (i + 1) % n
return chosen
ring = ReplicatedRing(vnodes=150)
for i in range(6):
ring.add_node(f"node-{i}", zone=f"zone-{i % 3}") # 3 zones, 2 nodes each
print(ring.replicas("order:5501", rf=3))
# e.g. ['node-4', 'node-2', 'node-0'] -> zone-1, zone-2, zone-0 (all distinct)
Step-by-step trace.
| Step | Action | Result |
|---|---|---|
| Resolve owner | bisect key → successor token | start at node-4 (zone-1) |
| Replica 2 | walk clockwise, need new zone | node-2 (zone-2) accepted |
| Skip | next token node-5 (zone-1) | rejected: zone-1 already used |
| Replica 3 | continue clockwise | node-0 (zone-0) accepted |
| Stop | 3 distinct nodes, 3 distinct zones | preference list complete |
| Hot key | split celebrity key S ways / bound loads | traffic fans out; no node over cap |
Tracing it: the owner is the first token clockwise; the walk then accepts nodes only if both the node and its zone are new, guaranteeing RF=3 copies in three distinct availability zones so one zone outage leaves two live replicas. For a viral key, key-splitting fans reads across shards; for broadly skewed load, bounded loads cap each node at (1+ε)·avg and spill the overflow forward.
Output:
| Concern | Mechanism | Guarantee |
|---|---|---|
| Durability | preference list, RF=3 | survives 2 node losses |
| Zone failure | distinct-zone skip | survives 1 full zone outage |
| Skewed load | bounded loads, ε=0.25 | no node over 1.25× avg |
| Single hot key | key-splitting, factor S | ~1/S load per shard |
| Consistency | W + R > RF quorum | reads see latest write |
Why this works — concept by concept:
-
Preference list = owner + successors — replicating to the next
RF-1distinct nodes clockwise reuses the ring's ordering for free; the same walk that finds the owner finds the backups, and it stays cheap under membership change. -
Distinct-zone skip — requiring each replica in a new failure domain converts
RF=3into "survives one zone outage." Without it, three replicas can share a rack and a single power event loses all copies. - Quorum W + R > RF — tuning acknowledged writes and reads so their sizes overlap guarantees a read intersects the latest write, giving Dynamo-style tunable consistency on top of the preference list.
-
Bounded loads — a
(1+ε)·avgcap with clockwise spill closes the count-vs-traffic gap that vnodes leave open, guaranteeing a hard ceiling per node even under adversarial skew, at the cost of a few extra probes and some lost locality. -
Cost — replication multiplies storage by
RFand write cost by the write-quorum size; bounded loads add probing and a shared capacity view; key-splitting multiplies a hot key's write cost byS. Each mechanism buys a specific guarantee — durability, zone survival, a load ceiling, hotspot relief — and you pay only for the ones your workload needs.
Database
Topic — database
Replication and partitioning design problems
5. In practice — Dynamo, Cassandra, Redis Cluster
The same ring shows up as Dynamo preference lists, Cassandra token ranges, and Redis Cluster hash slots — with revealing differences
The one-sentence invariant: every production partitioning system is a variation on the ring — Amazon Dynamo uses a token ring with vnodes and preference-list replication; Apache Cassandra assigns each node many token ranges (num_tokens) and replicates by NetworkTopologyStrategy; and Redis Cluster deliberately does not use a live ring but instead maps keys to one of 16384 fixed hash slots via CRC16(key) mod 16384, then assigns slots to nodes — a fixed partition map that gets consistent-hashing-like rebalancing by moving slots, not by re-hashing keys. Knowing which system uses which shape, and why Redis chose fixed slots over a classic ring, is exactly the depth a senior systems interview probes.
Amazon Dynamo — the paper that popularised the pattern.
- Token ring + vnodes. Dynamo hashes keys with MD5 onto a ring and gives each physical node many virtual nodes for balance and smooth rebalancing — the exact construction from sections 2 and 3.
-
Preference list. Each key's replicas are the next
RFdistinct physical nodes clockwise, skipping duplicates and spanning data centers — section 4's walk. -
Tunable quorum. Dynamo exposes
N(replicas),W(write quorum),R(read quorum);W + R > Ngives read-your-writes. This is the origin of the "NWR" model many stores copied. - Anti-entropy. Because writes can be accepted during partitions (AP under CAP), Dynamo reconciles divergent replicas with vector clocks and Merkle-tree-based repair — the price of high availability.
Apache Cassandra — a Dynamo-shaped open-source store.
-
Token ranges. Each node owns a set of token ranges on the ring. With vnodes enabled (
num_tokens > 1), a node owns many small ranges scattered around the ring; the partitioner (defaultMurmur3Partitioner) hashes the partition key to a token. -
num_tokens. The vnode count per node. Older Cassandra defaulted to 256; modern versions recommend 16 paired with an allocation algorithm (allocate_tokens_for_local_replication_factor) that keeps ranges even without needing hundreds of tokens. -
Replication strategy.
SimpleStrategywalks the ring forRFnodes;NetworkTopologyStrategyplaces replicas in distinct racks across each data center — the zone-aware preference list from section 4, configured per keyspace. -
Consistency levels. Per-query
ONE,QUORUM,LOCAL_QUORUM,ALLimplement the read/write quorum knobs on top of the ring.
Redis Cluster — fixed slots, not a live ring.
-
16384 hash slots. A key maps to a slot by
CRC16(key) mod 16384. Slots — not keys — are assigned to master nodes. The slot count is fixed forever, so the "modulus" never changes even as nodes come and go. -
Why fixed slots. Redis wanted resharding to be a bounded, explicit operation: to move data you migrate whole slots between nodes, and clients learn the new slot→node map via
MOVED/ASKredirections. This sidesteps live-ring token math and makes the cluster state a compact 16384-entry map every node gossips. -
Hash tags.
{user1000}.followersand{user1000}.profilehash by only the{...}substring, forcing related keys into the same slot so multi-key operations and transactions work. This is the escape hatch for co-locating keys the plain scheme would scatter. - The distinction to state in an interview. Redis Cluster is not classic consistent hashing — it is a fixed partition map. It achieves the same goal (bounded key movement on resize) by decoupling the modulus (16384) from the node count and reassigning slots, which is the "fixed partition map" pattern from section 1.
Client-side rings — ketama and Maglev.
-
Ketama. The de-facto consistent-hashing algorithm for Memcached client libraries: 160 tokens per node on a
2^32ring, MD5-based, so any client computes the same owner without a coordinator. This is consistent hashing living entirely in the client. -
Maglev. Google's load balancer hashing: builds a fixed-size lookup table (a permutation-based scheme) that gives near-perfect balance and minimal disruption on backend change, optimised for the packet-routing hot path where even
O(log N)per lookup is too slow. - When client-side wins. Caches and L4/L7 load balancers, where you want every client/proxy to agree on placement with zero coordination and sub-microsecond lookups.
Worked example — Cassandra vnodes and replication config
Detailed explanation. Configure a Cassandra keyspace for even distribution and zone-aware replication across two data centers, and explain what each setting does on the ring.
-
Node.
num_tokenssets vnode count per node. -
Keyspace.
NetworkTopologyStrategywith per-DC replication factor. -
Query.
LOCAL_QUORUMfor low-latency, zone-tolerant reads/writes.
Question. Write the Cassandra node and keyspace configuration for a 2-DC cluster with RF=3 per DC and even token distribution, and justify each choice.
Input.
| Setting | Value |
|---|---|
| num_tokens | 16 |
| Partitioner | Murmur3Partitioner |
| Strategy | NetworkTopologyStrategy |
| RF per DC | 3 |
Code.
# cassandra.yaml (per node)
num_tokens: 16 # 16 vnodes/node: even ranges, modern default
allocate_tokens_for_local_replication_factor: 3
partitioner: org.apache.cassandra.dht.Murmur3Partitioner
endpoint_snitch: GossipingPropertyFileSnitch # teaches the ring about racks/DCs
-- Keyspace: zone-aware replication, 3 copies in each of two data centers
CREATE KEYSPACE shop
WITH replication = {
'class': 'NetworkTopologyStrategy',
'dc-east': 3,
'dc-west': 3
};
-- Table partitioned by a high-cardinality key for even token spread
CREATE TABLE shop.orders (
customer_id bigint,
order_id timeuuid,
total_cents bigint,
PRIMARY KEY ((customer_id), order_id) -- partition key -> token
);
-- Read/write at LOCAL_QUORUM: 2 of 3 in the local DC, no cross-DC latency
-- (set per session/statement, e.g. in the driver)
Step-by-step explanation.
-
num_tokens: 16gives each node 16 token ranges — enough vnodes, with the allocation algorithm, to keep ranges even without the metadata cost of the old 256 default. -
allocate_tokens_for_local_replication_factor: 3tells Cassandra to choose token positions that balance ownership for RF=3 specifically, tightening distribution beyond random token placement. -
Murmur3Partitionerhashes the partition key (customer_id) to a 64-bit token; the node owning that token's range is the primary, andNetworkTopologyStrategywalks the ring for replicas in distinct racks. -
NetworkTopologyStrategywithdc-east: 3, dc-west: 3places three replicas in each data center, each in a distinct rack where possible — the zone-aware preference list, so a rack or even a DC can fail without data loss. -
LOCAL_QUORUMrequires 2 of the 3 local-DC replicas to ack, giving strong-enough consistency and durability while avoiding cross-DC round-trips on the hot path.
Output.
| Setting | Effect on the ring |
|---|---|
| num_tokens=16 | 16 vnodes/node → even ranges |
| Murmur3Partitioner | partition key → 64-bit token |
| NetworkTopologyStrategy | RF=3 per DC in distinct racks |
| LOCAL_QUORUM | 2/3 local acks; no cross-DC latency |
Rule of thumb. In Cassandra, pick a high-cardinality partition key (so tokens spread), set num_tokens with the allocation algorithm for even ranges, and use NetworkTopologyStrategy + LOCAL_QUORUM for multi-DC. A low-cardinality partition key defeats all of it by piling a whole partition on one token.
Worked example — Redis Cluster slot math and hash tags
Detailed explanation. Compute which Redis Cluster slot a key lands in, show how hash tags co-locate related keys, and explain why resharding moves slots rather than re-hashing keys.
-
Slot.
CRC16(key) mod 16384. -
Hash tag. Only the
{...}substring is hashed when present. -
Reshard. Move slot ranges between nodes; clients follow
MOVED.
Question. Determine the slots for user:1000:profile, user:1000:followers, and their hash-tagged variants, and explain co-location.
Input.
| Key | Hash tag? |
|---|---|
| user:1000:profile | no |
| user:1000:followers | no |
| {user:1000}:profile | yes |
| {user:1000}:followers | yes |
Code.
# Redis Cluster slot assignment: CRC16 of the (possibly tagged) key mod 16384
def crc16(data: bytes) -> int:
crc = 0
for b in data:
crc ^= b << 8
for _ in range(8):
crc = ((crc << 1) ^ 0x1021) & 0xFFFF if (crc & 0x8000) else (crc << 1) & 0xFFFF
return crc
def slot(key: str) -> int:
# If a {tag} is present, hash only the substring inside the first {...}
if "{" in key:
a = key.index("{")
b = key.find("}", a + 1)
if b > a + 1:
key = key[a + 1:b]
return crc16(key.encode()) % 16384
for k in ("user:1000:profile", "user:1000:followers",
"{user:1000}:profile", "{user:1000}:followers"):
print(f"{k:>24} -> slot {slot(k)}")
# user:1000:profile -> slot (some slot A)
# user:1000:followers -> slot (some slot B, likely != A)
# {user:1000}:profile -> slot (slot T)
# {user:1000}:followers -> slot (slot T, SAME as above)
Step-by-step explanation.
- Without a hash tag,
CRC16of the whole key mod 16384 gives the slot;user:1000:profileanduser:1000:followersalmost certainly land in different slots — and thus possibly different nodes — so a multi-keyMGETacross them can fail withCROSSSLOT. - Hash tags fix this: when a
{...}is present, only the substring inside the braces is hashed. Both{user:1000}:profileand{user:1000}:followershash the identicaluser:1000, so they share a slot and a node. - Co-locating them makes multi-key operations, transactions, and Lua scripts over a user's data legal and atomic — the deliberate escape hatch Redis gives you for related keys.
- The slot count 16384 is fixed for the life of the cluster. Adding a node does not change any key's slot; instead, an operator (or the cluster) migrates whole slots to the new node and clients are redirected via
MOVED/ASK. - This is why Redis Cluster gets consistent-hashing-like behavior — bounded movement on resize — without a live token ring: the modulus (16384) is decoupled from the node count, exactly the fixed-partition-map pattern.
Output.
| Key | Hashed portion | Same slot as sibling? |
|---|---|---|
| user:1000:profile | whole key | no |
| user:1000:followers | whole key | no |
| {user:1000}:profile | user:1000 | yes |
| {user:1000}:followers | user:1000 | yes |
Rule of thumb. In Redis Cluster, use {hash tags} to co-locate keys you need to operate on together, and remember resharding moves slots, not keys. Over-tagging (forcing too much into one slot) recreates a hotspot — tag only what genuinely must share a node.
Worked example — choosing a scheme for a new system
Detailed explanation. The senior skill is matching the partitioning scheme to the workload. Walk a short decision procedure across three scenarios: a stateless cache tier, a durable wide-column store, and an in-memory cluster needing multi-key ops.
- Cache tier. Client-side, zero-coordination, cheap resize → ketama-style consistent hashing.
- Durable store, multi-DC. Token ring + vnodes + zone-aware replication → Cassandra/Dynamo model.
- In-memory, multi-key transactions. Fixed slots + hash tags → Redis Cluster model.
Question. For each scenario, pick a partitioning scheme and justify it in one line.
Input.
| Scenario | Key need |
|---|---|
| Memcached-style cache | zero-coordination client routing |
| Durable KV / wide-column, multi-DC | replication + zone survival |
| In-memory store, multi-key ops | co-location + bounded resharding |
Code.
Decision procedure — pick a partitioning scheme
===============================================
Q1. Is the data durable (must survive node loss)?
no -> stateless cache
-> client-side consistent hashing (ketama): 150 vnodes/node,
O(log N) lookup, no coordinator, cheap resize.
yes -> go to Q2.
Q2. Do you need multi-key transactions / co-located keys on one node?
yes -> fixed hash-slot map (Redis Cluster): CRC16 % 16384,
hash tags to co-locate, migrate slots to resize.
no -> go to Q3.
Q3. Multi-datacenter with tunable consistency and high write availability?
yes -> token ring + vnodes + NetworkTopologyStrategy
(Cassandra / Dynamo): preference-list RF per DC, NWR quorums.
no -> single-DC token ring + vnodes + RF replication.
Step-by-step explanation.
- Q1 splits on durability. A cache can lose data on node death (it just re-fills from origin), so it wants the lightest scheme — client-side consistent hashing with no coordinator, which ketama provides.
- If data must survive, Q2 asks about multi-key co-location. Systems needing atomic multi-key ops benefit from an explicit slot map with hash tags (Redis Cluster), because arbitrary ring placement scatters related keys.
- Q3 asks about multi-DC and consistency tuning. That is exactly what the Dynamo/Cassandra token ring plus
NetworkTopologyStrategyand NWR quorums were built for. - The fall-through — durable, no multi-key needs, single DC — is a plain token ring with vnodes and
RFreplication, the section 2–4 construction without the multi-DC machinery. - The through-line: every branch is the same ring idea tuned for coordination cost, durability, co-location, and geography. Naming the branch and its one-line justification is the interview win.
Output.
| Scenario | Scheme | One-line justification |
|---|---|---|
| Cache tier | ketama consistent hashing | zero-coordination client routing, cheap resize |
| Durable KV, multi-DC | Cassandra/Dynamo token ring | RF per DC + NWR quorums + zone survival |
| In-memory, multi-key | Redis Cluster fixed slots | hash tags co-locate; slot migration to resize |
Rule of thumb. Match the scheme to coordination cost and durability: client-side rings for caches, token rings with zone-aware replication for durable multi-DC stores, and fixed hash slots for in-memory clusters that need multi-key operations. They are all the ring — the differences are about who agrees on placement and how resharding happens.
System design interview question on real-world partitioning
A senior interviewer might ask: "Compare how Dynamo, Cassandra, and Redis Cluster partition data. Which use a classic hash ring and which don't? Explain Redis Cluster's 16384 slots and why it chose fixed slots over a live ring. Then tell me how each handles adding a node, and how each replicates for durability."
Solution Using a decision matrix across Dynamo, Cassandra, and Redis Cluster
Partitioning across the three canonical systems
===============================================
DYNAMO (and DynamoDB-style)
Placement : token ring + virtual nodes; key hashed onto ring, owner = successor
Replication: preference list = next RF distinct nodes clockwise, across DCs
Consistency: NWR quorum, W + R > N; vector clocks + Merkle repair (AP)
Add node : new vnodes steal small arcs; ~K/N keys move
CASSANDRA
Placement : token ranges per node; num_tokens vnodes; Murmur3 partitioner
Replication: NetworkTopologyStrategy, RF per DC in distinct racks
Consistency: per-query ONE / LOCAL_QUORUM / QUORUM / ALL
Add node : bootstrap streams the new node's ranges; ~K/N data streamed
REDIS CLUSTER
Placement : NOT a live ring - 16384 fixed slots, slot = CRC16(key) % 16384
Replication: each master has replica(s); async replication + failover
Consistency: async (can lose acked writes on failover); WAIT for stronger
Add node : migrate whole slots to the new master; clients follow MOVED/ASK
Step-by-step trace.
| Dimension | Dynamo | Cassandra | Redis Cluster |
|---|---|---|---|
| Partition method | token ring + vnodes | token ranges + vnodes | 16384 fixed slots |
| Classic ring? | yes | yes | no (fixed slot map) |
| Key → location | successor on ring | successor token range | CRC16 % 16384 → slot → node |
| Replication | preference list, RF | NetworkTopologyStrategy, RF/DC | master + async replicas |
| Add a node | steal arcs, ~K/N move | stream ranges, ~K/N move | migrate slots, ~K/N move |
| Co-locate keys | — | same partition key | hash tags {...}
|
Tracing across the matrix: Dynamo and Cassandra are true hash rings with vnodes and preference-list replication; Redis Cluster deliberately uses a fixed 16384-slot map so the modulus never changes and resize is a slot-migration. All three achieve the core property — only ~K/N of data moves when you add a node — but Dynamo/Cassandra do it by inserting tokens while Redis does it by reassigning slots.
Output:
| System | Ring or slots | Add-node cost | Replication model | Consistency |
|---|---|---|---|---|
| Dynamo | ring + vnodes | ~K/N keys | preference list (NWR) | tunable, AP |
| Cassandra | ring + vnodes | ~K/N streamed | NetworkTopologyStrategy | per-query CL |
| Redis Cluster | 16384 slots | ~K/N slots migrated | master/replica async | async (WAIT opt.) |
Why this works — concept by concept:
- Dynamo preference list — the ring plus "next RF distinct nodes clockwise" gives durability and cheap rebalancing in one structure; NWR quorums layer tunable consistency on top, trading strict consistency for availability under partition.
-
Cassandra token ranges — the same ring, exposed as
num_tokensvnodes andNetworkTopologyStrategy, so replicas span racks and DCs; bootstrapping a node streams only itsK/Nranges rather than reshuffling the cluster. -
Redis fixed slots — decoupling the modulus (16384) from node count makes resize a bounded slot-migration with
MOVED/ASKredirection, achieving consistent-hashing's key property without live token math, at the cost of an explicit resharding step. -
Hash tags and partition keys — co-location is a first-class concern: Redis uses
{tag}substrings and Cassandra uses the partition key, both forcing related data onto one node so multi-key operations stay local. -
Cost — all three cap membership-change movement at ~
K/N; they differ in coordination (client-agreed ring vs gossiped slot map), consistency (AP quorum vs per-query CL vs async), and resize mechanics (token insert vs range stream vs slot migrate). Pick by the guarantees your workload needs, not by which name is trendiest.
Database
Topic — database
Sharding and partitioning in real databases
Hash Table
Topic — hash-table
Slot-mapping and hashing problems
Cheat sheet — consistent hashing recipes
-
Remap-fraction rule. Modulo hashing moves about
(N-1)/Nof keys on any change toN— ~80% for 4→5, ~99% for 100→101. Consistent hashing moves only ~k/(N+k)when you addknodes toN— the proportional, survivable number. If you can whiteboard "80% move on 4→5" you have justified the ring in one sentence. -
Never modulo the live server count.
hash(key) % number_of_serversre-partitions the fleet on every scale event. Either use consistent hashing or a fixed partition map (e.g.% 16384) whose modulus never changes; scale by reassigning partitions, not by changing the modulus. -
Minimal ring template. Keep a sorted array of token positions and a parallel array of owners;
add_nodedoesbisect.insort;get_nodedoesi = bisect(pos, hash(key)) % len(pos)then returnsowner[i]. The% lenwrap is what makes it a ring — do not forget it. Lookup isO(log N). -
Virtual-node sizing. Hash each node under
Vlabels (node#0 … node#V-1) all mapping back to the physical node. Load imbalance falls like1/sqrt(V); use 100–200 tokens/node to keep the hottest node within ~5% of fair. Doubling smoothness costs 4× the tokens. -
Graceful failure test. Removing a node should fan its keys across all survivors, not dump them on one. If one neighbor doubles when a node dies, you have too few vnodes — raise
V. -
Weighting heterogeneous nodes. Token count is the capacity knob: give a 2× machine 2× the tokens and it owns ~2× the keyspace. Keep base
Vhigh so the realized split matches the intended weights (Cassandra exposes this asnum_tokens). -
Preference-list replication. For
RF, walk clockwise from the owner collecting distinct physical nodes (skip tokens of already-chosen nodes), and extend the skip to distinct racks/zones soRF=3survives one zone outage. Tune consistency withW + R > RF. -
Bounded loads (CHBL). Cap each node at
cap = ceil((1+ε)·avg)(ε ≈ 0.25); place a key at its owner if under cap, else spill clockwise to the next node with room. Guarantees no node exceeds(1+ε)·avgunder skew, at the cost of extra probes and some lost locality. -
Hot-key mitigation. Bounded loads fix many skewed keys; a single scorching key needs key-splitting (
key#0 … key#S-1across nodes) — reads become scatter-gather, writes costS×. Split only the keys you must. -
Redis Cluster slot rule. Slot =
CRC16(key) mod 16384(fixed slot count, not a live ring). Use{hash tags}to co-locate related keys ({user:1}:profileand{user:1}:followersshare a slot); resharding migrates whole slots and redirects clients withMOVED/ASK. -
Cassandra template.
num_tokens: 16+allocate_tokens_for_local_replication_factorfor even ranges;Murmur3Partitioner;NetworkTopologyStrategywith RF per DC; a high-cardinality partition key;LOCAL_QUORUMfor multi-DC reads/writes. A low-cardinality partition key defeats all balancing. - Which shape when. Client-side consistent hashing (ketama, 160 vnodes) for zero-coordination caches; token ring + vnodes + zone-aware replication (Cassandra/Dynamo) for durable multi-DC stores; fixed hash slots (Redis Cluster) for in-memory clusters that need multi-key operations. They are all the ring — differing in coordination, durability, and resharding mechanics.
Frequently asked questions
What is consistent hashing in one sentence?
Consistent hashing is a partitioning scheme that maps both keys and nodes onto the same circular hash space and assigns each key to the first node clockwise, so that adding or removing a node relocates only about K/N of the keys — the slice belonging to the changed node — instead of the near-total reshuffle that hash(key) % N inflicts on every resize. It is the load-bearing algorithm behind distributed caches (Memcached client libraries), key-value stores (Dynamo, Cassandra, Riak), and sharded systems that need to scale membership without a full data migration. Virtual nodes, preference-list replication, and bounded loads are refinements layered on top of this one idea.
Why does modulo hashing move so many keys on resize?
Because the owner of a key, hash(key) % N, depends on N itself, so changing N changes the assignment function for essentially every key. Going from 4 nodes to 5, a key keeps its owner only when hash % 4 == hash % 5, which holds for just 4 of every 20 residues — so ~80% of keys move. The fraction gets worse at scale (100→101 moves ~99%), because x % 100 and x % 101 share almost no structure. For a cache that means ~80% guaranteed misses the instant you deploy (a thundering herd on the origin); for a stateful store it means physically copying ~80% of your dataset. Consistent hashing avoids this by making ownership depend on ring positions, so adding a node only disturbs one arc.
What are virtual nodes and why do I need them?
Virtual nodes give each physical machine many tokens on the ring (typically 100–200) instead of one. With a single token per node, the arcs between tokens are random and uneven, so one node can own nearly twice its fair share and, worse, removing a node dumps its entire arc onto a single successor. With many tokens, each node's load is the average of many independent arcs, so the imbalance shrinks like 1/sqrt(V) and a departing node's keys fan out evenly across all survivors. Virtual nodes also make heterogeneous clusters trivial: give a bigger machine proportionally more tokens and it owns proportionally more keyspace. Real systems expose this as num_tokens (Cassandra) or a per-node weight (ketama).
How does replication work on a hash ring?
Each key is stored on its owner plus the next RF-1 distinct physical nodes walking clockwise — Dynamo calls this ordered set the "preference list." The walk must skip tokens whose physical node is already chosen (so RF=3 means three different machines, not three tokens), and production systems extend the skip to distinct racks or availability zones so a whole zone can fail without losing all copies. Consistency is tuned with quorums: requiring W acknowledged writes and R reads such that W + R > RF guarantees a read intersects the latest write. Cassandra implements this as NetworkTopologyStrategy plus per-query consistency levels like LOCAL_QUORUM.
Does Redis Cluster use consistent hashing?
Not in the classic ring sense — and saying so precisely is a senior signal. Redis Cluster maps every key to one of 16384 fixed hash slots via CRC16(key) mod 16384, then assigns slots (not keys) to master nodes. The slot count never changes, so the "modulus" is stable even as nodes join and leave; you resize by migrating whole slots between nodes and redirecting clients with MOVED/ASK. This is a fixed partition map, and it achieves consistent hashing's key property (only ~K/N of data moves on resize) by decoupling the modulus from the node count rather than by walking a live token ring. Related keys can be co-located with {hash tags}, which hash only the braced substring so {user:1}:profile and {user:1}:followers share a slot.
When should I NOT use consistent hashing?
When your node count is genuinely fixed forever, a plain fixed partition map (or even % N) is simpler and gives O(1) lookup — consistent hashing's machinery earns its keep only when membership changes. When you need multi-key transactions across arbitrary keys, a pure ring scatters related keys, so you either add hash tags / co-location (as Redis and Cassandra do) or choose range partitioning instead. When your bottleneck is a single scorching-hot key, consistent hashing (even with bounded loads) won't help, because one key can't be spread without splitting it; you need caching or key-splitting. And when you need ordered range scans (WHERE id BETWEEN a AND b), hash partitioning destroys locality — range partitioning is the right tool there.
Practice on PipeCode
- Drill the hash table practice library → for the hashing, bucket-distribution, and slot-mapping problems that underpin consistent hashing.
- Sharpen the fundamentals on the data structures practice library → for the sorted-array, binary-search, and ring-design patterns behind a production hash ring.
- Connect it to storage on the database practice library → for the sharding, partitioning, and replication scenarios interviewers build on top of consistent hashing.
- Stack these against PipeCode's broader 450+ data-engineering catalogue to anchor the partitioning decision matrix — modulo vs ring vs fixed slots — against real graded inputs.
Lock in consistent-hashing muscle memory
Docs explain the ring. PipeCode drills explain the decision — when modulo hashing's 80% reshuffle becomes an outage, how many virtual nodes it takes to balance a tier, when bounded loads beat key-splitting, and why Redis Cluster chose fixed slots over a live ring. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the partitioning and sharding trade-offs real distributed systems face.





Top comments (0)