Partitioning and consistent hashing

Why did one machine do all the work?

The kitchen now cooks in Pune, Mumbai and Nashik. The read replicas from lesson 3.1 took the menu pages and tracking screens off the main database. Every write still lands on one primary, and its disk grows by about 40 GB a month.

So the orders table is split across four database machines. Each machine will own a quarter of the orders. That needs a rule for which machine an order belongs to, and the first rule chosen is the dish: hash(dish_id). The kitchens ask about dishes all day. How many thalis are left, how many dal orders came in. With this rule, each of those questions goes to one machine.

For three weeks the four machines share the load. Then comes festival morning, and the special thali. By noon it is 54% of every order in the three cities.

One machine holds the thali. Its disk queue is full and its orders take seconds to save. Machines one, three and four are nearly idle. Adding a fifth machine would change nothing: the thali can only live in one place.

A wedding caterer who splits four kitchens by dish, biryani in one and sweets in another, has the same problem when six hundred guests want biryani at once: one kitchen is overwhelmed and three stand idle. There is also a second cost. If the thali’s orders were spread over several machines to share the load, no single machine would know how many thalis are left.

Each machine was given a quarter of the menu, on the assumption that a quarter of the menu means a quarter of the work. On festival day one dish was more than half the work. The rule that splits the data also decides which requests land on the same machine. Before reading on, pick the key that would have spread the festival’s orders.

Pause and predict Which key would spread the festival’s orders and still answer each kitchen’s screen from one machine?

Copies spread reads; slices spread writes

Replication keeps full copies on several machines. That multiplies read capacity, but every copy must apply every write, so the write load and the data size per machine do not shrink at all.

To spread writes and storage, each machine must hold only part of the data. That is partitioning, also called sharding: the data is cut into slices, and each slice lives on one machine (usually with its own replicas).

Partition key

The field (or fields) whose value decides which partition a row belongs to. Every read or write that knows the key can go straight to one partition.

For example: With the key kitchen_id, all of kitchen 42’s orders live together, and the kitchen screen asks one machine.

Pause and predict Orders are placed on four machines by hash(dish_id). Kitchen 42’s screen asks for all of its orders today. How many of the four machines must it ask?

The dish rule failed in two ways at once, and both are the general problems of partitioning:

• Skew. A partition key with one hugely popular value gives that value’s partition most of the work. This is a hot partition, and adding machines does not help, because one key cannot be on two partitions.
• Questions that do not match the key. Any query that does not name the key must ask every partition and merge the answers (scatter-gather). With dish_id as the key, the kitchen screen (“kitchen 42’s orders today”) asks all four machines.

A mental model

The caterer’s story has a second half.

Two ways to map a key to a partition

• Range partitioning gives each partition a contiguous range of keys: orders from January in one, February in the next; or customer names A to F. Keys stay sorted, so range scans (“this week’s orders”) touch few partitions. The danger is sequential keys: if the key is a timestamp, every new write goes to the newest range, and one partition takes all of today’s load. Bigtable and HBase partition this way, splitting a range in two when it grows too big.
• Hash partitioning hashes the key first and maps the hash, rather than the key itself, to a partition. A good hash scatters neighbouring keys, so load spreads evenly even when keys arrive in order. The price is that neighbouring keys no longer sit together, so “orders between 12:00 and 12:30” must ask every partition. Cassandra and DynamoDB hash the partition key.

Many stores combine them: hash the first part of the key to pick the partition, and keep rows sorted by the rest within it. A Cassandra table keyed by (kitchen_id, created_at) hashes the kitchen and sorts each kitchen’s orders by time, so “kitchen 42, last hour” is one partition and one sorted range.

The festival day in numbers

20,000 orders from the simulator’s festival day, 54% of them the thali, placed on 4 partitions (an even share is 5,000 each):

Range by day answers every daily question from one partition, and that is the problem: one partition is doing 100% of today’s writes. It is a fine layout for an archive, and a bad one for a live table.

Adding a fifth machine: what moves?

Partitioning is not done once. Data grows and machines are added, and the data must rebalance. How much moves depends on how keys were mapped to machines.

Never assign rows to machines with hash(key) mod N, where N is the number of machines. Instead, either create many more partitions than machines up front and move whole partitions (Elasticsearch fixes the number of primary shards when an index is created; a Kafka topic’s partition count can only grow, and growing it changes which partition a key maps to), or split partitions as they grow (Bigtable splits a tablet when it gets too big; its paper gives about 100 to 200 MB per tablet by default), or use consistent hashing with virtual nodes (Cassandra’s default configuration file gives each node 16 tokens).

Which key for the orders?

Two keys are left standing, and they serve different goals: perfectly even writes, or kitchen screens that touch one machine.

Both keys spread the writes. The difference is which question has to ask every partition. Every kitchen’s screen asks for its orders over and over, all day. A customer fetching one order asks far less often, and the app can still go straight to one partition if the order reference includes the kitchen id. And the thali stock count does not belong in the orders table at all: it is one counter row, decremented with a conditional update (lesson 2.3), and the order rows carry the dish as an ordinary column.

Choosing a key is choosing groups: rows that one write touches together should share a partition. One way to find the groups: for every transaction, join the keys it touches into one set, as in union-find. Each set that results can live on one partition, and no transaction will ever need to commit across two partitions. Here every order touches one kitchen, so the kitchen is the group.

The monthly totals now ask all 64 partitions. They run once a month on a replica, so that cost is acceptable. A report that ran every second would be a reason to choose a different key.

What to remember

1. Replication copies everything and spreads reads; partitioning slices the data and spreads writes and storage. Real systems do both.
2. A partition key must spread load evenly and keep the common questions on one partition. Queries that do not name it become scatter-gather.
3. Range partitioning keeps order and makes scans cheap, but sequential keys pile onto one partition. Hash spreads evenly and loses order.
4. No hash fixes a hot key. Choose a different key, or salt the known-hot key and pay on reads.
5. Plan for rebalancing: many fixed partitions or virtual nodes move about 1/N of the data; mod N moves almost all of it.

Choose the partition key for the question asked most, then check it for one hot value.

In the real world

The simulator places one day of orders by write count. Real partitioning is weighed on more: row sizes, read traffic per partition, how big a partition may grow before it is split, and where each partition’s replicas live. Production systems also move partitions while serving traffic, which needs care: the old and new owner must agree on who answers during the move, a coordination problem lesson 3.4 takes up.

The orders are now spread by kitchen across four machines, each with replicas. Copies on several machines that can disagree raise the next question, in lesson 3.3: when two copies differ, what exactly can the system promise?