Rebalancing and Resharding

In 1997, David Karger and his colleagues at MIT published a paper that would fundamentally reshape how distributed systems handle data placement. Consistent hashing—the algorithm they introduced—solved a problem that had plagued distributed systems: how to redistribute data when servers are added or removed without requiring massive data movement.

The elegance of consistent hashing lies in its guarantee: when a cluster changes from N nodes to N+1 nodes, only 1/(N+1) of the keys need to be redistributed, rather than nearly all of them. This property transforms rebalancing from an expensive, disruptive operation into a manageable, incremental process.

Before understanding why consistent hashing matters, we must understand why naive approaches fail. Consider the simplest approach to distributing data across N servers:

Modulo Hashing:

server = hash(key) % N

This approach seems reasonable: hash the key to get a number, take modulo N to get a server index. Data is evenly distributed (assuming a good hash function), and lookups are O(1).

The Catastrophic Failure Mode:

The problem emerges when the cluster size changes. Consider a cluster with 3 servers:

Now add one server (N=4). The same keys map to different servers:

The Mathematics of Modulo Redistribution:

When changing from N to N+1 servers, the probability that a key stays on the same server is:

P(same server) = 1/(N+1)  (approximately)

This means nearly all keys must move. For a 3-server to 4-server transition:

• Expected keys to move: ~75%
• For 10→11 servers: ~91% of keys move
• For 100→101 servers: ~99% of keys move

This is catastrophic for large-scale systems. Adding a single server to a 100-node cluster storing 100 TB of data would require moving ~99 TB of data.

The Real-World Impact:

• Extended Downtime: Moving petabytes of data takes hours or days
• Network Saturation: Bulk data transfer overwhelms network capacity
• Cache Invalidation: All caches become invalid when keys relocate
• Inconsistency Windows: During migration, data exists in multiple places

Consistent hashing solves the redistribution problem by fundamentally changing how we think about key-to-server mapping. Instead of using modulo arithmetic, we conceptualize both keys and servers as points on a circle (ring).

The Hash Ring Concept:

1. Define a Ring: Conceptualize the hash space as a ring (0 to 2³² - 1, wrapping around)
2. Place Servers on Ring: Hash each server identifier to get its position on the ring
3. Place Keys on Ring: Hash each key to get its position on the ring
4. Assign Keys to Servers: Each key is assigned to the first server encountered when walking clockwise from the key's position

Visual Representation:

                    0°
                    |
            S3 ●----+----● S1
               /         \
              /           \
             /             \
   270° ----●               ●---- 90°
            |       K1 ✕    |
            |     K2 ✕      |
             \      K3 ✕   /
              \           /
               \         /
            S2 ●---------●
                    |
                   180°

Keys K1, K2, K3 are all assigned to Server S2
(the first server clockwise from each key's position)

The Key Insight: Minimal Redistribution

When a server is added or removed, only the keys between the affected server and its predecessor need to move:

Adding Server S4:

Before:     ... ---[ S2 ]-----[ S3 ]--- ...
            Keys K1, K2, K3 belong to S3

After:      ... ---[ S2 ]--[ S4 ]--[ S3 ]--- ...
            Keys K1, K2 now belong to S4
            Key K3 still belongs to S3

Only keys between S2 and S4 need to move from S3 to S4. All other keys remain unchanged.

Mathematical Guarantee:

When changing from N to N+1 servers:

Expected keys to move = K / (N + 1)

Where K = total number of keys

This is a dramatic improvement:

• Moving from 3→4 servers: ~25% of keys move (vs. ~75% with modulo)
• Moving from 10→11 servers: ~9% of keys move (vs. ~91% with modulo)
• Moving from 100→101 servers: ~1% of keys move (vs. ~99% with modulo)

While basic consistent hashing solves the redistribution problem, it introduces a new challenge: uneven load distribution. With only a few physical servers on the ring, some servers may own larger portions of the key space than others simply due to the random nature of hash function outputs.

The Imbalance Problem:

Consider 3 servers placed on a ring by hashing their identifiers:

Server A: position 10°
Server B: position 170°
Server C: position 200°

Key space distribution:
- Server A: 170° (from 200° to 10°, wrapping around)
- Server B: 160° (from 10° to 170°)
- Server C: 30° (from 170° to 200°)

Server A handles 5.7x more keys than Server C!

The Virtual Node Solution:

Instead of placing each physical server once on the ring, place it multiple times using different hash inputs:

Physical Server A → Virtual Nodes A-1, A-2, A-3, ..., A-100
Physical Server B → Virtual Nodes B-1, B-2, B-3, ..., B-100
Physical Server C → Virtual Nodes C-1, C-2, C-3, ..., C-100

Each virtual node is hashed to its own position on the ring. A key is assigned to the virtual node encountered first clockwise, which maps to a physical server.

Virtual Nodes in Practice:

Benefits Beyond Load Balancing:

1. Heterogeneous Hardware: Assign more virtual nodes to more powerful servers

   • Server with 64GB RAM: 200 virtual nodes
   • Server with 32GB RAM: 100 virtual nodes

2. Graceful Addition/Removal: When adding a server, its virtual nodes are spread across the ring, taking a small portion from each existing server rather than a large portion from one

3. Finer Replication Control: Each virtual node can be replicated independently, enabling rack-aware or zone-aware placement

Implementation Considerations:

def get_virtual_nodes(server_id, num_vnodes=100):
    virtual_nodes = []
    for i in range(num_vnodes):
        vnode_id = f"{server_id}:vnode-{i}"
        position = hash(vnode_id) % RING_SIZE
        virtual_nodes.append((position, server_id))
    return virtual_nodes

def build_ring(servers, vnodes_per_server=100):
    ring = []
    for server in servers:
        ring.extend(get_virtual_nodes(server, vnodes_per_server))
    ring.sort(key=lambda x: x[0])  # Sort by position
    return ring

Implementing consistent hashing correctly requires attention to several details: efficient lookup, atomic updates, and proper handling of edge cases.

Core Data Structures:

The ring is typically implemented as a sorted array or balanced tree of (position, server) pairs:

class ConsistentHashRing:
    def __init__(self, nodes=None, vnodes=100):
        self.vnodes = vnodes
        self.ring = []  # Sorted list of (position, node_id)
        self.nodes = set()
        
        if nodes:
            for node in nodes:
                self.add_node(node)
    
    def _hash(self, key):
        """Hash function: use a cryptographic hash for uniform distribution"""
        import hashlib
        return int(hashlib.md5(key.encode()).hexdigest(), 16) % (2**32)
    
    def add_node(self, node_id):
        """Add a node to the ring with its virtual nodes"""
        if node_id in self.nodes:
            return
        
        self.nodes.add(node_id)
        for i in range(self.vnodes):
            vnode_key = f"{node_id}:vnode:{i}"
            position = self._hash(vnode_key)
            self.ring.append((position, node_id))
        
        self.ring.sort(key=lambda x: x[0])
    
    def remove_node(self, node_id):
        """Remove a node and all its virtual nodes from the ring"""
        if node_id not in self.nodes:
            return
        
        self.nodes.discard(node_id)
        self.ring = [(pos, nid) for pos, nid in self.ring if nid != node_id]

Optimization Techniques:

1. Binary Search: Use binary search for O(log n) lookup instead of O(n) linear scan
2. Caching: Cache ring structure locally; invalidate on membership changes
3. Read-Copy-Update: Create new ring on changes; atomically swap pointer
4. Ring Slicing: For extreme scale, partition the ring itself across coordinators

Thread Safety:

import threading

class ThreadSafeConsistentHashRing:
    def __init__(self):
        self._ring = ConsistentHashRing()
        self._lock = threading.RWLock()  # Conceptual; use appropriate primitive
    
    def get_node(self, key):
        with self._lock.read_lock():
            return self._ring.get_node(key)
    
    def add_node(self, node_id):
        with self._lock.write_lock():
            self._ring.add_node(node_id)

Consistent hashing naturally extends to support replication by walking further around the ring to find additional replica locations.

The N Replicas Pattern:

For replication factor N, a key is stored on N consecutive distinct physical servers walking clockwise:

Key K at position 100°
Replication factor N = 3

Walk clockwise, collecting distinct physical servers:
- Position 110°: Virtual node of Server A → Server A (replica 1)
- Position 125°: Virtual node of Server A → Skip (already have A)
- Position 140°: Virtual node of Server B → Server B (replica 2)
- Position 155°: Virtual node of Server C → Server C (replica 3)

Key K is replicated to Servers A, B, C

Rack-Aware Replica Placement:

In production environments, replicas should span failure domains (racks, availability zones):

def get_replicas_rack_aware(key, ring, replication_factor, rack_map):
    """
    Select replicas ensuring they're on different racks.
    rack_map: {server_id: rack_id}
    """
    key_hash = ring._hash(key)
    start_idx = ring._find_position(key_hash)
    
    replicas = []
    racks_used = set()
    idx = start_idx
    
    while len(replicas) < replication_factor:
        node_id = ring.ring[idx][1]
        rack = rack_map.get(node_id)
        
        # Only add if this is a new rack (or we've exhausted rack diversity)
        if rack not in racks_used or len(racks_used) >= available_racks:
            if node_id not in replicas:
                replicas.append(node_id)
                racks_used.add(rack)
        
        idx = (idx + 1) % len(ring.ring)
        
        # Prevent infinite loop if we can't find enough diversity
        if idx == start_idx:
            break
    
    return replicas

Preference Lists:

Systems like Dynamo maintain preference lists—ordered lists of nodes that should store each key. The first N healthy nodes in the preference list hold the replicas:

Preference list for key K: [A, B, C, D, E]
Replication factor: 3

Normal case: Replicas on A, B, C
If A is down: Replicas on B, C, D (D temporarily holds A's replica)
If A and B are down: Replicas on C, D, E

The true power of consistent hashing emerges when cluster membership changes. Properly handling additions and removals is crucial for maintaining availability and data integrity.

Adding a New Node:

When a new node joins the cluster:

1. Calculate Virtual Node Positions: Hash the new node's virtual node identifiers
2. Identify Affected Key Ranges: Determine which key ranges will map to the new node
3. Initiate Data Transfer: Request affected data from current owners
4. Accept Traffic Gradually: Start serving reads, then writes for migrated data
5. Complete Integration: Remove data from previous owners after confirmation

Removing a Node (Planned):

Planned removal (decommissioning) follows an orderly process:

1. Mark Node for Removal: Set node status to 'leaving'
2. Stop Accepting New Data: Redirect writes to successor nodes
3. Transfer Data: Stream owned data to successors on the ring
4. Verify Transfer: Confirm all data is safely replicated
5. Update Ring: Remove node from the ring
6. Shutdown: Stop node processes

Removing a Node (Failure):

Unplanned removal requires different handling:

1. Failure Detection: Health checks determine node is unresponsive
2. Declare Node Dead: After timeout (typically 30-60 seconds)
3. Activate Replicas: Next nodes in preference lists take over
4. Repair Data: Restore replication factor from remaining replicas
5. Update Ring: Remove failed node after sustained outage
6. Handle Recovery: If node returns, resolve any conflicts

Key Insight: Minimal Movement

In both cases, only data owned by the changing node moves:

Adding Node X with 100 virtual nodes to a 10-node cluster:
- Each existing node loses ~10% of its data to X
- Total data movement: ~10% of entire dataset
- Compare to modulo: ~100% would move

Consistent hashing underpins many of the world's most scalable data systems. Understanding how production systems apply these concepts provides practical insights.

Amazon DynamoDB:

DynamoDB, inspired by the original Dynamo paper, uses consistent hashing as its core distribution mechanism:

• Partition Key Hashing: The partition key is hashed to determine placement
• Virtual Nodes: Each partition is divided across multiple storage nodes
• Automatic Rebalancing: When capacity is exceeded, partitions split automatically
• Adaptive Capacity: Hot partitions can burst beyond their provisioned throughput

Apache Cassandra's Token Ring:

Cassandra's implementation is particularly instructive:

1. Token Assignment: Each node is assigned a set of tokens (positions on the ring)
2. Partitioner: The partitioner (Murmur3Partitioner) hashes partition keys to tokens
3. Replication Strategy: NetworkTopologyStrategy ensures replicas span racks and data centers
4. Consistent Hashing Lookup: Coordinators locate data by checking which nodes own the token range

-- Cassandra keyspace with replication
CREATE KEYSPACE my_app WITH replication = {
    'class': 'NetworkTopologyStrategy',
    'us-east': 3,
    'eu-west': 3
};

Memcached and Redis Clusters:

Cache clusters also use consistent hashing:

• Memcached: Client libraries (libmemcached, various language bindings) implement consistent hashing client-side
• Redis Cluster: Uses hash slots (16384 slots) assigned to nodes; slots map to nodes rather than pure consistent hashing

CDN Edge Servers:

Content Delivery Networks use consistent hashing to route requests to edge servers, ensuring the same content is cached on predictable servers.

Consistent hashing is the algorithmic foundation that makes modern distributed database rebalancing possible. The key insights from this page:

What's Next:

With the algorithmic foundation established, the next page explores the challenges of online resharding—the complexities that arise when resharding must occur while the system continues serving traffic. We'll examine consistency challenges, coordination protocols, and failure scenarios that make online resharding one of the most demanding operations in distributed systems.