Lesson 013 · Phase 1, Foundations

Sharding: Cutting the Database So It Fits

What a shard is, the arithmetic of splitting one dataset across machines, why a sharded system always costs you the maximum rather than the average, and what stops working.

19 min read

Lesson 13 · 22 published · 90 planned

On this page
The systems in this lessonUsed here: Marlow Books, Stagefront, and Galewatch.

Made up for this course and reused from lesson to lesson so their numbers become familiar. None of them exist. All three

Marlow Books · A small online bookshop
Four people, one server and one Postgres database. About 40 requests a second on a normal day and ten times that in the week before Christmas. The one box that the early lessons stress until it breaks.
Stagefront · An event ticketing service
Quiet most of the time, then a stadium show goes on sale at 10:00 and two hundred thousand people press the same button in the same minute. Oversold seats are a lawsuit, so correctness matters as much as speed.
Galewatch · Telemetry for wind farms
Nine hundred turbines, a reading every two seconds, over links that drop for hours in bad weather and come back with a backlog. Dashboards that lag by seconds, reports that scan a year.

On a Thursday in February the on-call engineer at Galewatch, a wind farm telemetry company that exists only in this course, started a restore drill at nine in the morning and went to make coffee. The database came back at twenty past three.

Galewatch collects a reading from each of nine hundred turbines every two seconds and sells the owners dashboards. Nothing was broken that Thursday. Nothing was overloaded. The drill was a scheduled exercise, restoring last night's backup onto an empty machine to prove that it could be done, and it was done. It took six hours and twenty minutes.

The recovery target in the contract signed the previous October is one hour.

What had changed was retention. Until November, Galewatch kept a rolling fourteen days of readings live, 109 gigabytes, and everything older went to an archive nobody queried interactively. Lesson 002 has the van: 8.4 terabytes of one customer's history copied onto drives and driven five hundred kilometres overnight, because the site link would have taken thirty nine days. Customers had been asking for years to stop needing the van. In November they got it. The live database went from a fortnight to a year, which is lesson 001's 2.8 terabytes of readings and four and a half with the indexes on top.

The writes never moved. Four hundred and fifty readings a second in October, four hundred and fifty in December. CPU never went above thirty percent. The panel an engineer opens at Tarrow Ridge, the sixty turbine farm in the portable building, still answered in under ten milliseconds, because a seek down lesson 002's index on (turbine_id, recorded_at) grows with the logarithm of the table, and going from a fortnight to a year adds at most one level to the tree.

Nothing on any dashboard was wrong. The machine had simply become one object too large to put back.

That is the honest reason a lot of systems get sharded. The textbook gives a different one.

Cutting the table is not cutting the machine

Sharding means splitting one logical dataset across several machines so that each machine holds some of the rows and no machine holds all of them. Each machine's slice is a shard. It is the first idea in this course where no single machine can answer the question.

Set it against replication, which lesson 011 built. A replica holds every row. That is the whole point of it, and it is why any replica can answer any read. Lesson 011's line was that reads divide and writes copy, and that line is also the ceiling: stand five replicas behind a primary and each one carries all of the writes for a fifth of the reads, so once replay alone fills a machine, the sixth replica buys you nothing.

A shard turns both halves over. Writes divide, because a write goes to one shard and the other fifteen never hear about it. Reads divide unevenly: a read that names the key goes to one shard, and a read that doesn't goes to all of them.

replication                 sharding
  every box has every row     no box has every row

  read   ask anyone           read   ask the right one
  write  ask the primary      write  ask the right one

                              and "which one" is now
                              a piece of code you own

That last line is the part people underestimate. With one database, "where is this row" was never a question anybody had to answer. Now something has to answer it before every statement, and if that something is wrong you do not get an error. You get zero rows, from a perfectly healthy machine, in four milliseconds.

There are three places that code can live, and they are the same three shapes lesson 011 gave for routing reads to replicas. In the application, as a function from key to connection, which is the cheapest to build and then has to exist in every service that touches the data. In a proxy speaking the database's own protocol, which puts the map in one place and adds a tier that can be down. Or inside a database product that shards for you, which is the least code you will ever write and means you have bought a distributed system with a SQL face and inherited its failure modes instead of choosing them. Whichever you take, the map from key to shard is production state now, as load bearing as the rows, and it has to be right on every machine at the same time.

Galewatch has been doing something that looks like sharding for years, and it isn't. It keeps one hot readings table per farm. Lesson 002 priced what that was worth: Tarrow Ridge's fortnight is 7.3 gigabytes and 3.6 seconds to scan unindexed, against 109 gigabytes and nearly a minute for all nine hundred turbines in one table.

That is partitioning: one dataset cut into pieces that live on the same machine. Have it. It is cheap, it needs no routing layer, and it buys two things an index cannot. Dropping the oldest day becomes a file unlink instead of a DELETE across 38.88 million rows. And vacuum, statistics and index rebuilds happen a piece at a time, so no single maintenance operation has to swallow the whole table.

What partitioning does not buy is any of the things that hurt. One disk. One buffer pool. One write-ahead log. One hundred connections. One machine to restore. Every one of Galewatch's per-farm tables sat on the volume the drill spent six hours refilling.

Cutting the table is not cutting the machine.

The number that forces the cut

The textbook says you shard when the data no longer fits on one machine or the writes no longer fit. Galewatch meets neither condition.

Four and a half terabytes on one volume is unremarkable; you can rent that in a web form. The writes are 450 a second, and lesson 005 timed a Galewatch write at 20 milliseconds and derived a ceiling of 5,000 a second from the hundred connection limit, so ingestion is running at under a tenth of a limit it already knows about. Reads are answered by an index in under ten milliseconds. There is no graph in the building with a worrying shape on it.

And the restore takes six hours and twenty minutes against a contract that says one.

Start by turning the drill into a rate, because the drill is the only honest measurement anybody has of how fast this data moves in bulk.

4,500 GB / 22,800 seconds = just under 200 MB a second

Two hundred megabytes a second, close enough to the van's 230 that the comparison stops being a joke. A network restore of a multi-terabyte database and a car full of hard drives land in the same order of magnitude, and both of them get there by moving bulk data at bulk data speeds.

Now size the cut from the operation you cannot be slow at.

recovery target      1 hour
measured rate        200 MB/s
so a shard holds     720 GB, maximum

4,500 GB / 720 GB  = 6.25, so at least seven shards

An hour at two hundred megabytes a second is 720 gigabytes, so that is the most any one shard can be allowed to hold, and 4,500 divided by 720 is 6.25. Seven covers today and nothing else. Lesson 005 established how Galewatch grows: not on a curve but in steps, when somebody in sales signs a contract, which is why the company has a notice period rather than a runway. So the sizing question goes to sales rather than to a graph: what is the largest fleet you could sell in eighteen months? The answer that came back was a doubling. After a doubling, seven shards is 1.3 terabytes each and an hour and three quarters, so you would need thirteen. Sixteen is the next number a human would actually operate.

At sixteen, each shard holds 281 gigabytes and restores in about twenty three minutes. That number assumes all sixteen restore at once, and that the 200 megabytes a second was a property of the machine rather than of the one pipe they would all be pulling the backup through. Check which it was before the twenty three minutes goes in a runbook, because a restore that serialises on a single bucket hands you straight back the six hours you were trying to lose.

That is the whole derivation, and notice what is missing from it. No query appears anywhere. The query was fine before and it is fine after. You size a cut for the operation you cannot be slow at, and for most companies that operation happens once a year, at night, in a drill nobody watches.

What divides and what multiplies

Quantity On one machine Across sixteen shards
Rows and bytes 4.5 TB 281 GB each
Writes 450 a second 28 a second each
Restore 6 h 20 23 min each
Machines to run 1 16, and 011 wants a replica on each
Pool per ingest box 25 connections 25 per shard, if nobody thinks

That last row is the one that bites, and it is Little's law from lesson 002 doing its usual quiet work. Each shard now takes 28 writes a second at 20 milliseconds apiece, so 0.56 writes are in flight on it at any moment. A pool of two or three per shard is generous. Nobody sets two, because 25 is a number in a configuration file that gets copied, and four ingest boxes holding 25 connections to each of sixteen shards is 1,600 connections against sixteen limits of 100, which is 1,600 slots. You will have solved a connection limit by multiplying both sides of it and landing on exactly the same number, and the fifth ingest box will break it again the way it did in lesson 005.

The bill multiplies too, and there is no clever way around it. Sixteen boxes at lesson 005's $250 a month for eight vCPUs and 32 gigabytes is $4,000; lesson 011 says each of them wants a replica, so call it thirty two machines and $8,000 a month to hold four and a half terabytes that would sit happily on one volume. You are not buying capacity here. You are buying a smaller unit of failure and a shorter worst hour, and those cost real money.

One thing does get genuinely cheaper. Each replica now replays a sixteenth of the log, and lesson 012 spent its length on what single threaded replay costs when it has the whole log to chew, so per-shard lag falls. Sharding is the only move in this course that makes replication cheaper.

Which is the moment to say the disappointing part out loud. You do not end up with no primary. You end up with sixteen primaries, each one carrying every problem lessons 011 and 012 described, at a sixteenth of the size. The failover, the promotion, the fencing, the lag, the read that comes back without your own write: all still there, sixteen times, and now nobody can hold the whole system in their head.

Twenty rows, eleven machines

Queries against a sharded system come in two shapes, and the difference between them is the whole operational story.

A targeted query names the shard key, so the router sends it to one machine. Tarrow Ridge turbine 14, yesterday's readings. One shard, one index seek, and it is slightly faster than before because that shard's index is a sixteenth the size.

A scatter-gather query, also called a fan-out, does not name the key, so it goes to every shard and something merges the answers. Which turbines across the fleet ran yesterday at under forty percent of nameplate, the output a turbine is rated for. Sixteen queries, sixteen partial answers, one merge you wrote yourself and now maintain.

The trap is the query that looks targeted and isn't. Lesson 010 gave a model for how many pages M matching rows land on when they're scattered over P of them, P times one minus e to the minus M over P, and the same model works one level up with shards in place of pages. Shard Galewatch by turbine and ask for twenty named turbines:

16 x (1 - e^-(20/16)) = 16 x 0.714 = 11.4 shards

Twenty rows, eleven machines. That query has a WHERE clause, a small result and an index behind it, and it is a fan-out. Push it a little further and it gets worse: Tarrow Ridge's sixty turbines come out at 16 times one minus e to the minus sixty over sixteen, which is 15.6 of your 16 machines. The flagship panel of lesson 002, the one the engineers in the portable building open all day, would touch every machine the company owns.

Then the tail arrives, exactly as lesson 002 promised it would. That lesson showed sixty calls each with a one in a hundred chance of taking 400 milliseconds giving forty five percent of page loads a slow call, and called it the tail becoming the typical. Same arithmetic here. Say a per-shard read is 4 milliseconds at p50 and 90 at p99:

16 shards:  0.99^16  = 0.85, so 15% of fleet queries wait 90 ms
100 shards: 0.99^100 = 0.37, so 63% of them do

A fan-out query does not cost the average shard's time. It costs the slowest one's.

Now put a replica behind each shard, because you will, and run the same sum on staleness instead of latency. If each replica has a one in a hundred chance of being eighty milliseconds behind, a fleet query reading sixteen of them has a fifteen percent chance that some part of the answer is stale, and no part of the answer says which part. Lesson 012's promises are per-shard promises. Nothing hands them to you across a fan-out.

In a sharded system you never pay the average. You pay the maximum: the slowest shard, the stalest replica, the biggest piece, the longest restore.

The things that quietly stop working

Identifiers. A BIGSERIAL column is a counter inside one database. Sixteen databases have sixteen counters, and each of them is about to hand out row number 1. UUIDs solve it, and they are random by design, so every insert dirties a different leaf page instead of them all landing at the end of the tree together. That is lesson 010's correlation showing up on the write path, and it is not free. Or you put the shard number into the identifier and keep a per-shard counter for the rest. Whichever you pick, pick it before there is data.

Unique constraints. UNIQUE is a promise one machine keeps by looking at every row it holds. No machine holds every row now, so the promise only survives on the shard key, or on something the shard key is part of. Galewatch's turbine serial numbers are unique across the fleet and are not the shard key, so two commissioning jobs running on two shards on the same afternoon can both accept WTG-4471 and neither will complain. The fix is a small table somewhere that owns serial numbers, which every write must consult, which is a single machine everybody depends on. You have un-sharded the one thing you needed most.

Joins. A join is cheap because a database can put two rows next to each other. Across machines it can't, so the join moves into your application and becomes a fan-out plus a merge. Galewatch joins readings to a turbines table of nine hundred rows holding make, nameplate and commissioning date. The answer is to copy that small table onto every shard as a reference table, and it works beautifully until somebody recommissions a turbine, at which point you have lesson 009's invalidation problem with sixteen addresses and no TTL to hide behind.

Transactions. Inside one shard, the database's guarantees hold and lesson 015 will say what they are. Across two shards there is no transaction, only two commits that can half succeed. Stagefront, the ticketing service in this course where two hundred thousand people press the same button at 10:00, sells four adjacent seats as one purchase. Shard the seats and that purchase spans up to four machines, so you need either two-phase commit, where a coordinator makes every machine promise it can commit before any of them does, which is slow and has a failure mode of its own, or lesson 039's sagas, which means compensating a charge you have already taken. In a business where oversold seats are a lawsuit.

And Stagefront has the sharper problem underneath that one. Shard it by event instead, so one show's seats stay on one machine and every purchase is a clean single-shard transaction. Now the on-sale minute is one machine at a hundred percent CPU with fifteen idle machines watching. Sharding spreads load across data. It does nothing about load concentrated in time on one key, which is precisely the shape of every interesting ticketing, auction and flash-sale problem. Lesson 047 owns hot keys.

Moving a shard, and the outage that made this famous

Everything above assumes the cut is already made. Changing it is the expensive part, and it is expensive in a specific way: while a row is moving from shard 9 to shard 17, it is briefly in both places or in neither, and every reader has to agree which map it's using. Lesson 031 is about doing this without reshuffling everything, and it exists because the naive version, hashing the key modulo the number of shards, moves nearly every row the moment that number changes.

In October 2010 Foursquare went down for about eleven hours on a Monday, and again on the Tuesday while they were still repairing it. The database behind check-ins was MongoDB, sharded across two machines by user identifier. One of the two shards grew faster than the other until its working set no longer fitted in that machine's memory, and lesson 010's change of shape arrived on schedule: reads that had been served from RAM started going to disk, latency collapsed, and the site went with it.

The obvious fix did not work quickly. They added a third shard, but the data had to be migrated onto it from a machine that was already saturated, and after the migration the original shard's files were fragmented enough that it still did not fit in memory. Getting the space back meant taking the shard down and compacting it.

Two things in that are worth keeping. The failure was not one shard being too big; it was two shards being unequal, which is lesson 014's subject. And a rebalance is an operation you can usually only start once you already have a problem, using a machine that already has the problem. Plan the cut for the fleet you will have, because the cheap moment to choose your shard count is before there is any data in it, and there is no cheap moment after.

What Galewatch actually did, and what it cost

Nineteen farms, sixteen machines, and a choice between two cuts.

Shard by turbine and the shards come out even: 281 gigabytes each, 28 writes a second each, and every farm panel in the product becomes a sixteen way fan-out with a p99 attached. Shard by farm and Tarrow Ridge is one machine answering in under ten milliseconds exactly as it does today, and the shards are as uneven as the farms are. Four and a half terabytes over nine hundred turbines is five gigabytes per turbine-year. Kilmore Sands, the biggest of the nineteen, has a hundred and forty turbines, so its shard is 700 gigabytes against a budget of 720.

They took the farms, doubled the small sites up to fill sixteen machines, and kept the queries. Which means they also gave up the headroom the whole derivation was for, because headroom in a sharded system is headroom on the biggest shard, and Kilmore Sands is already full. June's drill came back in fifty eight minutes.

Not twenty three. A sharded restore finishes when the slowest shard finishes, and Kilmore Sands is 700 gigabytes at 200 megabytes a second. The number that went into the runbook was fifty eight, two minutes inside a contractual hour, on day one, before a single new contract. Lesson 014 is about that choice and about living with it.

Marlow should not do any of this

Marlow Books is a four person online bookshop that exists only in this course, and it is the counterexample. Lesson 010 measured its entire books table at 240 megabytes and the shop's whole database load at 1.04 seconds of work a second, which lesson 011 pointed out is six and a half percent of a sixteen core machine. Lesson 005 measured the pair of boxes at 200 pages a second, then 260, then 265, because Postgres never left box A and the ceiling never moved.

That ceiling is real, and sharding is genuinely the move that would remove it. It is still the wrong move, because the machine sitting under it is doing 1.04 seconds of work a second on sixteen cores and is, by any reading, asleep.

If your working set fits in memory, you don't have a sharding problem. You have an indexing problem, which is lesson 010 and a day's work, or a caching problem, which is lesson 008 and an afternoon. Sharding is a year, a routing layer, a rewrite of every query that touches two shards, and a founder who deploys on Friday afternoons doing it to sixteen databases.

Be very sure it is the last thing left.

Recap

A replica answers who else has this row; a shard answers who has it at all. Replication copies rows so any machine can answer. Sharding divides them so exactly one machine can, and something now has to know which, before every statement.

Cutting the table is not cutting the machine. Partitioning gives you cheap drops and per-piece maintenance on the same disk, the same log and the same connection limit. It is worth having and it is not sharding.

Size the cut for the operation you cannot be slow at. Galewatch's shard count came out of a restore drill and a contractual hour, and no query appears anywhere in the arithmetic. The queries were never the problem.

Twenty rows, eleven machines. Lesson 010's scatter model works at shard granularity too, so a query with a small result and a good index can still be a fan-out. Check the model before you assume a query is targeted.

You never pay the average, you pay the maximum. The slowest shard sets your latency, the stalest replica sets your freshness, and the biggest shard sets your restore time and your capacity. Every one of them is a maximum over N.

Sharding does not remove the primary, it removes the single primary. Everything in lessons 011 and 012 is still true, once per shard, at a sixteenth of the size, and now spread across sixteen machines nobody can hold in their head at once.

Check your understanding

  1. A service holds 900 gigabytes on one database, takes 300 writes a second against a measured ceiling of 4,000, and answers every query from an index in under fifteen milliseconds. The team wants to shard it. Give the two questions you would ask before agreeing, and say what answer to each would change your mind.

  2. Galewatch's fleet panel asks for the twenty worst performing turbines in the last hour. Explain why sharding by turbine makes this a fan-out even though only twenty rows come back, and say what you would measure on the existing unsharded system to predict how slow it will get.

  3. A team shards by customer identifier and keeps the existing per-service pool size of 20 connections. There are six services, four instances each, and twelve shards. Work out the total connection count, compare it against the shards' combined limit, and say what the pool size should actually be if a shard takes 40 writes a second at 25 milliseconds each.

  4. Somebody proposes sharding Stagefront's seat inventory by seat identifier so that load spreads evenly during an on-sale. Say what this fixes, name the specific purchase that breaks, and describe what would have to be true about the business for you to accept the trade anyway.

  5. You inherit a system sharded into four pieces two years ago, and one shard is now three times the size of the others. Say what you would measure first, what makes fixing it harder than the original cut was, and one thing you would put in place today so the next team is not in this position.

Next lesson

014 Choosing a Shard Key and Living With It. Today showed what a cut costs and let Galewatch pick between even shards and cheap queries without arguing the choice properly; next lesson argues it, and takes seriously the fact that a shard key is the one decision in this course you cannot undo cheaply.

Finished reading?

Marking a lesson done keeps your place on the course index. It is stored only in this browser.

Tip: use the ← and → keys to move between lessons.