Replication, partitioning and sharding

When one database server is not enough, there are exactly two things to do with the data: copy it to more machines, or split it across more machines. Copying is replication; splitting is partitioning, which is called sharding when each part lives on a different server. Real systems do both, and each answers a different problem:

Knowing which problem you have is most of the decision. Read-heavy load on data that fits one machine wants replicas, not shards.

#Replication: one leader, several followers

The common arrangement: one leader accepts writes and ships its changes to followers, which serve reads.

Synchronous or asynchronous is the first choice. A synchronous follower must confirm each write before the client hears success — no data is lost if the leader dies, and every write waits for the slowest follower. An asynchronous follower is sent the change and the leader does not wait — writes are fast, and the follower is behind by some amount that is usually milliseconds and occasionally minutes.

Most production setups use asynchronous replication, sometimes with one synchronous follower so that at least two copies of every committed write exist. That makes replication lag a normal condition, and it produces three well-known anomalies:

• Reading your own write. A user updates their profile, the page reloads from a follower that has not received the change, and the edit appears to have vanished. Fix: read from the leader for a short time after a user writes, or for that user's own data.
• Going back in time. Two reads hit two followers with different lag; the second shows older data than the first. Fix: pin a user's reads to one follower.
• Stale reads for decisions. A follower says there is stock; the leader already sold it. Fix: make decisions on the leader. Replicas are for reads that may be slightly old, not for reads that authorise something.

Failover is the hard part. When the leader dies, a follower is promoted — but an asynchronous follower may be missing the last few writes, which are then lost, and if the old leader comes back believing it is still the leader, two nodes accept writes. That is split brain, and it is why failover needs fencing and consensus, which the CAP and consensus lesson takes up.

#Partitioning: which rows go where

To split data, every row needs a partition key, and a rule from key to partition. There are two rules.

Range partitioning — keys from A to F on partition 1, G to M on partition 2. Range queries are efficient, because adjacent keys are together. But a key that grows over time, like a timestamp, sends all new writes to the last partition, and a popular range makes one partition hot.

Hash partitioning — hash the key, and use the hash to choose a partition. Keys spread evenly regardless of their pattern. Range queries across keys now touch every partition.

#Hash mod N, and the day you add a node

The obvious hash rule is partition = hash(key) mod N. Here is what it does when a cluster grows from four nodes to five, for 100,000 keys:

hash mod N, 4 → 5 nodes:               80.2% of keys move   busiest node holds 1.02× the quietest

Perfectly even — and four keys in five change node. Adding capacity to a busy cluster means moving 80% of its data across the network at the moment it has the least headroom to do so. (The expected value is exactly 80%: a key stays put only if its hash gives the same remainder mod 4 and mod 5.)

Consistent hashing places nodes and keys on the same circle of hash values; a key belongs to the next node clockwise. Adding a node takes over only the arc before it. The same 100,000 keys, 4 → 5 nodes, with each node placed on the ring as 1, 10 or 100 virtual nodes:

consistent hashing,   1 vnodes, 4 → 5:   9.0% of keys move   busiest node holds 6.82× the quietest
consistent hashing,  10 vnodes, 4 → 5:  16.4% of keys move   busiest node holds 1.75× the quietest
consistent hashing, 100 vnodes, 4 → 5:  18.8% of keys move   busiest node holds 1.13× the quietest
of the keys that moved on the ring: 18848 went to the new node, 0 moved between old nodes

Two lessons are in those lines:

• Only the keys the new node needs move, and they move only to it. The ideal for a fifth node is a fifth of the keys, 20%. With 100 virtual nodes the ring moved 18.8%, all of it onto the new node, none between existing nodes.
• One position per node is badly uneven. With a single point each, the new node happened to land on a small arc, so only 9% moved — and the busiest node held nearly seven times what the quietest did. Virtual nodes, many points per physical node, are what make the ring even. Low movement with one vnode is not a success; it is the new node doing almost nothing.

Many systems avoid moving individual keys altogether: they create many more partitions than nodes up front — hundreds — and rebalance by moving whole partitions between nodes. Kafka topics and Elasticsearch indices both work this way: the number of partitions or primary shards is fixed when they are created. It makes rebalancing a matter of copying whole units, at the price of choosing the partition count before you know how big you will get.

#The hot partition

Hashing spreads keys evenly. It does nothing to spread traffic when one key is far busier than the rest. Partition by customer id, and let one customer be 30% of all requests, on five nodes:

traffic per node with one customer at 30%: 14.6% 43.2% 14.8% 14.2% 13.2%

The node holding the big customer carries 43% of traffic, three times its neighbours. No hash function fixes that, because every request for that customer has the same key. The options are all structural:

• Split the hot key — append a suffix (customer-big#0 … #9) so its data spreads over several partitions, and read from all of them. The cost is that everything for that customer is now a scatter-gather.
• Isolate it — give the largest tenants their own partition or their own cluster.
• Choose a different key — partition by order id rather than customer id, if the queries allow it.

#Queries that cross partitions

Partitioning's lasting cost is paid by queries. A query that includes the partition key goes to one partition. A query that does not must ask every partition and merge the results — slower, and as slow as the slowest partition.

The same applies to anything that needs several partitions at once: a join between rows on different shards, a unique constraint across shards, a transaction touching two. Each goes from one database feature to a distributed-systems problem. That is why the partition key should be chosen from the most important queries, and why the choosing-a-data-store lesson's advice — do not shard until the estimate says you must — is advice about all of these costs at once.