Spreading data across machines by sharding
Open in the interactive app: practice, quiz, mock interview →In one minute
- Sharding spreads one dataset across several independent database servers, each holding a disjoint subset of rows, so write throughput, storage and memory scale with the number of machines.
- Every sharded system has three parts: a shard key, a function from key to shard, and a routing layer that knows which machine holds which shard right now.
- Good shard keys keep most queries on one shard; bad ones turn every request into a scatter-gather and every transaction into a distributed one.
- Map keys to many small logical shards (Redis Cluster's 16,384 slots, Notion's 480 schemas), then place logical shards on physical machines. Resharding becomes moving whole shards, not rehashing rows.
- The hard parts are not the hash function. They are cross-shard queries, uniqueness and joins, hot tenants, and moving data online without losing writes. The problem shapes are "hot spots", "fan-out" and "multi-step consistency".
How it works
When one machine stops being enough
Partitioning inside one server (previous lesson) still shares one CPU, one WAL and one disk, and replicas add read capacity but take every write. When the primary's write rate, working set or disk outgrows the largest instance you can buy, different rows need different primaries. That is sharding: each partition lives on its own server with its own replicas.
Choosing the shard key
A shard key should have high cardinality (millions of values, not 50 countries), even frequency (no single value with 30% of traffic), and, most important, it should be present in almost every query. For a SaaS product that is usually tenant_id: one tenant's data lives together, joins and transactions within a tenant stay local, and tenants spread across shards. For a social app it might be user_id, which keeps a user's posts together but makes "all comments on this post" a cross-shard question if comments are keyed by commenter.
Monotonic keys like timestamps are poor shard keys under range placement, because all new writes hit the newest shard.
Placement strategies
- Hash:
slot = hash(key) mod Swith S fixed and large, then a table maps slots to servers. Even spread, no range locality. - Range: contiguous key ranges per shard, split as they grow. Good for scans; needs care with hot tails. MongoDB's balancer, for example, aims for equal data per shard and moves ranges only when two shards differ by three times the range size, 384 MB at the default 128 MB (MongoDB docs).
- Directory (lookup): a table maps each key, or each tenant, to a shard. Fully flexible, so a big tenant can get a dedicated shard, at the cost of a lookup on every request and a directory that must be highly available.
- Geographic: shard by region for latency and data residency, usually combined with one of the above inside each region.
Routing
In client-side routing the application library holds the shard map (Instagram's and Notion's Postgres sharding work this way). In proxy routing a tier such as Vitess vtgate, MongoDB mongos or the Citus coordinator parses the query, finds the shard key in the WHERE clause and forwards the query, or fans it out and merges results when it cannot. In server-side redirection any node accepts a request and either serves it or tells the client where to go, as Redis Cluster does with MOVED (Redis Cluster spec).
What you give up
Within one shard you keep transactions, foreign keys and joins; across shards you lose them by default. The shard key itself is hard to change later. MongoDB only gained online resharding to a new key in 5.0 (MongoDB docs), and most hand-built systems never get it. Unique non-key columns need a global table, queries without the key fan out, and multi-shard writes need two-phase commit or a saga. Sequences must become globally unique IDs, for example Instagram's 64-bit layout of 41 bits of milliseconds, 13 bits of logical shard and 10 bits of per-shard sequence, which also lets you find a row's shard from its ID.
Down the stack
Layer 1The contract
A statement whose shard key resolves to one shard gets that shard's full guarantees: transactions, constraints, fast index lookups. A statement touching several shards gets weaker promises: Vitess's default transaction_mode is MULTI, which commits multi-shard writes shard by shard with no atomicity guarantee; TWOPC makes them atomic but needs semi-sync replication and adds commit latency (Vitess docs); a scatter read sees each shard at a slightly different moment, not one global snapshot. During resharding the contract is that no committed write is lost and that each key has exactly one writable owner at any instant, with a brief window where writes to moving keys block or fail and are retried.
Layer 2The mechanism
Redis Cluster computes HASH_SLOT = CRC16(key) mod 16384 using CRC16-XMODEM (Redis Cluster spec). If the key contains a non-empty {...} section, only that hash tag is hashed, so {user1000}.following and {user1000}.followers share a slot and can be used together in one multi-key command. Nodes gossip bitmaps of the slots they own. To move a slot, the operator marks it MIGRATING on the source and IMPORTING on the target and runs MIGRATE on its keys in batches. Meanwhile a request for an already moved key gets -ASK <slot> <host:port>, and the client sends ASKING plus the command to the target for that one request. CLUSTER SETSLOT <slot> NODE <id> then makes the move permanent, and clients get -MOVED and update their slot map.
Vitess maps each row through its primary vindex to a 64-bit keyspace ID, and names shards by keyspace ID ranges: -80 and 80- split the space in half. Resharding -80 into -40 and 40-80 runs a VReplication workflow (Vitess docs): a copy phase streams rows to the new shards, then the workflow tails the source's binlog by GTID to catch up. VDiff compares source and target row by row. SwitchTraffic moves reads first, then writes: it stops writes on the source, waits for the target to reach the final GTID, flips the routing in the topology server, and starts reverse replication so you can switch back if the new shards misbehave.
Layer 3The machine
Scatter queries pay for tail latency. If each shard answers in under 20 ms 99% of the time, a query over 64 shards waits for the slowest: the chance that at least one is slow is 1 - 0.99^64, about 47%, so the fan-out query's median is near a single shard's p99. Dean and Barroso's version: with a 1-in-100 slow server, a request that fans out to 100 servers is slow 63% of the time (The Tail at Scale). Connections multiply too. Two hundred application pods, each with a pool of 10 to each of 32 shards, hold 64,000 server connections; at a few MB of memory per Postgres backend that is hundreds of GB, which is why sharded Postgres nearly always runs a pooler such as PgBouncer. Notion hit exactly this when it tripled its shard count: it had to shard its PgBouncer layer too, because old and new shards would otherwise have exceeded the connection limits during the migration (Notion).
Copying a 2 TB shard at a throttled 200 MB/s takes about 3 hours while the source keeps serving and its binlog grows. The cutover is short: stop writes, apply the last few MB of binlog in well under a second, update routing, resume. If you turn on vtgate buffering (--enable-buffer, off by default), it holds queries for up to --buffer-window (10 seconds) during the switch, so clients see added latency instead of errors (Vitess docs).
Why it's built this way
- A single primary's write path is limited by one log device and one CPU set, so writes must be split across primaries.
- Rehashing every row when a server is added is too expensive, so keys map to a fixed, large set of logical shards and only the shard-to-server map changes.
- Clients cannot all learn a new map at the same instant, so servers redirect stale clients (
MOVED,ASK) or a proxy owns the map. - Copying a large shard takes hours and the source must stay online, so moves are copy plus change-log catch-up, with a short write freeze only at the end.
- A bad cutover is costly, so systems verify with a diff and keep reverse replication running for a cheap rollback.
Diagram
sequenceDiagram
participant C as Client
participant A as Node A (slot 12218 MIGRATING)
participant B as Node B (slot 12218 IMPORTING)
C->>A: GET {user1000}.following
A-->>C: -ASK 12218 nodeB
C->>B: ASKING
C->>B: GET {user1000}.following
B-->>C: value
Note over A,B: MIGRATE finishes, SETSLOT 12218 NODE B
C->>A: GET {user1000}.followers
A-->>C: -MOVED 12218 nodeB
C->>B: GET {user1000}.followersWorked example
A B2B messaging product stores 18 TB of messages and peaks at 60,000 writes per second. Its largest practical Postgres primary sustains about 8,000 writes per second with headroom and holds 4 TB comfortably. Writes need at least 8 primaries and storage needs at least 5, so plan 16 physical shards for growth.
The shard key is workspace_id, since every message query is scoped to a workspace. Keys hash into 1,024 logical shards, 64 per server. One 40,000-employee customer generates 6% of writes, 3,600 per second, half a server's capacity, so the directory pins its logical shard to a server of its own: a config change, not a code change.
A year later traffic doubles and the team adds 16 servers. Each old server hands 32 of its 64 logical shards to a new one, moving half the data, 9 TB, which is the minimum for an even spread. The next step, from 32 to 40 servers, shows the real benefit: with logical shards only one fifth of the data moves, while rehashing rows from mod 32 to mod 40 would relocate about 80% of them. At 200 MB/s per stream and 8 parallel moves, 9 TB copies in under 2 hours, followed by one short cutover per logical shard.
In the wild
- Slack, 2022: a cache keyed one way, a database sharded another. A gradual Consul agent upgrade removed memcached nodes, and Slack's cache control plane replaced them with empty ones. Group DM membership was cached by channel ID, but the Vitess table was sharded by user, so every miss became a scatter query across all shards. The database overloaded and could not refill the cache, retries piled on, and many users could not load Slack. The fix moved the query to a table sharded by channel and audited other cache-fronted scatter queries (Slack).
- Stripe, 2019: one shard without a primary stalls the whole API. A latent bug from a database upgrade meant one shard could not elect a new primary when its old one failed, and it rejected writes. API workers blocked on that shard, so the API was severely degraded for about 27 minutes, and rolling back the election protocol then clashed with a recent config change and caused a second, 93-minute outage. Stripe's fixes included shard fault isolation and circuit breaking so one bad shard cannot take the fleet with it (Stripe).
- Notion, 2023: re-sharding from 32 to 96 machines. By late 2022 some shards ran above 90% CPU at peak and PgBouncer was near its connection limits. Notion spread its 480 logical shards from 15 per machine to 5, copied them with Postgres logical replication (3 days cut to 12 hours by building indexes afterwards), validated with dark reads, and cut over by updating PgBouncer mappings with reverse replication ready. Peak CPU dropped to about 20% with no observed downtime (Notion).
Trade-offs
| Option | Strengths | Weaknesses | Choose when |
|---|---|---|---|
| Hash into logical slots | Even spread, cheap rebalancing by moving slots | No range locality; scatter for range queries | Point lookups by key, many similar tenants |
| Range shards | Range scans and ordered pagination stay local | Hot tail with monotonic keys; needs splitting | Time-ordered or lexicographic access |
| Directory per tenant | Isolate or move any tenant; heterogeneous shard sizes | Directory lookup on every request; must be highly available | Multi-tenant SaaS with very uneven tenants |
| Proxy routing (vtgate, mongos) | Apps see one database; central control of resharding | Extra hop, proxy tier to run and scale | Many services, many languages |
| Client-side routing | No extra hop, simple infrastructure | Every client must update its map; harder to evolve | One codebase, small team, few clients |
Pitfalls in production
- Sharding too early. A sharded database costs every future feature a cross-shard question. Exhaust vertical scaling, read replicas, partitioning and caching first, and shard when one primary's write or storage limit is a year away, not ten.
- A shard key absent from hot queries. If the inbox query filters by
recipient_idbut the key issender_id, every inbox load is a 64-way scatter. Model the top queries before choosing, and maintain a second copy keyed the other way if needed. - One tenant outgrows a shard. Hashing cannot split a single key. Plan for it with a directory that can place a giant tenant alone, or a compound key (
tenant_id,bucket) for the few tenants that need it. - Global uniqueness lost silently. Two shards each accept the same email address. Keep a small global table for unique columns, or reserve uniqueness with a write there first.
- Big-bang resharding. Moving everything in one weekend with writes stopped turns a routine operation into an outage. Move logical shards one at a time with copy, catch-up, verify and a fast cutover, and keep a reverse stream for rollback.
Leadership lens
- Ask what percentage of production queries will touch more than one shard; above a few percent the design needs a second index or a different key.
- Sharding is an organisational commitment: tooling for moves, a shard-aware migration process and on-call who understand it. Budget a platform team, or buy it (Vitess, Citus, a distributed SQL database).
- Decide early whether resharding is routine or exceptional. Routine needs automated, verified, reversible moves; exceptional means you are betting on not growing.
- Make the noisy-tenant plan explicit before the largest customer signs; placing them alone on day one beats moving them during an incident.
Check yourself
Why do production systems hash keys into thousands of logical shards instead of directly onto N servers?
Show answer
Adding servers then moves whole logical shards, the minimum needed for balance, with only the shard-to-server map changing. Hashing directly onto N servers changes most keys' owners whenever N changes.
In Redis Cluster, what is the difference between
-ASKand-MOVED?Show answer
ASKis a one-request redirect while a slot is mid-migration, and the client must not update its slot map.MOVEDmeans the slot has permanently changed owner, so the client updates its map.Why does Vitess switch reads before writes during resharding?
Show answer
Reads on the new shards can be validated while the source is still authoritative and writes are unaffected; if they misbehave you switch reads back. Writes switch last, behind a brief freeze, because that is the irreversible step.
Sources and further study
Primary
- Corbett et al., Spanner (OSDI 2012): a database that shards across Paxos groups and reshards automatically, the alternative to hand-built sharding.
- Dean and Barroso, The Tail at Scale (CACM 2013): why fan-out queries inherit the slowest shard's latency, and hedged requests as a fix.
Systems docs
- Vitess: Distributed transactions:
SINGLE,MULTIandTWOPCmodes and what each promises. - Vitess: VTGate buffering: holding queries during failovers and resharding cutovers.
- MongoDB: Balancer: data-size-based balancing, migration thresholds and range size.
- Citus: Concepts: distributed, reference and local tables, and co-location by distribution column.
Books
- Designing Data-Intensive Applications, Martin Kleppmann, chapter 6 "Partitioning" (1st edition) or chapter 7 "Sharding" (2nd edition), which adds sharding for multitenancy.
- Designing Distributed Systems, Brendan Burns, chapter 6 "Sharded Services" and chapter 7 "Scatter/Gather".
- System Design Interview, Volume 1, Alex Xu, chapter 1 "Scale From Zero To Millions Of Users": sharding as the last step after replicas and caching, with resharding, celebrity and join problems.
Other curricula
- Azure Architecture Center: Sharding pattern: lookup, range, hash and geographic strategies, with virtual shards over physical ones and a clear "when not to shard" list.
- donnemartin/system-design-primer: lists sharding with its costs (joins, rebalancing, hot users) and points to consistent hashing, at an overview level.
Watch
- PlanetScale: Query Planning for a Sharded System like Vitess (CMU Database Group): Sugu Sougoumarane on how vtgate routes and plans queries across shards.
- The ONE thing you lose when sharding a database (PlanetScale): a short look at what cross-shard operations cost.
- Good and bad sharding keys (Ben Dicken): how shard key choice decides hot spots and fan-out.
Free resources
- Redis Cluster specification: hash slots, hash tags,
MOVEDandASKredirection, and live slot migration. - Vitess: Vindexes: how a column maps to a keyspace ID, including lookup vindexes for secondary keys.
- Vitess: Resharding: the copy, catch-up, verify and switch-traffic workflow step by step.
- Herding elephants: lessons learned from sharding Postgres at Notion: 480 logical shards over 32 databases, and how the migration ran.
- How Figma's databases team lived to tell the scale: horizontal sharding of Postgres with a query proxy and logical shards.