9 — Partitioning

Partitioning, or sharding, splits data across machines, so that each holds only part of it. It’s the main way to scale writes and storage past a single machine. It’s also a large step in complexity, so it should follow, not precede, the need. Bigger machines, replicas, caches and archiving old data all come first.

What it costs

Once data is split, several easy things become hard:

• Joins across shards are slow, or have to happen in the application.
• Transactions across shards need coordination.
• A unique constraint across shards is hard to enforce.
• Backups and schema changes have to run on every shard.

So the aim is to choose a partition key that keeps most operations on one shard.

Range partitioning

Range partitioning stores keys in order, with each shard holding one range: A to F, G to M and so on.

Range queries are efficient, because adjacent keys sit together. The danger is hot spots. If the key is a timestamp, all new writes go to the last shard. A common fix is to prefix the key with something that spreads the load, such as a sensor ID.

Hash partitioning

Hash partitioning applies a hash function to the key, then uses the result to pick a shard. This spreads keys evenly. It gives up range queries, because neighbouring keys end up on different shards.

The simplest version is hash(key) % N. This program measures what happens when one of four servers fails:

import hashlib
h=lambda k:int.from_bytes(hashlib.md5(k.encode()).digest()[:4],'big')
ks=[f'user_profile:{i}' for i in range(100000)]
moved = sum(h(k) % 4 != h(k) % 3 for k in ks)
print(f"Moved going from 4 to 3 servers: {100 * moved / len(ks):.1f}%")

Output: 75.0% of keys moved. Nearly everything is now on the wrong server. For a cache, that’s a mass miss. For a database, that’s moving most of the data. The cause is that every key’s result depends on $N$.

Consistent hashing

Consistent hashing avoids this by placing both servers and keys on a ring. A key belongs to the next server clockwise. Removing a server affects only the keys that were on it, and they move to the next server around the ring.

import hashlib
from bisect import bisect_right
from collections import defaultdict

class ConsistentHashRing:
    # A production-style Consistent Hash Ring using MD5 hashing and Virtual Nodes (vnodes)
    # to ensure uniform key distribution across physical storage/cache shards.
    def __init__(self, physical_nodes=None, vnodes_per_node: int = 100):
        self.vnodes_per_node = vnodes_per_node
        self.ring: list[int] = []               # Sorted list of vnode hash positions
        self.vnode_to_node: dict[int, str] = {} # Maps ring position -> physical node ID
        
        if physical_nodes:
            for node in physical_nodes:
                self.add_node(node)

    def _hash(self, key: str) -> int:
        # Returns a deterministic 32-bit integer hash for a given string.
        digest = hashlib.md5(key.encode("utf-8")).digest()
        return int.from_bytes(digest[:4], byteorder="big")

    def add_node(self, node_id: str) -> None:
        # Inserts a physical node into the ring via multiple virtual nodes.
        for i in range(self.vnodes_per_node):
            vnode_key = f"{node_id}#vnode{i}"
            position = self._hash(vnode_key)
            self.vnode_to_node[position] = node_id
            self.ring.append(position)
        self.ring.sort()

    def remove_node(self, node_id: str) -> None:
        # Removes all virtual nodes associated with a physical node.
        for i in range(self.vnodes_per_node):
            vnode_key = f"{node_id}#vnode{i}"
            position = self._hash(vnode_key)
            if position in self.vnode_to_node:
                del self.vnode_to_node[position]
                self.ring.remove(position)

    def get_node(self, item_key: str) -> str:
        # Finds the physical node responsible for the given key using binary search.
        if not self.ring:
            raise ValueError("Hash ring is empty.")
        
        item_hash = self._hash(item_key)
        # Find the first vnode position clockwise (>= item_hash)
        idx = bisect_right(self.ring, item_hash)
        
        # Wrap around to the start of the ring if we passed the highest vnode
        if idx == len(self.ring):
            idx = 0
            
        vnode_pos = self.ring[idx]
        return self.vnode_to_node[vnode_pos]

if __name__ == "__main__":
    shards = ["shard-alpha", "shard-beta", "shard-gamma", "shard-delta"]
    ring = ConsistentHashRing(shards, vnodes_per_node=150)

    # 1. Map 10,000 user records across the 4 shards
    initial_mapping = {}
    distribution = defaultdict(int)
    for user_id in range(10_000):
        key = f"user_profile:{user_id}"
        target = ring.get_node(key)
        initial_mapping[key] = target
        distribution[target] += 1

    print("Initial Key Distribution Across 4 Shards:")
    for shard, count in sorted(distribution.items()):
        print(f"  {shard}: {count} keys ({count / 100:.1f}%)")

    # 2. Simulate shard-gamma crashing
    ring.remove_node("shard-gamma")
    remapped_keys = sum(
        1 for key, old_shard in initial_mapping.items()
        if ring.get_node(key) != old_shard
    )
    print(f"\nKeys moved after removing 1 of 4 shards: {remapped_keys}/10000 ({remapped_keys / 100:.1f}%)")

Output:

Initial Key Distribution Across 4 Shards:
  shard-alpha: 2387 keys (23.9%)
  shard-beta: 2717 keys (27.2%)
  shard-delta: 2586 keys (25.9%)
  shard-gamma: 2310 keys (23.1%)

Keys moved after removing 1 of 4 shards: 2310/10000 (23.1%)

The keys that moved equal shard-gamma’s count exactly, so no other key was disturbed.

Virtual nodes

With one point per server, the ring is uneven, and when a server fails, all its load lands on one neighbour. Virtual nodes give each server many points. A simulation with 10 servers and 100,000 keys:

More points give a better balance, at the cost of a larger routing table. They also let a bigger server take more points, and they spread a failed server’s load across many neighbours.

Fixed slots

An alternative is a fixed number of slots, such as the 16,384 hash slots in Redis Cluster. Keys map to slots, and slots map to nodes. Resharding means moving slots, and the key-to-slot mapping never changes. This gives the same benefit with a simpler mental model.

Choosing a partition key

The choice matters a great deal, and it’s hard to change later.

A good key has high cardinality, an even distribution, and it matches the main queries.

Hot keys

No hash solves the case of one key getting a huge share of traffic. A celebrity with millions of followers is one key on one shard. Options:

• Cache it in front of the shard.
• Split it by adding a random suffix to writes, and reading from all the sub-keys.
• Move it to a dedicated shard.

Secondary indexes

Partitioning by user ID makes lookup by user easy. But how would a query find users by email? There are two approaches.

Scatter-gather queries suffer from the tail amplification in chapter 8: with 32 shards, there’s a 27.5% chance that at least one responds at its p99.

Resharding

Moving data between shards on a live system is one of the harder operations in practice. A common sequence is:

1. Set up the new shard.
2. Copy existing data, while also copying ongoing changes.
3. Compare the two copies.
4. Switch reads and writes over.
5. Delete the old copy.

Throttling matters, because the copying itself adds load. Many systems start with far more logical shards than machines, say 1,024 on 8 servers. Growth then means moving logical shards, rather than splitting them.

Finding the right shard

Something has to know which shard holds a key.

The map itself often lives in a consensus system such as ZooKeeper or etcd (chapter 11).

Generating unique IDs

Auto-increment doesn’t work across shards. Common alternatives:

Snowflake-style IDs use 41 bits for a millisecond timestamp, which lasts about 69.7 years from a chosen start date, plus 10 bits for the machine and 12 for a sequence. That allows 4,096 IDs per millisecond per machine, and IDs sort roughly by time. Clock skew between machines means the order is only approximate.

The takeaway

Partitioning buys scale at the price of joins, transactions and simplicity, so it’s worth delaying. Hash partitioning spreads load, range partitioning keeps order, and consistent hashing limits what moves. The partition key is the decision that lasts, so choose it around the main queries and expect hot keys.