Too Much Data for One Box

Replicas fixed reads. They did nothing for two other things, both of which have now arrived at once, and neither of which a copy can help with.

The first is that the data is too big. Courtyard’s messages, every message anyone has ever sent, kept forever because people scroll back, have outgrown the largest disk you can rent. It is not the load this time but the size: Chapter 1’s wall, for storage.

The second is that writes are slower than they were. Every write goes to the primary and then to every replica, each of which applies it. Five replicas means every message is written six times, on six disks. Adding readers made writing more expensive. Chapter 3’s medicine has a side effect, and it is the one thing replicas cannot address at all, because a copy has to hold everything and apply everything. There is only one way to hold more than one machine can hold and write more than one machine can write, and that is to stop having every machine hold everything.

So you split the data, not copied this time but divided. Machine A holds some of the messages, machine B holds others, machine C the rest. Each holds a third of the data, on a disk a third the size, and each takes a third of the writes. Add a fourth machine and each holds a quarter. Capacity and write throughput both grow with the number of machines, which is what copies could never do.

The field calls this partitioning, and each machine’s piece a shard. The underlying move is one you have made three times already: impose a structure on the names so that the middle can route on a prefix. Part I Chapter 5 split addresses by network. Chapter 20 split resources by path. Part II Chapter 4 split names by suffix. Now you split data by some part of its identity, and that part is called the shard key. Choosing it is the entire chapter.

Everything about a partitioned system follows from which piece of each record decides where it lives, so let’s try the options against Courtyard’s actual queries.

Suppose you shard by user. Everything belonging to Mira, her messages, her photos, her likes, lives on the shard her user id maps to. Look at what is fast. Her profile page, which needs all her things, touches one machine. The application computes which shard from her id, asks it, and is done. That is the query Courtyard runs most, and it is as fast as it was on one box.

Then someone famous signs up, with a million followers. Every one of their posts is read a million times, and every read goes to their shard. That one machine takes a thousand times the load of the others and melts, while the other shards sit idle. The data is evenly divided but the load is not. This is a hot shard, and its cause is a hot key, a single value of the shard key that is vastly more popular than the rest. Any key that maps a natural entity to a machine will have this problem, because popularity is never even.

Suppose instead you shard by post. Give each message and photo a random-ish id and shard on that. Now the load is perfectly even. A celebrity’s million posts are scattered across every machine and each machine takes a fair share, and the hot shard is gone.

But Mira’s profile page, which asks for all of Mira’s photos, now touches every shard, because her photos are everywhere. The application has to ask all N machines, wait for all N to answer, and merge. This is scatter-gather, and it has a nasty property: the page takes as long as the slowest of the N responses. On a quiet system that is fine. On a busy one some machine is always having a bad moment, and with N machines the chance that one of them is slow rises fast. The median shard answers in two milliseconds; the slowest of twenty answers in fifty; the page takes fifty. You have traded a hot shard for a slow page, and the query you run most is the one that got slow.

Suppose, for chat specifically, you shard by conversation, so that every message in a given chat lives together. Opening a chat is one shard, and sending is one shard. That is most of chat’s traffic, and it is fast. Then a group chat with a hundred thousand members turns out to be the hot key, and you are back at the first problem in a different costume.

The honest conclusion, and it is the rung, is that there is no good key. There is a key that makes your dominant query fast, and it makes the others pay. You choose it by naming the query you must make fast, the one that runs most or the one users are most sensitive to, and accepting that the queries which cut across the key will scatter, and that popular values of the key will run hot. Often different data wants different keys, messages by conversation and profiles by user, and then a query that needs both is a scatter, and you either design around it or keep a second copy arranged the other way, which is Chapter 11.

You have chosen a key, and you have four shards. The rule is simple: take the key, hash it to a big number, and the shard is that number modulo four. Every application server knows the rule, and there is nothing to look up. Then you need a fifth shard.

You change the rule to modulo five and work out where every record lives under the new rule. A key that hashed to 12 was on shard 0, since 12 mod 4 is 0; now it is on shard 2, since 12 mod 5 is 2. A key that hashed to 13 was on shard 1; now it is on shard 3. Go through them all and about four in every five records are now on the wrong machine. Adding one shard means moving eighty percent of the data, with every shard sending most of its contents to every other shard, all at once, while trying to serve traffic. That is not a scaling operation. It is a migration, and it is the reason people put off adding capacity until it is an emergency, which is the worst possible time.

The fix is to change how keys map to machines so that adding one moves only its own share. Picture the hash space as a circle, a ring. Each shard is placed at a point on the ring. A key hashes to a point on the ring too, and it belongs to the first shard you reach walking clockwise from there. Now add a fifth shard by placing it somewhere on the ring. The only keys that move are the ones between the new shard and the shard before it, the ones that used to walk past that point to the next shard and now stop at the new one. In this simplest form the new shard takes keys from a single neighbour, the one whose arc it landed in, and nobody else moves anything. That is already a vast improvement on modulo N, where everybody moved.

This is consistent hashing. A shard at a single point has two problems, though. It takes whatever-sized arc it happens to land in, so the slices are uneven; and, as just described, a new shard drains its whole share from one unlucky neighbour, which does all the copying while the rest idle. One refinement fixes both. Give each shard many points scattered around the ring, a few hundred virtual nodes, so that the arcs average out to even slices, and a new shard takes a little from every neighbour instead of a lot from one, spreading the copying work across the whole fleet.

There is one more thing the single box never needed. Something has to answer the question of which shard, for every request. If the rule is a function, a hash and a ring that everyone knows, then every application server can compute it and there is nothing to look up. That is stateless, and stateless is easy. But a function cannot express exceptions: this celebrity’s data was moved to its own machine; that shard is being drained. For that you need a table, from key range to shard, that the application consults. And a table that many machines consult, and that changes, is a copy that goes stale. A server holding an old version sends a write to a shard that no longer owns that range. It is Part II Chapter 4’s lifetimes and Chapter 3’s lag over again: the router is a small distributed database of its own, with the same failure. The question of which machine currently owns a shard, when two might think they do, belongs to Chapter 19.

Consider, finally, what can now go wrong that could not before. On one machine, “show me Mira’s page and also mark her as active” was two operations in one place, and either both happened or neither did. Across shards it is two operations on two machines, and the second can fail after the first has succeeded. A like recorded on the post’s shard and a counter incremented on the user’s shard can half-apply, and then the count is wrong. Anything that used to be a single, all-or-nothing change across two pieces of data is now a coordination problem if those pieces hash differently. You design the key so that things which change together live together, and you accept that some will not, and those will need Chapter 15’s care.

Scatter-gather brought tail latency, a page as slow as your slowest shard. Hot keys brought a machine that fails under a load the others never see. And the router brought a table that can be stale. Revision four of the design document records all of it: disk capacity and write throughput ran out; the data was divided by a key across shards; cross-shard operations can half-complete, popular keys concentrate load, scattered queries inherit the slowest shard’s latency, and the map from key to shard is itself a copy that can go stale.