The idea: moving partitions between nodes
A partitioned database splits its data over many nodes. Over time that has to change: more data needs more disks, more queries need more CPUs, and a failed machine's work must go somewhere. Moving data from one node to another is called rebalancing. A good rebalancing scheme does three things: afterwards the load is shared fairly, the database keeps serving reads and writes while data moves, and it moves no more data than it has to.
The first tab puts the same 40 keys through three strategies at once. Each key k01 … k40 has a hash h between 0 and 999. The hashes are chosen so that h mod 20 takes every value exactly twice, which makes the counts on the page exact.
How not to do it: hash mod N
The obvious rule is node = h(key) mod N. It spreads keys evenly, but it ties every key to the number of nodes. The book's example: a key with hash 123456 is on node 6 with 10 nodes (123456 mod 10), on node 3 with 11 and on node 0 with 12. When N changes, almost every key changes node.
Going from 4 to 5 nodes, a key stays only if h mod 4 = h mod 5. That is true for 4 of every 20 hash values, so 80% of the keys move. Most of them don't even go to the new node: they move between the old ones (Demo: add a 5th node, lane A: 32 of 40 keys move, only 8 of them to N4). Removing a node is just as bad (Demo: remove a node).
Fixed number of partitions
The fix is to create many more partitions than nodes when the database is set up, and give each node several of them. Lane B has 20 partitions on 4 nodes: partition p = h mod 20, and partition p starts on node p mod 4, as in the book's Figure 6-6. The assignment of keys to partitions never changes. Only the assignment of partitions to nodes changes.
When N4 joins, it takes a few partitions from every node until each has 4: p4 from N0, p9 from N1, p14 from N2 and p19 from N3. That is 4 of 20 partitions, 8 of 40 keys. When a node leaves, only its own partitions move. Riak, Elasticsearch, Couchbase, Voldemort and Cassandra (since 1.2) work this way.
The catch is choosing the number. The partition count fixed at setup is the most nodes the cluster can ever have, so it must allow for growth; but every partition has some management overhead, so too many is wasteful. A stronger machine can be given more partitions than a weak one.
Dynamic partitioning
Fixed partitions fit hash partitioning, where keys spread evenly. With key-range partitioning, fixed boundaries chosen badly would put all the data in one partition. So HBase and RethinkDB create partitions as the data grows. When a partition passes a size limit (HBase's default is 10 GB; on the page, 8 keys), it splits in two at its middle key, and one half can move to another node. If a partition shrinks, it can merge with a neighbour, much like a B-tree node.
The number of partitions follows the amount of data. But an empty database starts with a single partition, so until the first split every write goes to one node and the others sit idle. HBase and MongoDB let you pre-split an empty database into several ranges, which only works if you know in advance how the keys will be distributed (Demo: empty vs pre-split). MongoDB splits dynamically for hash-partitioned collections too.
Other strategies
Before version 1.2, Cassandra used consistent hashing with random boundaries and one large range per node. Adding a node meant splitting an existing node's range, an expensive operation that was hard to run in the background, and the load spread poorly. The usual answer is many small ranges per node (virtual nodes), which is the fixed-partitions idea again. See Consistent Hashing for the ring, virtual nodes and Redis Cluster's 16384 fixed slots, and Range vs Hash Partitioning for hot spots and range scans.
| Strategy | Key → node | Adding a node moves | Used by |
|---|---|---|---|
| hash mod N | h mod N | about N/(N+1) of all keys | nobody, on purpose |
| fixed partitions | h mod P → table | whole partitions, about 1/(N+1) of the data | Riak, Elasticsearch, Couchbase, Voldemort, Cassandra |
| dynamic partitioning | key range → table | whole ranges; splits as data grows | HBase, RethinkDB, MongoDB |
Request routing
Once partitions can move, a client asking for the key "Danube" needs to know which IP address and port to connect to. This is service discovery. There are three ways to answer it, and the second tab draws each hop (Demo: three ways to route):
- Any node forwards. The client connects to any node, for example through a round-robin load balancer. If that node doesn't own the partition, it forwards the request. Cassandra and Riak do this, and the nodes agree on the map by a gossip protocol.
- Routing tier. All requests go to a partition-aware load balancer that forwards them to the owner. MongoDB's
mongosand LinkedIn's Espresso (with Helix) work this way. - Partition-aware client. The client keeps the map itself and connects straight to the owner. It saves a hop, but its copy can go stale.
Whoever makes the routing decision must learn when partitions move, and everyone must agree. Many systems keep the authoritative partition → node table in a coordination service such as ZooKeeper (HBase, SolrCloud, Kafka, Helix), or in their own config servers (MongoDB). Routing tiers subscribe with a watch, so ZooKeeper notifies them when the table changes. DNS is enough to find the machines' IP addresses, which change far less often than partition ownership.
A move is not instant. While a partition is being copied, the old assignment still serves its reads and writes; only when the copy is complete does the table switch (Demo: a request during a move). A client with a cached map then sends to the old node, gets a "not owner" answer, re-reads the map and retries (Demo: stale client cache). Redis Cluster's MOVED redirect is this answer.
Automatic or manual rebalancing
Fully automatic rebalancing needs no operator, but it can be dangerous together with automatic failure detection. Rebalancing is expensive: it reroutes requests and copies a lot of data. In the third tab, a burst makes N2 slow for a few ticks. The failure detector declares it dead and its partitions move to the other nodes, which now serve more and also pay for the copying. They go over capacity, are declared dead in turn, and the load piles onto fewer and fewer nodes until the cluster is down (Demo: automatic → cascade).
That is why Couchbase, Riak and Voldemort compute a suggested assignment automatically but wait for an administrator to commit it. A human in the loop is slower, but here the burst ends, N2 recovers and nothing moves (Demo: manual, admin waits).
What the page leaves out
Real partitions have replicas, so a move copies from a replica and the old copy stays until the new one is caught up. Transfers are throttled so they don't swamp the network. Real systems use 64-bit hashes (Murmur3, MD5) and split by size in bytes or by load, not by key count. ZooKeeper itself runs a consensus protocol to keep its table consistent, and the routing tier, the nodes and the clients see changes with some delay.