Sharding and partitioning

Splitting rows across machines#

Sharding (horizontal partitioning) puts different rows on different machines. Every shard has the same schema and a slice of the rows. The column that decides which slice a row belongs to is the partition key.

The router, whether it is a library in your app or a proxy in front of the database, uses the partition key to pick a shard. A query that includes the key goes to exactly one shard. A query that doesn't has to ask all of them. Nearly every trade-off on this page comes back to those two sentences.

The key decides where today's writes land#

Range partitioning by time still has a place. Time-range scans stay on one or two shards, and deleting a year of data is dropping a partition instead of deleting a billion rows. You accept a permanently hot write shard to get that.

Each common key makes some queries cheap and others expensive:

Choose the key from the query you run most often. For Instagram that is usually "everything for this user", which points to user_id.

Even keys, uneven traffic#

Hashing user_id gives each shard about the same number of users. It does nothing about how much traffic each user brings.

What teams actually do about it:

• Cache in front. The celebrity's profile and recent posts are the most cacheable rows you have. Most of those 40,000 reads should never reach the shard.
• Extra read replicas for the hot shard only. Sharding and replication combine: each shard can have its own replicas, and the busy one can have more.
• Handle the largest accounts on a separate path. Feeds commonly do this. Posts from accounts with millions of followers are merged in at read time instead of being copied to every follower when they are written. The news feed problem goes into this.

Uneven B2B tenants are the same problem with a different name: put your largest customers on dedicated shards on purpose, instead of hoping the hash spreads them out.

Queries without the key visit every shard#

A query that doesn't include the partition key becomes a scatter-gather. The router sends it to every shard, waits for every answer, and merges them.

A scatter-gather query is only as fast as its slowest shard, and more shards make a slow one more likely. Suppose each shard has a 1% chance of a slow response at any moment (a GC pause, a compaction, a noisy neighbour):

Adding shards to scale writes makes every scatter-gather query slower at the tail. If a common read has to scatter, either the key is wrong or that read should be precomputed: a secondary index sharded by the other key, or a separate store kept up to date as writes happen.

Adding a shard later#

The obvious placement rule is shard = hash(key) % N. It spreads keys evenly and needs no lookup table. It falls apart the day you change N.

Consistent hashing places both shards and keys on a ring, and each key belongs to the next shard clockwise. A new shard takes over one arc, so only the keys in that arc move: about 1/(N+1) of them on average. In the clip D landed on a short arc and took only 6 keys, which shows the catch: with one position per shard the arcs come out uneven. Real implementations give each shard many positions (virtual nodes) so every shard owns many small arcs and the load evens out. This is the standard choice when nodes join and leave often, as in a distributed cache.

Most databases take a different route, logical shards. Create far more partitions than you have machines, say 1,024 buckets on 8 nodes. A key's bucket is hash(key) % 1024 and never changes. A small table maps each bucket to a node. Growing means moving whole buckets to a new node and editing that table.

Moving a bucket while it is taking writes follows the same steps whichever scheme you use:

1. Start writing to both the old and the new location.
2. Copy the existing rows to the new location in the background.
3. Compare the two copies until they match.
4. Switch reads to the new location.
5. Stop writing to the old location and delete it.

You can stop and roll back after any of those steps, which is what makes this safe to do on a live system.

Worked example: sharding Instagram's posts and likes#

Back to the hypothetical Instagram. Assume 50 million daily active users, each of whom creates 2 posts and 20 likes a day. Peak traffic is 3× the daily average. From the load test above, one primary handles 5,000 writes a second, and we plan to run each shard at 60% of that (3,000) to leave headroom.

Write rate

• Posts: 50M × 2 = 100M a day. 100M ÷ 86,400 s ≈ 1,157 a second, or about 3,472 at peak.
• Likes: 50M × 20 = 1B a day ≈ 11,574 a second, or about 34,722 at peak.
• Peak total: about 38,194 writes a second.

Shard count

38,194 ÷ 3,000 ≈ 12.7, so 13 shards. Using 1,024 logical buckets, 10 machines hold 79 buckets and 3 hold 78. Going to 16 machines later means moving buckets and editing the table, with no rehashing.

Storage

Posts at about 1 KB each come to 100 GB a day. Likes at about 50 bytes each come to 50 GB a day. Together that is 150 GB a day, or roughly 55 TB a year: about 4.2 TB per shard per year across 13 shards.

Keys

Shard posts by user_id, because the profile page reads one user's posts. Shard likes by post_id, because the common read is "how many likes does this post have". Then check for hot shards: a celebrity's post can collect millions of likes, all on one post_id. Keep that count in a cached counter that is flushed periodically instead of running COUNT(*) against one shard.

How this comes up in an interview#

"Why not add more replicas?" Because the bottleneck is writes and every replica applies every write. Say this before you say "shard", together with the write rate you estimated.

"What's your partition key?" Name the key, the query it keeps on one shard, and one query it turns into a scatter-gather. Interviewers want to hear what you gave up as well as what you got.

"What happens when one user is huge?" Hot shard. Cache, more replicas for that shard, or a separate path for the biggest accounts.

"How do you add capacity next year?" Logical shards or consistent hashing, then the dual-write, backfill, verify, switch-over steps. Mentioning that % N moves most keys shows you have seen the problem before.

"How would you run a query across all users?" Avoid it on the request path. Precompute it, or accept a scatter-gather only for rare internal tools.

Next: keeping each shard alive#

Each shard is still one machine that can die. Making each one survive a failure means keeping copies on other machines, and once there are copies you have to decide what a reader sees right after a write. That is the next concept, replication and quorums.