The Project | About Us | Contribute | Donations | License
HOME My courses Completed
◄ Previous
◄

Data Partitioning and Consistent Hashing

Next ►
►

Module 3 ended with a sentence worth rereading: everything we know about replicating, coordinating and compensating "treats the data as if it were already sitting somewhere". This lesson opens Module 4 by answering the question that comes first: when Kilometre Zero's orders, the positions of van-3 or the events in orders.events are too many for a single node, how do you spread them across many? Replication copies the same data to several nodes so it survives failures; partitioning (or sharding) spreads different data across nodes so that no single one has to store or serve everything. We will look at the partitioning strategies (by range, by hash, compound), the problem that appears when the number of nodes changes, the classic solution of consistent hashing with virtual nodes, its cousin rendezvous hashing, and how a request finds the right node. All of this is the foundation on which HDFS, Cassandra and Redis Cluster are built, which we will cover in the rest of the module, and it is also a design decision Kilometre Zero has to make today: the partition key for its orders and for its stock.

Contents

  1. Partitioning versus replication, and how they combine
  2. Key-range partitioning and hot spots
  3. Key-hash partitioning and compound partitioning
  4. Partitioned secondary indexes: local versus global
  5. The problem with hash modulo N
  6. Consistent hashing: the ring and virtual nodes
  7. Rendezvous hashing
  8. Rebalancing partitions
  9. Request routing and partition discovery
  10. Choosing the partition key at Kilometre Zero
  11. Common Mistakes and Tips
  12. Exercises
  13. Conclusion

  1. Partitioning versus replication, and how they combine

The two mechanisms address different problems, and they are often confused because in practice they show up together:

Aspect Replication (03-04) Partitioning (this lesson)
What it copies The same data on several nodes Different data on each node
Problem it solves Availability, fault tolerance, nearby reads Volume and throughput: no single node can handle the whole dataset
If a node crashes Another node has a copy Its partition is lost… unless it is replicated
Main cost Consistency between copies Uneven distribution, queries that span partitions
Key question How many copies, and with what guarantees? Which piece of data goes to which node?

In a real system every partition is replicated: the dataset is split into partitions P1…Pn, and each partition has a leader and several followers (or a leaderless quorum). A physical node is usually the leader for some partitions and a follower for others, so both load and risk are spread out:

flowchart LR
    subgraph Node A
        A1[P1 leader]
        A2[P2 follower]
        A3[P4 follower]
    end
    subgraph Node B
        B1[P2 leader]
        B2[P3 follower]
        B3[P1 follower]
    end
    subgraph Node C
        C1[P3 leader]
        C2[P4 leader]
        C3[P2 follower]
    end
    subgraph Node D
        D1[P4 follower]
        D2[P1 follower]
        D3[P3 follower]
    end

The goal of partitioning is to spread the data and the load evenly. If the split is uneven, some partitions take far more load than others: these are hot spots, and one hot partition turns a 10-node system into a system with the capacity of 1. Everything that follows revolves around two questions: how to assign keys to partitions so as to avoid hot spots, and how to assign partitions to nodes so that adding or removing machines does not force you to move all the data.

  1. Key-range partitioning and hot spots

The most intuitive strategy is to sort the keys and give each partition a contiguous range, like the volumes of an encyclopaedia (A–C, D–F…). The boundaries need not be regular: they adapt to the density of the data so that every partition ends up a similar size.

Advantage: range queries are efficient, because neighbouring keys live on the same node. If the orders are keyed by date, "all of last week's orders" is answered by one or two partitions.

Drawback: ranges concentrate load whenever the access pattern concentrates on one region of the key space. Kilometre Zero lived through this during "Grape Harvest Week": with orders partitioned by created_at, every write of the day goes to the same partition (the one whose range contains "today"), while the partitions for earlier days sit idle. Ten nodes and only one of them writing.

Range key Query it favours Hot spot
created_at Orders by period Writes always land on "today's" partition
Alphabetical customer_id One customer's orders Customers whose names start with common letters
product_slug One product's orders crianza-wine during Grape Harvest Week

A common remedy is to prefix the key with something that scatters it: for example, market + date (Girona, Lleida, Tarragona and Valencia spread today's writes over four partitions). There are still four hot spots, but they are spread out; and the query "this week's orders in Valencia" is still a range. This anticipates the compound partitioning of the next section.

  1. Key-hash partitioning and compound partitioning

To get rid of the hot spots caused by key proximity, you apply a hash function to the key and partition by the result. A good hash (MD5, SHA-1, MurmurHash, xxHash; it does not need to be cryptographic, but it does need to be uniform and stable across languages and versions) turns similar keys (P-2026-000123 and P-2026-000124) into values that are far apart, so the consecutive orders of Grape Harvest Week land in different partitions.

import hashlib

def key_hash(key: str) -> int:
    """Stable 64-bit hash of a text key (the first 8 bytes of MD5)."""
    return int.from_bytes(hashlib.md5(key.encode()).digest()[:8], "big")

for order in ["P-2026-000123", "P-2026-000124", "P-2026-000125", "P-2026-000126"]:
    print(order, key_hash(order) % 4)

Each line shows the order and the partition (0–3) it is assigned to: consecutive orders end up scattered. It is important not to use Python's hash() for this: since Python 3.3 it has been randomised per process for strings, so two different services would compute different partitions for the same key. That is why we use hashlib.

The price of hashing is losing range queries: "orders between 10 and 14 September" no longer lives in any particular partition, so you have to ask all of them (scatter/gather, section 4).

Compound partitioning combines the two: one part of the key is hashed to choose the partition, and the rest is used to sort within the partition. Cassandra makes this explicit with the partition key and the clustering columns (04-04), but the idea is general:

  • Partition key: hash(customer_id) → all of Anna's orders are in the same partition.
  • Sort key: created_at DESC → within that partition they are sorted, so "Anna's latest 20 orders" is a sequential read on a single node.

The query "Anna's orders between two dates" is efficient; "all customers' orders from yesterday" is not. This is the essence of query-driven modelling: the partition key is chosen according to the query that matters most, not according to the domain model.

  1. Partitioned secondary indexes: local versus global

Orders are partitioned by customer_id, but the delivery team needs "all orders awaiting delivery in Girona", which mentions no customer at all. That calls for a secondary index (on market and status), and in a partitioned system an index has to be partitioned too. There are two ways of doing it:

Local index (document-partitioned) Global index (term-partitioned)
Where it lives Each partition indexes its own data The index is partitioned by the indexed value (market = Girona lives in one specific partition)
Write Only touches the partition holding the data: fast, atomic with the data Touches the data's partition and the index's partition: slower, often asynchronous (eventual)
Read through the index You have to ask every partition and merge the results: scatter/gather A single partition answers
Read latency That of the slowest partition (tail latency) Low and predictable
Examples Cassandra (local secondary indexes), Elasticsearch by default DynamoDB GSI, Cassandra materialised views

Scatter/gather deserves a closer look: with 12 partitions, the query fires 12 subqueries in parallel and waits for the slowest. With a 1% probability that a partition takes longer than 500 ms, the probability that the whole query exceeds that time is 1 − 0.99¹² ≈ 11%. This is the same phenomenon that made high percentiles so important in 01-03.

For delivery, Kilometre Zero chooses an asynchronous global index, materialised from the order.created and payment.confirmed events in the orders.events topic (02-05): an orders_by_market table partitioned by market, which may lag a few milliseconds behind. Courier assignment can tolerate that lag; order creation could not tolerate a synchronous write to two partitions.

  1. The problem with hash modulo N

With the hash in hand, the most obvious way of assigning keys to nodes is node = hash(key) % N. It works perfectly until the day N changes. When you go from 4 to 5 nodes, hash % 4 and hash % 5 agree only for keys whose hash leaves the same remainder in both, and that happens for roughly 1 key in 5: 80% of the data changes node. In a store holding terabytes, that means hours of network traffic, cold caches and, if the store is a cache, an avalanche hitting the database.

We measure it with simulations/hash_modulo.py, which generates 100,000 order identifiers in Kilometre Zero's format:

# km0/simulations/hash_modulo.py
"""How many order keys change node when going from 4 to 5 nodes with hash % N?"""
import hashlib

def key_hash(key: str) -> int:
    return int.from_bytes(hashlib.md5(key.encode()).digest()[:8], "big")

def modulo_node(key: str, n_nodes: int) -> int:
    return key_hash(key) % n_nodes

def measure_movement(keys, before: int, after: int) -> float:
    """Fraction of keys whose node changes when going from `before` to `after` nodes."""
    moved = sum(1 for k in keys if modulo_node(k, before) != modulo_node(k, after))
    return moved / len(keys)

if __name__ == "__main__":
    keys = [f"P-2026-{i:06d}" for i in range(1, 100_001)]
    for before, after in [(4, 5), (5, 6), (10, 11), (4, 8)]:
        pct = measure_movement(keys, before, after) * 100
        print(f"{before:>2} -> {after:>2} nodes: {pct:5.1f}% of keys move")

Step by step:

  • key_hash is the same stable function from section 3.
  • modulo_node is the naive assignment: the remainder of dividing the hash by the number of nodes.
  • measure_movement compares, key by key, the node before and after, and counts the ones that change.
  • The main block tries several changes in cluster size.

Typical output:

 4 ->  5 nodes:  80.0% of keys move
 5 ->  6 nodes:  83.3% of keys move
10 -> 11 nodes:  91.0% of keys move
 4 ->  8 nodes:  49.7% of keys move

Ideally you would move only what is strictly necessary: when adding a fifth node to four, just 1/5 of the keys (the ones that end up on the new node). Look at the 4 → 8 case: doubling the number of nodes moves "only" half, because hash % 8 preserves the low bit of hash % 4; it is a trick some systems use (growing by doubling), but it is rigid and still moves more than necessary.

  1. Consistent hashing: the ring and virtual nodes

Consistent hashing (Karger et al., 1997, originally devised for distributed web caches) solves the problem by decoupling the assignment from the number of nodes. The idea:

  1. The output space of the hash (say 0…2⁶⁴−1) is pictured as a ring: the maximum value is followed by 0.
  2. Each node is hashed too (by its name or address) and takes up a position on the ring.
  3. A key is assigned to the first node you come across when walking the ring clockwise from the key's position.
flowchart TB
    subgraph Ring["Hash ring (0 … 2^64−1, clockwise)"]
        direction LR
        N1(("inv-bcn<br/>pos 0x1A…"))
        N2(("inv-vlc<br/>pos 0x6F…"))
        N3(("inv-gir<br/>pos 0xB3…"))
        N1 --> N2 --> N3 --> N1
    end
    K1["aged-cheese<br/>hash 0x4C… → inv-vlc"] -.-> N2
    K2["pink-tomato<br/>hash 0x9E… → inv-gir"] -.-> N3
    K3["crianza-wine<br/>hash 0xE1… → inv-bcn (wraps around)"] -.-> N1

When you add a node at some position on the ring, only the keys between its predecessor and itself change owner (they move from the successor to the new node). When you remove a node, its keys go to its successor and nothing else moves. With N nodes, adding one moves 1/(N+1) of the keys on average: exactly the minimum.

The trouble with the basic ring is that, with few nodes, their random positions carve up the space very badly: one node may end up with 50% of the ring and another with 5%. And when a node is removed, all of its load falls on a single successor. The solution is virtual nodes (vnodes or tokens): each physical node is placed on the ring K times (inv-bcn#0, inv-bcn#1, …, inv-bcn#149), at different positions. With 100–200 vnodes per node the arcs average out, the distribution approaches uniform, and the load of a crashed node is shared among many different successors. They also let you give more vnodes to more powerful machines (weights).

simulations/consistent_ring.py implements it and measures it:

# km0/simulations/consistent_ring.py
"""Consistent hashing ring with virtual nodes."""
import bisect
import hashlib
import statistics
from collections import Counter

def key_hash(key: str) -> int:
    return int.from_bytes(hashlib.md5(key.encode()).digest()[:8], "big")

class ConsistentRing:
    def __init__(self, nodes=(), vnodes: int = 150):
        self.vnodes = vnodes
        self._positions: list[int] = []         # sorted positions on the ring
        self._owner: dict[int, str] = {}        # position -> physical node name
        for n in nodes:
            self.add_node(n)

    def add_node(self, node: str) -> None:
        for i in range(self.vnodes):
            pos = key_hash(f"{node}#{i}")
            if pos in self._owner:              # extremely rare collision: skip the vnode
                continue
            bisect.insort(self._positions, pos)
            self._owner[pos] = node

    def remove_node(self, node: str) -> None:
        for i in range(self.vnodes):
            pos = key_hash(f"{node}#{i}")
            if self._owner.get(pos) == node:
                del self._owner[pos]
                self._positions.remove(pos)

    def node_for(self, key: str) -> str:
        if not self._positions:
            raise RuntimeError("empty ring")
        h = key_hash(key)
        idx = bisect.bisect_right(self._positions, h)    # first vnode to the right
        if idx == len(self._positions):                   # past the end: wrap around
            idx = 0
        return self._owner[self._positions[idx]]

def distribution(ring: ConsistentRing, keys) -> Counter:
    return Counter(ring.node_for(k) for k in keys)

def relative_deviation(counts: Counter) -> float:
    """Standard deviation of the number of keys per node, relative to the mean (in %)."""
    values = list(counts.values())
    return statistics.pstdev(values) / statistics.mean(values) * 100

def fraction_moved(before: ConsistentRing, after: ConsistentRing, keys) -> float:
    return sum(1 for k in keys if before.node_for(k) != after.node_for(k)) / len(keys)

if __name__ == "__main__":
    keys = [f"P-2026-{i:06d}" for i in range(1, 100_001)]
    nodes = ["inv-bcn", "inv-vlc", "inv-gir", "inv-lle"]

    for vn in (1, 10, 150):
        ring = ConsistentRing(nodes, vnodes=vn)
        counts = distribution(ring, keys)
        print(f"vnodes={vn:>3}: {dict(counts)}  deviation={relative_deviation(counts):.1f}%")

    before = ConsistentRing(nodes, vnodes=150)
    after = ConsistentRing(nodes + ["inv-tar"], vnodes=150)
    print(f"add inv-tar (4->5): {fraction_moved(before, after, keys)*100:.1f}% of keys move")

    without_vlc = ConsistentRing(nodes, vnodes=150)
    without_vlc.remove_node("inv-vlc")
    print(f"remove inv-vlc (4->3): {fraction_moved(before, without_vlc, keys)*100:.1f}% of keys move")
    print("distribution after removing inv-vlc:", dict(distribution(without_vlc, keys)))

How it works, line by line:

  • _positions is a sorted list of integers (the positions of all the vnodes) and _owner says which physical node each position belongs to. Keeping the list sorted with bisect.insort lets us find the successor in O(log V).
  • add_node generates K positions by hashing name#i; the vnodes of a given node end up scattered all round the ring.
  • remove_node deletes exactly those positions; it touches nothing else, so the keys of the other nodes are unaffected.
  • node_for hashes the key, uses bisect_right to find the first position greater than the hash (the neighbour "to the right") and, if it runs off the end, goes back to position 0: that is the ring.
  • relative_deviation summarises how even the distribution is: 0% would be perfect.

Representative output (the exact values depend on the hashes):

vnodes=  1: {'inv-lle': 70171, 'inv-gir': 13875, 'inv-bcn': 14486, 'inv-vlc': 1468}  deviation=106.4%
vnodes= 10: {'inv-bcn': 32315, 'inv-gir': 32959, 'inv-vlc': 10845, 'inv-lle': 23881}  deviation=35.7%
vnodes=150: {'inv-bcn': 24275, 'inv-gir': 27571, 'inv-vlc': 24835, 'inv-lle': 23319}  deviation=6.3%
add inv-tar (4->5): 19.2% of keys move
remove inv-vlc (4->3): 24.8% of keys move
distribution after removing inv-vlc: {'inv-bcn': 32671, 'inv-gir': 37988, 'inv-lle': 29341}

Three conclusions. First: without vnodes, inv-lle takes 70% of the keys and inv-vlc 1.5% (a useless split, and every run with different node names would give another split just as arbitrary); with 10 vnodes it improves, and with 150 the deviation drops to about 6%, and keeps shrinking with more vnodes (with 500, 4.6%; with 1,000, 2.3%: the deviation falls roughly with the square root of the number of vnodes, which is why Cassandra used 256 per node for years). Second: adding the fifth node moves 19% of the keys (versus 80% with modulo), the theoretical minimum of 1/5. Third: removing one of four moves exactly its own keys (25%), and those keys are shared among the three survivors rather than falling on just one, because the 150 vnodes of inv-vlc had different successors.

This ring, with vnodes and with replicas on the next nodes round the ring, is the one used by Dynamo, Cassandra, Riak and the routing of many Memcached clients. We will see it applied in 04-04.

  1. Rendezvous hashing

There is an alternative to the ring that is simpler to implement and has no distribution problems: rendezvous hashing, or highest random weight (HRW). For each key you compute weight(node, key) = hash(node + key) for every node and pick the one with the highest weight. Properties:

  • When a node is removed, only the keys that had it as the winner change (they go to their second best): minimal movement, just like the ring.
  • When one is added, only the keys for which the new node wins move: 1/(N+1) on average.
  • No vnodes: the distribution is uniform by construction, because each key "draws lots" among all the nodes.
  • It gives you, for free, a list of nodes sorted by preference, which is handy for choosing the R replicas (the top R).
  • O(N) cost per lookup, versus O(log V) for the ring: perfect for tens of nodes, worse for thousands.
def rendezvous_node(key: str, nodes: list[str]) -> str:
    return max(nodes, key=lambda n: key_hash(f"{n}|{key}"))

It is used, among others, by the partitioning in some load balancers and caching systems. For Kilometre Zero, with fewer than twenty nodes per service, it would be a perfectly valid option; the ring is more common because it is what ships with the databases we are going to use.

  1. Rebalancing partitions

So far we have assigned keys directly to nodes. Real systems usually introduce an intermediate level: keys are assigned to partitions, and partitions to nodes. Moving whole partitions between nodes (rebalancing) is more manageable than moving individual keys. There are three schemes:

Scheme How it works Advantages Drawbacks Who uses it
Fixed number of partitions Many more partitions than nodes are created (e.g. 1,024 for 10 nodes); when a node is added, it "steals" a few whole partitions from each existing node Simple; only complete partitions move; allows weights You have to get the number right up front: too many = overhead; too few = a ceiling on growth Riak, Elasticsearch, Couchbase, Redis Cluster (16,384 slots, 04-05)
Dynamic partitioning A partition that exceeds a size (e.g. 10 GB) is split in two; one that empties out is merged with its neighbour Adapts to the volume; works with both range and hash A new dataset starts with 1 partition (a single node doing the work): mitigated with pre-splitting HBase, MongoDB, CockroachDB
Proportional to the nodes Each node has a fixed number of partitions (vnodes); adding a node randomly splits existing partitions Partition size stays stable as you grow Random splits, requires hashing Cassandra, Ketama

Two operational rules:

  • Rebalancing must be gradual and bandwidth-limited: moving partitions saturates the network and the disks of the nodes involved, just when they are also still serving traffic.
  • Automatic rebalancing is convenient but dangerous when combined with failure detection: a slow (not dead) node may be declared down, the system starts moving its partitions, the extra load makes other nodes look slow… a cascade. Many operators prefer the system to propose the rebalance and a human to approve it.

  1. Request routing and partition discovery

If catalog wants to read the stock of aged-cheese, which node does it connect to? This is the service discovery problem applied to partitions, and it has three answers:

flowchart LR
    subgraph a["(a) Any node"]
        C1[Client] --> N1a[Node 2]
        N1a -- forwards --> N2a[Node 4<br/>owner]
    end
    subgraph b["(b) Routing tier"]
        C2[Client] --> R[Router / proxy]
        R --> N2b[Node 4<br/>owner]
    end
    subgraph c["(c) Partition-aware client"]
        C3[Client<br/>knows the map] --> N2c[Node 4<br/>owner]
    end
Option Who knows the partition map Example Comment
(a) Any node Every node (gossip protocol) Cassandra, Riak The client is simple; one extra network hop in the worst case
(b) Routing tier The router (often with the help of a coordinator) mongos in MongoDB, moxi in Couchbase, Redis proxies The router can become a bottleneck; it has to be replicated
(c) Partition-aware client The client library (it downloads the map and caches it) Redis Cluster (MOVED), HBase, Kafka clients Maximum performance; the client must cope with stale maps

In every case the underlying problem is the same: all participants must agree on which partition lives on which node, and that agreement must survive failures. It is exactly the consensus problem of 03-03, which is why many systems delegate the map to a coordination service: ZooKeeper (HBase, classic Kafka, SolrCloud) or etcd (Kubernetes, CockroachDB). Nodes register in ZooKeeper/etcd, the routing tier subscribes to changes, and when a partition moves to another node the router finds out within milliseconds. Other systems (Cassandra, Riak) avoid the external dependency and spread the map by gossip, accepting that for a few seconds some nodes will have an outdated view.

For Kilometre Zero, which already uses etcd to elect the leader of the outbox relay (leader_relay.py from 03-03), the natural choice for its own partitions (the inventory replicas inv-bcn and inv-vlc, and any that follow) is to keep the map in etcd under /km0/inventory/partitions/ and have the gRPC clients read it and subscribe with watch. A sketch:

# km0/services/inventory/partition_map.py
import etcd3, json

client = etcd3.client(host="etcd", port=2379)
PREFIX = "/km0/inventory/partitions/"

def publish(partition: str, node: str, lease_ttl: int = 15):
    lease = client.lease(lease_ttl)                      # if the node dies, the entry expires
    client.put(f"{PREFIX}{partition}", json.dumps({"node": node}), lease=lease)
    return lease                                          # the node must call lease.refresh() periodically

def load_map() -> dict[str, str]:
    return {meta.key.decode().removeprefix(PREFIX): json.loads(v)["node"]
            for v, meta in client.get_prefix(PREFIX)}

def watch(on_change):
    """Calls `on_change(map)` every time a partition moves to another node."""
    for _ in client.watch_prefix(PREFIX)[0]:
        on_change(load_map())

The lease is the same mechanism as in 03-03: if inv-vlc stops renewing it, its entry disappears and the clients know the partition has no owner, with no need for a separate failure detector.

  1. Choosing the partition key at Kilometre Zero

With all of the above, we can make the two decisions the team has pending. Remember that the partition key is chosen according to the dominant queries and the load distribution, and that it can be complemented with asynchronous global indexes for the secondary queries.

Orders (km0_orders, 40,000 orders/day during a campaign; dominant read: "my orders" in the app, and "order by id" on the confirmation page):

Candidate For Against Verdict
created_at (range) Queries by period for analytics Permanent hot spot on "today" (Grape Harvest Week) Rejected
order_id (hash) Perfect distribution; "order by id" in one hop "My orders" is a scatter/gather over all partitions Only for the orders_by_id table
customer_id (hash) + created_at DESC (sort) "My orders" on one node, already sorted; even distribution (thousands of customers) A customer with a huge number of orders (a restaurant) creates a large partition; "order by id" needs to know the customer Chosen as the main key
market (hash) delivery queries by city Only 4 values: 4 partitions at most, with Valencia twice the size Rejected as a key; kept as the asynchronous global index orders_by_market
producer_id Producer dashboard Very uneven (La Vega Farm has 30 times more orders than Roble Alto Winery) Asynchronous global index

Decision: the partition key for orders is customer_id, clustered by created_at DESC; a second table orders_by_id is maintained (the "one table per query" pattern we will develop in 04-04), fed by the same write, and the indexes by market and by producer are materialised from orders.events. For the customer with too many orders, a time bucket is added to the key (customer_id + year-month), so that no partition grows without bound: it is the compound partitioning of section 3 taken into the partition key itself.

Stock (km0_inventory, replicas inv-bcn/inv-vlc, counters decremented by stock.reserved):

Candidate For Against Verdict
product_slug (hash) Each counter on a single node: atomic decrements with no cross-node coordination aged-cheese during Artisan Cheese Week is a hot key Chosen; the hot key is handled with queues and with a cache (04-05), not by changing the partition
producer_id A producer updates all of its stock on one node La Vega Farm concentrates 40% of the catalogue in one partition Rejected
market Stock by city for the website Stock belongs to the producer, not the market: it would duplicate counters and require cross-partition transactions Rejected

The important lesson here is that a counter that must be consistent (CP, 03-02) must live entirely within one partition: decrementing stock that is spread over two nodes would require 2PC (03-05) on every reservation. That is why product_slug wins even though it has hot keys.

Common Mistakes and Tips

  • Using Python's hash() (or the default hashCode of other languages) as the partitioning hash. It is randomised per process or implementation-dependent. Use an explicit, documented hash (MD5, MurmurHash3, xxHash) and pin its version.
  • Confusing partitioning with replication. Partitioning without replication reduces availability: every node that crashes takes its share of the data with it. Each partition needs its replication factor (04-04).
  • Choosing the partition key from the domain model rather than from the queries. "Orders have an id, so the key is order_id" produces a scatter/gather on the most frequent query. Start by listing the queries and how often they run.
  • Keys with few distinct values (market, status, type). They cap the number of partitions and create imbalances. High cardinality first.
  • Ignoring partitions that grow without bound (the customer with a million orders, the delivery fleet with 2.4 million positions a day). Add a time bucket to the key.
  • Consistent hashing without virtual nodes. Uneven distribution and, when a node crashes, all its load on a single successor. Use 100–200 vnodes, or rendezvous hashing.
  • Aggressive automatic rebalancing. It can trigger cascades when a node is merely slow. Limit the bandwidth and consider manual approval.
  • Forgetting that the partition map is distributed state. It needs consensus (ZooKeeper/etcd) or gossip, and clients must tolerate stale maps (retrying after a MOVED or equivalent).

A final tip: when you are torn between range and hash, ask yourself whether the range query is really needed on the hot path or whether a global index or the data lake (04-02) can serve it offline. It is almost always the latter.

Exercises

Exercise 1. Modify consistent_ring.py to support weights: add_node(node, weight=1.0) must create int(vnodes * weight) virtual nodes. Build a ring with inv-bcn (weight 2.0), inv-vlc (1.0) and inv-gir (1.0), distribute the 100,000 keys and check that inv-bcn receives roughly 50%. What happens to remove_node if you do not store the weight?

Exercise 2. The couriers' positions (140 vans like van-3, 2.4 million positions a day between them) are to be stored partitioned. The queries are: (a) "latest position of courier R" (thousands per minute, from the customer-facing website), (b) "route of courier R between two times today" (support), (c) "all couriers in Valencia right now" (operations dashboard). Propose the partition key (and sort key where applicable), state which query is left as a scatter/gather or a global index, and explain why a bare courier_id produces an unbounded partition and how you fix it.

Exercise 3. Implement rendezvous_node(key, nodes) and a function rendezvous_replicas(key, nodes, r) that returns the r nodes with the highest weight. With 5 nodes and 100,000 keys, measure (a) the relative deviation of the distribution, (b) the fraction of keys whose primary node changes when a sixth node is added, and (c) the fraction of keys whose set of 3 replicas changes. Compare (b) with the ring's result.

Solutions

Solution 1:

class WeightedConsistentRing(ConsistentRing):
    def __init__(self, vnodes: int = 150):
        super().__init__(vnodes=vnodes)
        self._weights: dict[str, int] = {}        # node -> actual number of vnodes created

    def add_node(self, node: str, weight: float = 1.0) -> None:
        n = max(1, int(self.vnodes * weight))
        self._weights[node] = n
        for i in range(n):
            pos = key_hash(f"{node}#{i}")
            if pos not in self._owner:
                bisect.insort(self._positions, pos)
                self._owner[pos] = node

    def remove_node(self, node: str) -> None:
        for i in range(self._weights.pop(node, 0)):
            pos = key_hash(f"{node}#{i}")
            if self._owner.get(pos) == node:
                del self._owner[pos]
                self._positions.remove(pos)

ring = WeightedConsistentRing()
ring.add_node("inv-bcn", 2.0); ring.add_node("inv-vlc"); ring.add_node("inv-gir")
print(distribution(ring, keys))   # inv-bcn ≈ 50,000, the others ≈ 25,000 each

If remove_node recomputes using self.vnodes instead of the actual number, for inv-bcn it would delete only 150 of its 300 vnodes and leave 150 positions on the ring pointing at a dead node: the keys landing on them would go to a node that does not exist. That is why you have to store how many vnodes were created per node (or walk through them from _owner).

Solution 2:

Partition key courier_id + day (for example van-3|2026-09-14), with sort key ts DESC. (a) "Latest position" is the first row of today's partition: one node, minimal read. (b) "Route between two times" is a range within the same partition, sorted by time. (c) "All couriers in Valencia right now" mentions no courier: a scatter/gather over 140 partitions would be acceptable (140 small reads), but an asynchronous global index latest_position_by_market is better, updated with every position (or every N seconds) and partitioned by market, which is what the dashboard actually shows. With a bare courier_id, van-3 would pile up some 17,000 positions a day without limit (more than 6 million a year in a single partition, which in Cassandra would start to degrade reads and compactions); the daily bucket bounds the partition and also makes deleting old data trivial (you drop the whole day's partition). This data was AP in the table in 03-02, and nothing above changes that: with W=1 the position is accepted even if replicas are missing.

Solution 3:

def rendezvous_node(key, nodes):
    return max(nodes, key=lambda n: key_hash(f"{n}|{key}"))

def rendezvous_replicas(key, nodes, r):
    return sorted(nodes, key=lambda n: key_hash(f"{n}|{key}"), reverse=True)[:r]

nodes5 = ["n1", "n2", "n3", "n4", "n5"]; nodes6 = nodes5 + ["n6"]
counts = Counter(rendezvous_node(k, nodes5) for k in keys)
print("deviation:", round(relative_deviation(counts), 2), "%")               # ≈ 0.3-0.6% without vnodes
moved = sum(rendezvous_node(k, nodes5) != rendezvous_node(k, nodes6) for k in keys) / len(keys)
print("primary changes:", round(moved * 100, 1), "%")                        # ≈ 16.7% (= 1/6)
moved_r = sum(set(rendezvous_replicas(k, nodes5, 3)) != set(rendezvous_replicas(k, nodes6, 3))
              for k in keys) / len(keys)
print("replica set changes:", round(moved_r * 100, 1), "%")                  # ≈ 50% (= 3/6)

(a) The deviation comes out at around 0.3%, with no need for virtual nodes and better than the ring's 6% with 150 vnodes, because each key chooses among all the nodes independently (the ring, by contrast, depends on how evenly the vnode positions happen to fall). (b) The primary node changes for 1/6 of the keys, the same as the ring with vnodes (the theoretical minimum) and a very long way from the 83% of modulo. (c) With 3 replicas, the set changes for roughly half the keys (3/6): the new node makes it into the top three with probability 3/6, and in each case one replica moves, not all three; the volume of data moved is still minimal (each key moves at most one copy), even though the fraction of affected sets is larger.

Conclusion

Partitioning means spreading different data across nodes, and replication means copying the same data; real systems do both, replicating every partition. We have seen that range partitioning preserves interval queries but concentrates load (the orders of Grape Harvest Week all writing to "today's" partition), that hash partitioning scatters the keys at the expense of those ranges, and that compound partitioning (hash of the customer, sorted by date) recovers the best of both within each partition. Secondary indexes force a choice between local indexes with scatter/gather and asynchronous global indexes, and Kilometre Zero has chosen the latter for delivery, feeding them from orders.events. The naive hash % N assignment moves 80% of the data when going from 4 to 5 nodes, as hash_modulo.py measured; consistent hashing cuts that to the 20% minimum, and the virtual nodes of ConsistentRing bring the deviation of the distribution down from over 100% to around 6% (and lower the more vnodes you use) and share the load of a crashed node among all the others. Rendezvous hashing achieves an even more uniform distribution with neither ring nor vnodes. Above the keys, systems move whole partitions (fixed, dynamic or proportional to the nodes) and publish the partition map through gossip or a coordinator such as ZooKeeper or etcd, which we already knew from the leader election in 03-03. And we have settled two Kilometre Zero decisions: orders partitioned by customer_id (with a monthly bucket and a secondary orders_by_id table) and stock partitioned by product_slug, because a CP counter must live entirely within one partition.

With the keys distributed, it is time to look at the systems that store them. The next three topics are three different ways of storing bytes at scale: files, objects and database records. We start with the oldest, and the one closest to what you already know: distributed file systems, with NFS, HDFS and Ceph, and with HDFS as the data lake where Kilometre Zero will keep the orders.events events that Module 5 will process in bulk.

◄ Previous
◄
Next ►
►

Distributed Architectures Course

Module 1: Introduction to Distributed Systems

Module 2: Communication in Distributed Systems

Module 3: Consistency and Replication

Module 4: Distributed Storage

Module 5: Distributed Computing

Module 6: Security in Distributed Systems

Module 7: Monitoring and Maintenance

Module 8: Case Studies and Applications

The Project | About Us | Contribute | Donations | License
© Copyright 2026. All rights reserved