DEV Community

Cover image for Consistent Hashing: How Distributed Systems Partition Data
Gowtham Potureddi
Gowtham Potureddi

Posted on

Consistent Hashing: How Distributed Systems Partition Data

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.

PipeCode blog header for consistent hashing — bold white headline 'Consistent Hashing' over a hero hash ring with node medallions and keys, one node removed and only its keys remapping clockwise to the next node, around a central purple seal, on a dark gradient.

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


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 N is 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) % N needs no shared state beyond N.
  • 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/N keys for one node out of N. This is the property modulo hashing catastrophically lacks, and it is the entire reason consistent hashing was 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 N to N+1, the expected fraction of keys that stay is roughly 1/(N+1) in the worst intuition and never better than a small constant; in practice you should assume that any change to N re-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) % 256 and then map partitions to physical nodes through a separate lookup table. Here N (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 hashing or 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 = 4 then N = 5.
  • Measurement. Count keys whose bucket differs between the two runs.
  • Expectation. Around 80% of keys move — matching the 4/20 residue 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%)
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. node_for is 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 the n remainders is equally likely — balance is fine.
  2. before records every key's owner in the 4-node world; after records the owner after a single node is added, making it a 5-node world. Nothing about the keys changed — only the modulus.
  3. The comparison counts keys whose owner differs. Because hash % 4 and hash % 5 are effectively independent for a uniform hash, agreement happens only for the residues where they coincide — 4 out of every 20, or 20%.
  4. 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, because hash % 100 and hash % 101 share almost no structure.
  5. 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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. 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.
  2. 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.
  3. 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.
  4. 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.
  5. 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/5 of 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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. The routing rule user_id % 8 lives in application code, so every read and write already assumes 8 shards. Changing to % 9 is a code change and a data change — both must land atomically or queries route to the wrong shard.
  2. The SQL counts how many rows would compute a different shard under the new modulus. As with the cache case, x % 8 and x % 9 rarely agree, so ~88.9% of rows must relocate.
  3. 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.
  4. The migration typically runs as: stand up shard 9, dual-write new data under the % 9 rule, 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.
  5. Had the tier used consistent hashing (or a fixed 256-partition map decoupled from instance count), adding a ninth instance would move only ~1/9 of 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%
Enter fullscreen mode Exit fullscreen mode

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^32 circle 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 add k nodes to N means 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 is O(all keys). Ring lookup is O(log N) per key via binary search over the sorted token positions, and resize is O(K/N) keys moved plus O(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

Practice →

Data Structures Topic — data-structures Data-structure design problems

Practice →


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.

Iconographic diagram of modulo hashing failing on resize — keys mapped by hash(key) % N, and when N changes from 4 to 5 almost every key arrow jumps to a different node, shown as a storm of remapped arrows.

The construction — three steps to a working ring.

  • Hash the nodes onto the circle. For each node, compute hash(node_id) mod 2^32 and 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 from 2^32 - 1 back to 0.
  • 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 N roughly-evenly-spaced tokens, one node owns about 1/N of the circle, so adding or removing a node moves about K/N keys — 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, or O(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 N is 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^32 positions.
  • Store. Parallel sorted arrays: _keys (positions) and _nodes (owners).
  • Lookup. bisect for 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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. _hash folds 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.
  2. add_node finds the insertion point with bisect and splices the token into both parallel arrays, keeping them sorted so lookups stay O(log N).
  3. remove_node locates the exact token by position and confirms the node identity before removing — guarding against a hash collision removing the wrong entry.
  4. get_node binary-searches for the first token strictly after the key's position. If the key hashes past the largest token, bisect returns len(self._pos), and % len wraps the index back to token 0 — that wrap is the "circle" made concrete.
  5. Because ownership is a pure function of token positions, calling get_node again 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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. 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.
  2. before snapshots every key's owner. Adding "E" inserts exactly one new token; the ring's other three tokens do not move.
  3. "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/5 of the circle once there are five tokens.
  4. 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.
  5. 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)
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. 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.
  2. 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.
  3. 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.
  4. 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.
  5. The cure is many small tokens per node: with V tokens each, the arcs a node owns average out and the imbalance shrinks like 1/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]
Enter fullscreen mode Exit fullscreen mode

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. bisect finds that successor in O(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 pays O(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.
  • CostO(log N) lookup, O(N) membership mutation for the array form, and O(K/N) keys relocated per node change. Compared with modulo's O(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

Practice →

Hash Table Topic — hash-table Hashing and key-lookup problems

Practice →


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.

Iconographic hash ring diagram — keys placed on a circular ring walking clockwise to the next node token that owns them, with three node tokens spaced around the ring.

How virtual nodes are constructed.

  • Many labels per node. For physical node A and V virtual nodes, hash A#0, A#1, … A#(V-1) onto the ring — V separate tokens that all map back to A. A lookup that lands on any of them resolves to A.
  • 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 V tokens 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 V tokens, its load is the sum of V independent arcs; the sum concentrates around the mean, and the relative standard deviation falls like 1/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 V arcs 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 V tokens picks up V small 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 V per node.
  • The practice. Cassandra exposes this as num_tokens; give beefier hosts a larger value. Client libraries like ketama let 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 V is 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 × V entries), 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_tokens to 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 V in {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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. add_node now inserts V tokens per physical node, each hashed from a distinct label node#r. All V tokens map back to the same physical node via the parallel _node array.
  2. get_node is unchanged — it finds the successor token and returns its physical node. The vnode machinery is invisible at lookup time; only the token density changed.
  3. At V=1 the ring is the bare ring from section 2, and the hottest node sits near 1.8× fair — the lumpiness we measured before.
  4. As V rises, each node's load is the sum of more independent arcs, so the distribution tightens. By V=50 the hottest node is within ~12% of fair; by V=200 within ~5%.
  5. The improvement tracks 1/sqrt(V): quadrupling V roughly 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=1 and V=150.
  • Action. Remove one node; track which survivors absorb its keys.
  • Expectation. V=1 dumps everything on one successor; V=150 spreads 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}
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. orphans are the keys that node-3 owned before removal — the load that must go somewhere.
  2. 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.
  3. 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).
  4. 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.
  5. 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 × weight tokens 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%
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. add_weighted multiplies the token count by the node's weight, so node-big drops 300 tokens versus 150 for the others.
  2. Since keyspace ownership is proportional to token count once V is large enough to average out, node-big's 300 of 600 total tokens claim ~50% of the ring.
  3. node-a and node-b split the remaining half, ~25% each — exactly proportional to their weight-1 share.
  4. 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.
  5. The averaging caveat applies: at low base V the 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()}
Enter fullscreen mode Exit fullscreen mode

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 V distinct 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 V independent 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_tokens or per-node weights.
  • Cost — the token array grows to N × V entries and gossip/metadata scales with it, so lookups stay O(log(N·V)) and memory and update cost rise linearly in V. 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

Practice →

Hash Table Topic — hash-table Hashing and bucket-distribution problems

Practice →


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.

Iconographic virtual-nodes diagram — three physical nodes each owning many small virtual points scattered evenly around the ring so load is balanced.

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 next RF-1 nodes 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=3 means 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=3 should land in three zones so one zone outage can't take all copies.
  • Read/write quorums. With RF replicas you tune consistency by requiring W acknowledged writes and R reads such that W + R > RF guarantees 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 / N and a per-node capacity cap = ceil((1+ε)·avg) for a small ε (say 0.25). No node may hold more than cap items.
  • 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+ε)·avg while 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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. Resolve the key's start index exactly like a normal lookup — the successor token position on the ring.
  2. Walk clockwise from there, reading the physical node of each token. Because vnodes interleave, consecutive tokens often belong to different machines, but not always.
  3. The if node not in result guard is the crucial correctness check: it skips any token whose physical node is already in the replica set, so we collect RF distinct machines, not RF tokens.
  4. Stepping with (i + 1) % n wraps 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.
  5. The loop also stops if the ring has fewer than RF physical nodes (the len(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
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. cap is the hard ceiling: 25% above the average of 2,000 keys/node, so 2,500. No node may hold more.
  2. 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.
  3. 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."
  4. Because total keys (10,000) fit under total capacity (5 × 2,500 = 12,500), the walk always finds room; the else on the for guards the impossible-over-capacity case.
  5. 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+ε)·avg regardless 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:1 as carrying 40% of traffic.
  • Split. Route reads to celebrity:1#{0..S-1}, each on a different node; writes fan out to all S.
  • Trade-off. Reads become scatter-gather across S nodes; writes cost .

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'}
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. A single key celebrity:1 maps 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.
  2. Key-splitting creates S derived keys celebrity:1#0 … celebrity:1#5, which hash to (typically) S different owners — spreading the read load across many nodes.
  3. hot_read picks 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.
  4. hot_write_targets shows the cost: a write must update all S shards to keep reads consistent, so writes are more expensive and must be applied to every shard atomically or with a versioning scheme.
  5. 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)
Enter fullscreen mode Exit fullscreen mode

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-1 distinct 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=3 into "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+ε)·avg cap 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 RF and 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 by S. 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

Practice →

Data Structures Topic — data-structures Ring and interval design problems

Practice →


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.

Iconographic token-ring diagram — a Cassandra/Dynamo style ring with tokens and replicas placed on the next distinct nodes clockwise, showing a preference list of replica nodes.

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 RF distinct 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 > N gives 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 (default Murmur3Partitioner) 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. SimpleStrategy walks the ring for RF nodes; NetworkTopologyStrategy places 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, ALL implement 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/ASK redirections. This sidesteps live-ring token math and makes the cluster state a compact 16384-entry map every node gossips.
  • Hash tags. {user1000}.followers and {user1000}.profile hash 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^32 ring, 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_tokens sets vnode count per node.
  • Keyspace. NetworkTopologyStrategy with per-DC replication factor.
  • Query. LOCAL_QUORUM for 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
Enter fullscreen mode Exit fullscreen mode
-- 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)
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. num_tokens: 16 gives each node 16 token ranges — enough vnodes, with the allocation algorithm, to keep ranges even without the metadata cost of the old 256 default.
  2. allocate_tokens_for_local_replication_factor: 3 tells Cassandra to choose token positions that balance ownership for RF=3 specifically, tightening distribution beyond random token placement.
  3. Murmur3Partitioner hashes the partition key (customer_id) to a 64-bit token; the node owning that token's range is the primary, and NetworkTopologyStrategy walks the ring for replicas in distinct racks.
  4. NetworkTopologyStrategy with dc-east: 3, dc-west: 3 places 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.
  5. LOCAL_QUORUM requires 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)
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. Without a hash tag, CRC16 of the whole key mod 16384 gives the slot; user:1000:profile and user:1000:followers almost certainly land in different slots — and thus possibly different nodes — so a multi-key MGET across them can fail with CROSSSLOT.
  2. Hash tags fix this: when a {...} is present, only the substring inside the braces is hashed. Both {user:1000}:profile and {user:1000}:followers hash the identical user:1000, so they share a slot and a node.
  3. 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.
  4. 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.
  5. 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.
Enter fullscreen mode Exit fullscreen mode

Step-by-step explanation.

  1. 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.
  2. 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.
  3. Q3 asks about multi-DC and consistency tuning. That is exactly what the Dynamo/Cassandra token ring plus NetworkTopologyStrategy and NWR quorums were built for.
  4. The fall-through — durable, no multi-key needs, single DC — is a plain token ring with vnodes and RF replication, the section 2–4 construction without the multi-DC machinery.
  5. 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
Enter fullscreen mode Exit fullscreen mode

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_tokens vnodes and NetworkTopologyStrategy, so replicas span racks and DCs; bootstrapping a node streams only its K/N ranges rather than reshuffling the cluster.
  • Redis fixed slots — decoupling the modulus (16384) from node count makes resize a bounded slot-migration with MOVED/ASK redirection, 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

Practice →

Hash Table
Topic — hash-table
Slot-mapping and hashing problems

Practice →


Cheat sheet — consistent hashing recipes

  • Remap-fraction rule. Modulo hashing moves about (N-1)/N of keys on any change to N — ~80% for 4→5, ~99% for 100→101. Consistent hashing moves only ~k/(N+k) when you add k nodes to N — 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_servers re-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_node does bisect.insort; get_node does i = bisect(pos, hash(key)) % len(pos) then returns owner[i]. The % len wrap is what makes it a ring — do not forget it. Lookup is O(log N).
  • Virtual-node sizing. Hash each node under V labels (node#0 … node#V-1) all mapping back to the physical node. Load imbalance falls like 1/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 V high so the realized split matches the intended weights (Cassandra exposes this as num_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 so RF=3 survives one zone outage. Tune consistency with W + 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+ε)·avg under 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-1 across nodes) — reads become scatter-gather, writes cost . 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}:profile and {user:1}:followers share a slot); resharding migrates whole slots and redirects clients with MOVED/ASK.
  • Cassandra template. num_tokens: 16 + allocate_tokens_for_local_replication_factor for even ranges; Murmur3Partitioner; NetworkTopologyStrategy with RF per DC; a high-cardinality partition key; LOCAL_QUORUM for 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.

Practice hash-table problems →
Practice database problems →

Top comments (0)