Distributed systems · 14 min

Sharding

Split one table across many database machines. It buys room and speed, and makes every query and transaction that crosses machines harder.

You run the database for an online shop. Every order lives in one table, orders, on one Postgres machine. For three years that was fine.

Now the disk is filling up, and the forecast for next month’s sale says three times today’s traffic. A bigger machine buys a year at most. At some point the table has to live on more than one machine, and that changes everything you do with it.

Think of it as a dial. At one end is one database: simple, but capped by one machine. At the other end is many shards: room to grow, but every query and transaction gets harder.

One machine, and the wall it hits

Here is the shop in numbers. It has 1.2 billion orders. A row is about 500 bytes, and its indexes (the lookup structures that make “orders for user 42” fast) roughly double that, so call it 1 KB per order. That is 1.2 TB on a 2 TB disk.

The shop adds about 2 million orders a day, so 2 GB a day. The 800 GB left lasts about 400 days. Backups, index rebuilds and schema changes already take hours, because they walk the whole 1.2 TB.

Then there is speed. On a normal day the table serves about 4,000 requests a second: new orders, “where is my parcel?”, “show my past orders”. Your load tests show this machine stays fast up to about 5,000 a second for that mix; past that, every request slows down. The sale-day forecast is 12,000.

You can buy a bigger machine, and you should, for as long as it works: it is the cheapest fix. You can add read-only copies of the database to take some reads. But every new order is a write, and every write still lands on the one main machine. The 12,000 does not fit.

The other end: many shards

The idea is plain. Run several Postgres machines. Give each the same orders table, empty. Then put each order on exactly one of them, chosen by a rule. Four machines each hold about a quarter of the rows and take about a quarter of the requests.

That is the scaling half of the dial. Here is the cost half. One database could do all of these without thinking; with shards each one becomes work:

• A query that does not mention the shard key has to ask every shard and combine the answers.
• A join between rows on different shards cannot happen inside one database any more. Your code has to fetch from both and stitch.
• A transaction (“take the payment and mark the order paid, both or neither”) that touches two shards needs a two-step protocol (both shards first promise they can commit, then both commit) to stay all-or-nothing. It is slower and harder to get right.
• Adding a shard later means moving rows between machines while the shop stays open.
• There is no single auto-increment counter for order ids any more, because each shard would count on its own and hand out the same numbers. Ids have to come from a scheme that never repeats across machines.

Three choices that turn the dial

Once you decide to shard, you make three choices: how many shards, which column is the shard key, and how the rule maps a key value to a shard. There are two common rules, hash and range.

Hash: run the key through a hash function, which turns any value into a big, scrambled-looking number (the same input always gives the same number). Divide by the number of shards and keep the remainder. For user 42 with 4 shards: hash(“user:42”) is 599,759,890, and 599,759,890 divided by 4 leaves 2, so user 42’s orders go to shard 2. User 7 hashes to 2,474,178,244, which leaves 0: shard 0.

Range: give each shard one stretch of key values. With 1,000 users and 4 shards, shard 0 holds users 1 to 250, shard 1 holds 251 to 500, and so on. User 42 goes to shard 0, and so does user 43.

The difference is what happens to neighbours. Hash scatters them: users 42 and 43 can land anywhere, which spreads load evenly. Range keeps them together, which is handy when you want a stretch of values at once, like “orders from the last 7 days” if the key is the order date.

The demo below holds 2,000 sample orders from the shop’s last 30 days, from 1,000 users in 8 countries. The sample is lopsided in three ways real shops are:

• One big customer: user 42 is a reseller in Japan with 10% of all orders, and its warehouse software checks the status of every one of its orders all day. So its orders get far more requests than their share of rows: 15% of all requests are about user 42’s orders.
• One big country: Japan has close to half the orders.
• “Today” is hot: every request reads or changes one order, and half of them are about today’s orders, which are being paid for, packed and tracked.

The sale-day peak, 12,000 requests a second, is shared between the orders by those rules. Each shard is drawn with two bars: how many rows it stores, and how many requests a second it serves.

Hot spots: one shard doing everyone’s work

The demo showed three causes, one for each lopsided thing in the data:

• By order date, today’s shard takes every new order and half of all requests. Adding shards does not help: today still lives on one of them.
• By country with Range and 4 shards, Japan’s shard holds about half the rows and well over half the work. There are only 8 countries, so there are only 8 values to spread. However many shards you add, all of Japan’s orders share one of them. Few values also make hash lumpy: with Country, Hash and 4 shards, one shard gets no country at all and stores nothing.
• By user id, the load spreads well, except for user 42. One customer’s orders always share one shard, so that shard stays the busiest however many you add.

The rule behind all three: a good shard key has many different values, and requests are spread fairly evenly across them. User id has a million values in a real shop, and most users are small. Date and country fail both tests.

For the rare giant key like user 42, there are standard fixes. You can give it its own shard. You can add a suffix to its key (“42#0” to “42#7”) so its orders spread across 8 shards, at the price of asking all 8 when you read them. Or you can put a cache in front of its most repeated reads.

Queries that don’t know the key

The router can only skip shards when the query names the shard key. “Orders for user 42” names it, so the router computes one shard. “Orders in Japan” does not, so the router has no idea where Japanese orders live and must ask every shard.

Fan-out costs more than it looks. Every shard does the work of the query, so 8 shards do 8 times the work of one. And the slowest one sets the pace. If each shard is slow 1 time in 100, a query that asks 8 shards hits at least one slow shard about 8 times in 100. (The chance that all 8 are fast is 0.99 multiplied by itself 8 times, about 0.92, which leaves about 0.08 for “at least one is slow”.)

Joins and transactions follow the same rule. If the orders table and the payments table are both sharded by user id, a user’s orders and payments sit on the same shard. Joining them, or updating both in one transaction, stays inside one database. Citus, an extension that shards Postgres, calls these “co-located” tables and builds on exactly this.

So choose the shard key from your most common queries. For the shop, nearly every request is “this user’s orders” or “this order”, which is why user id wins. The rare “all orders in Japan this week” report can fan out, or better, run on a separate copy of the data built for reports.

Adding a shard later

Moving 80% of a 1.2 TB table between live machines is days of copying while orders keep arriving. That is why plain “hash mod N” (divide by N shards and keep the remainder) is rarely used as is. Consistent hashing is the usual fix: it changes the rule so that a new shard only takes keys from the others, and only about 1 in N of the keys move (its own article shows how).

Another common answer is logical shards: pick a large fixed number of small shards, say 4,096, and find each key’s small shard with hash mod 4,096. That number never changes. A lookup table then says which machine holds which small shard. Growing means moving some whole small shards to a new machine and updating the table; no row ever needs a new shard number. Instagram did this with thousands of logical shards on its Postgres machines.

Setting the dial for the shop

Put it together. The shop needs to survive 12,000 requests a second when one machine copes with 5,000. Four shards would be enough on paper (12,000 ÷ 4 = 3,000 each), but the demo showed that skew pushes the busiest shard over. Six shards keep every one under the line. Logical shards spread the same way, so 6 machines should give a similar picture; measure it before the sale.

• Shard key: user id, because most requests name a user, and orders and payments for one user can stay together.
• Method: hash into many logical shards, so adding machines later moves whole pieces instead of 80% of rows.
• Count: 6 machines for the sale, leaving room before the busiest shard reaches 5,000.
• Exceptions: user 42 gets its reads cached or its own shard; reports by date or country run on a separate copy.

And before any of that: if a bigger disk, read copies and archiving old orders would carry you another year, take that year. Sharding is the right end of the dial only when the left end has really run out.