1. The problem: why one database server eventually isn't enough

In the replication lesson we scaled reads by adding replicas. But every replica stores all the data and applies every write. Three limits remain:

• Data size. Assume Instagram stores 2 billion users (~1 KB each = 2 TB), 100 billion posts' metadata (~500 B each = 50 TB), and trillions of likes (hundreds of TB). A single server tops out at tens of TB of fast storage — and long before that, indexes stop fitting in RAM, backups take days and a replica rebuild takes a week.
• Write throughput. Assume ~60,000 likes/s on average and over 100,000/s at peak, plus posts, comments and follows. One primary can sustain perhaps 5,000–20,000 write transactions/s depending on hardware. Replicas don't help: they repeat the writes.
• Blast radius. With one primary, one bad disk, one runaway query or one botched migration affects 100% of users. With 1,000 shards, a single shard failing affects ~0.1%.

The answer is to partition the data: give each server responsibility for a subset of the rows. That's sharding. (Terminology: MongoDB and Elasticsearch say shard; Cassandra says partition or token range; HBase says region; Bigtable tablet; CockroachDB range; Couchbase vBucket. Same idea.)

2. Vertical partitioning vs horizontal sharding

There are two ways to split a database, and real systems use both:

• Vertical partitioning splits by columns or by tables. (a) Move rarely-used or bulky columns of a table into a separate table (e.g. users core fields vs user_profile_extras with a long bio and settings JSON), so the hot table is narrow and cache-friendly. (b) At the architecture level, give each service its own database: users in one cluster, comments in another, DMs in a third — functional partitioning, the natural outcome of microservices. It's simple and often the first step, but each table/service still lives on one machine, so it doesn't fix a single table that's too big or too write-heavy.
• Horizontal partitioning (sharding) splits by rows: every shard has the same schema but holds a different subset of rows — users 0–9,999,999 on shard 1, the next 10M on shard 2, or rows placed by a hash. This scales without a ceiling, at the cost of the complications covered in the rest of this lesson.

3. Choosing a shard key

The shard key (partition key) is the column whose value decides which shard a row lives on. It's the most consequential and least reversible decision in a sharded design — changing it later means moving essentially all the data. Evaluate candidates against these criteria:

1. High cardinality. Many distinct values, far more than the number of shards. user_id (billions) ✅; country (~200) ❌ — you can never have more than 200 useful shards, and they'd be wildly uneven.
2. Even distribution of data and load. Not just row counts — request rates too. A key can spread rows evenly yet concentrate traffic (one celebrity's rows get 10,000× the reads).
3. Query locality. The hottest queries should include the shard key, so they hit one shard. Data that's read together should share a shard key so it's co-located.
4. No write hotspots. Monotonically increasing keys (timestamps, auto-increment IDs) under range sharding send all new writes to the last shard.
5. Immutable. If the key value can change (e.g. username, country of residence), the row must move between shards on update. Use a stable surrogate ID.
6. Supports growth. The scheme must let you add capacity without rewriting everything (Section 6).

3.1 Instagram shard-key decisions

4. Sharding strategies

4.1 Range-based sharding

Each shard owns a contiguous range of key values: users 1–100M on shard A, 100M–200M on B, and so on (or usernames a–f, g–m…). A small routing table lists the boundaries.

• ✅ Range queries are efficient — "all events between 10:00 and 11:00" touches one or two shards.
• ✅ Easy to split a shard that grows too large (cut the range in half) — HBase, Bigtable, CockroachDB and Spanner split ranges automatically.
• ❌ Hotspots with sequential keys: if the key is a timestamp or auto-increment ID, every new write lands on the last range. Also skew when values cluster (many usernames start with "s").

4.2 Hash-based sharding

Apply a hash function to the key and use the hash to pick the shard: shard = hash(user_id) mod S. A good hash (MurmurHash, xxHash, MD5) scatters even sequential keys uniformly.

• ✅ Even distribution of data and writes, no sequential hotspots.
• ✅ Routing is pure computation — no lookup table.
• ❌ Range queries die: adjacent keys land on random shards, so "users 1000–2000" hits every shard.
• ❌ Naive mod N resharding moves almost everything: change S from 4 to 5 and ~80% of keys change shard. The fixes are a fixed number of logical partitions (Section 6) or consistent hashing.
• Hybrid used by Cassandra: hash the partition key to pick nodes, but keep rows sorted by clustering key inside the partition — so "posts of user X between dates" is still a range scan within one partition.

4.3 Directory (lookup-table) sharding

A directory service stores an explicit mapping key (or key group) → shard. Routing = look up the key, then go to that shard.

• ✅ Maximum flexibility: move one big tenant/user to a dedicated shard, place shards on heterogeneous hardware, rebalance one key at a time.
• ❌ Extra hop on every request (mitigate with aggressive caching in each app server); the directory must be highly available and consistent — it becomes critical infrastructure (often stored in ZooKeeper/etcd or a replicated DB).
• Common in multi-tenant SaaS (tenant → shard) and as the "logical → physical" map in Instagram's design, where the directory is tiny because it maps thousands of logical shards, not billions of keys.

4.4 Geographic sharding

Place each user's data in a shard located in (or near) their region: EU users in Frankfurt, Indian users in Mumbai.

• ✅ Low latency for the user's own data; data-residency compliance (GDPR, national data laws).
• ❌ Uneven (one region may have 10× the users); people travel and move (home region must be migratable); social interactions cross regions constantly (a Brazilian liking a Japanese post), so many operations become cross-shard and cross-continent anyway. Usually combined with hash sharding within each region. See Multi-Region Architecture.

 RANGE  (key = user_id)                 HASH  (shard = hash(user_id) mod 4)
   [1 ……… 100M]   → shard A               user 17   → h=…3A → shard 2
   [100M … 200M]  → shard B               user 18   → h=…F1 → shard 1   (neighbours scatter)
   [200M … 300M]  → shard C  ← all new    user 19   → h=…07 → shard 3
                               signups!   user 20   → h=…C4 → shard 0

 DIRECTORY (lookup table)               GEO (by home region)
   tenant "acme"   → shard 7              user in Germany  → EU shard group
   tenant "globex" → shard 2              user in India    → APAC shard group
   user 42 (celebrity) → dedicated shard  user in Brazil   → LATAM shard group

5. Hotspots and the celebrity problem

Even with a perfect hash, sharding balances keys, not traffic per key. A single key can be orders of magnitude hotter than average — and a key always lives on one shard. Instagram's extreme case: accounts with hundreds of millions of followers.

Assume a celebrity posts and the post gets 10 M likes in the first hour.
  10,000,000 / 3,600 s  ≈ 2,800 like-writes/s — all with the same post_id → one shard.
  Peak in the first minutes may be 10× that: ~28,000 writes/s on ONE shard,
  while the average shard handles a few hundred. Meanwhile 50 M viewers read the
  post's like count and comments → hundreds of thousands of reads/s on the same shard.

Mitigations, usually combined:

• Cache hot reads. The post, its counts and top comments go into Redis/Memcached (and media into the CDN); the shard sees a tiny fraction of reads. See Caching.
• Split hot keys ("salting"). Write likes for a hot post to post_id#0 … post_id#15, chosen randomly or by hash(user_id) % 16, so they spread over up to 16 shards; reads aggregate the sub-keys. Do this only for keys detected as hot — it multiplies read cost.
• Counter aggregation. Don't increment a row per like: buffer increments in memory or a stream processor and apply +N once per second. See Likes & Counters.
• Change the algorithm for whales. The feed uses fan-out-on-write for normal users but fan-out-on-read for celebrities, so one post doesn't trigger 600M timeline writes.
• Isolate. Use the directory to move the hottest accounts' logical shards onto dedicated, beefier servers.
• Rate-limit and queue non-urgent writes (notifications about the likes can be batched: "Alice and 9,999 others liked your post"). See Rate Limiting.

6. Logical vs physical shards, and resharding

6.1 Why naive resharding hurts

Suppose you start with 4 database servers and route with user_id mod 4. A year later you need 5. Under mod 5, a key stays put only if id mod 4 == id mod 5, which is true for just 4 of every 20 consecutive IDs — 80% of all rows must move, while the system is serving traffic. Doubling from 4 to 8 is kinder (exactly 50% move), but still means copying half your data at once. The consistent hashing lesson works through this arithmetic in detail.

6.2 The fix: many logical shards, few physical servers

Decouple the two concerns:

• Logical shards — a large, fixed number of buckets (e.g. 4,096) chosen once and never changed. Keys map to logical shards with a function that therefore never changes: logical = user_id mod 4096.
• Physical shards — the actual database servers. A small mapping table assigns logical shards to servers: with 8 servers, each hosts 512 logical shards.

To grow, you move whole logical shards from crowded servers to new ones and update the mapping. No key ever changes logical shard, so nothing is re-hashed; you only move as many logical shards as needed to balance the new server. Data movement is also simple, because a logical shard is a self-contained unit (a schema, a database, a set of files).

6.3 Instagram's publicly described approach

NUM_LOGICAL_SHARDS = 4096                 # fixed forever (13-bit field allows up to 8192)

# Small mapping table: cached in every app server, stored in a config service (ZooKeeper/etcd)
LOGICAL_TO_PHYSICAL = {}                  # logical shard -> database host
for ls in range(NUM_LOGICAL_SHARDS):
    LOGICAL_TO_PHYSICAL[ls] = f"pg-db-{ls // 512 + 1:02d}"   # 8 servers × 512 logical shards

def shard_for_user(user_id: int) -> int:
    return user_id % NUM_LOGICAL_SHARDS

def shard_for_object(obj_id: int) -> int:
    # Instagram-style ID: [41 bits time][13 bits shard][10 bits sequence]
    return (obj_id >> 10) & 0x1FFF

def connection_for_user(user_id: int):
    ls = shard_for_user(user_id)
    host = LOGICAL_TO_PHYSICAL[ls]
    return get_pool(host), f"shard_{ls:04d}"   # query tables inside schema shard_NNNN

# A new post by user 42 is created on shard_for_user(42); its ID embeds that shard,
# so later "GET /p/<post_id>" routes with shard_for_object(post_id) — no directory lookup.

6.4 How many logical shards?

Enough that you'll never need more physical servers than logical shards (e.g. if you might one day run 1,000 servers, choose ≥ 4,000 logical shards so each server can still hold several and you can balance finely), but not so many that per-shard overhead (connections, schemas, file handles, monitoring) explodes. Thousands is the common sweet spot. Elasticsearch, Riak and Couchbase (1,024 vBuckets) and Redis Cluster (16,384 hash slots) all use this "fixed number of partitions" approach.

7. Cross-shard queries, scatter-gather and joins

A query that includes the shard key goes to one shard — fast and isolated. A query that doesn't must be sent to every shard and the results merged: scatter-gather.

 "Find posts with caption containing #sunset created in the last hour"  (no shard key!)

                       +--------------------+
   client ---------->  | query router /     |
                       | coordinator        |
                       +--------------------+
            scatter  /     |      |      \   (same query to all N shards, in parallel)
                    v      v      v       v
               shard 1  shard 2  shard 3 … shard N
                    \      |      |       /
            gather   v     v      v      v   (each returns its top-k; coordinator merges,
                       +--------------------+   re-sorts, applies LIMIT)
                       |  merge + sort + k  |
                       +--------------------+
 Latency = the SLOWEST shard.  Load = N × (one query).  Availability = all N must answer.

Why it's expensive:

• Load multiplies. With 1,000 shards, one query becomes 1,000 queries. 1,000 such requests per second = a million shard queries per second.
• Tail latency dominates. If each shard is slow (above its p99) 1% of the time, the chance that at least one of 100 shards is slow is 1 − 0.99100 ≈ 63%. Your median scatter-gather query is as slow as a single shard's p99.
• Pagination and sorting are awkward: for "top 20 overall" each shard must return its top 20, and deep pages get expensive.

Strategies:

1. Design so hot queries include the shard key. This is the whole point of choosing the key from your access patterns.
2. Denormalise / duplicate by another key — store likes by post_id and by user_id.
3. Use a purpose-built secondary system for queries that inherently span everything: search goes to Elasticsearch, analytics to the data warehouse.
4. Precompute — maintain aggregates (counts, leaderboards) incrementally from events instead of computing them across shards on request.
5. Bound fan-out — scatter to only the shards that can contain matches (e.g. range sharding by time), use timeouts and return partial results when acceptable.

7.1 Joins across shards

A SQL join between rows on different servers can't be executed by either database alone. Options:

• Co-location: shard related tables by the same key so joins are shard-local — posts and media both sharded by the author's user_id (Citus calls these co-located tables; Vitess uses keyspace IDs).
• Reference (broadcast) tables: small, rarely-changing tables (countries, hashtag categories, feature flags) are copied to every shard.
• Application-side joins: fetch post IDs from shard A, then batch-fetch authors from their shards (multiget), and stitch in code — typically with a cache in front. This is how feeds are "hydrated".
• Denormalisation: copy the author's name/avatar into the data you read.

8. Secondary indexes in a sharded database: local vs global

The shard key decides where a row lives. But what about lookups by another column — e.g. "find user by username" when users are sharded by user_id? There are two ways to index a non-shard-key column:

Instagram example: login by username and @mention resolution happen constantly, so a global index — a username → user_id mapping sharded by username — is the right call. Claiming a username is a single-row insert on that index shard with a uniqueness constraint (the single authority for that value), followed by creating the user row. "Posts tagged at location X" is rarer and fits better in the search/discovery system than as a global index on the posts shards.

9. Distributed transactions: two-phase commit, and why to avoid it

Within one shard you have ordinary ACID transactions. An operation that must atomically change rows on two shards needs a distributed transaction. The classic protocol is two-phase commit (2PC):

 Coordinator                 Shard A (follower list)        Shard B (following list)
     | ---- PREPARE (tx 77) ---->|                                |
     | ---- PREPARE (tx 77) ---------------------------------------->|
     |                           | write intent, lock rows,       | write intent, lock rows,
     |                           | fsync, vote YES                | fsync, vote YES
     | <-------- YES ------------|                                |
     | <-------- YES -------------------------------------------------|
     | (log decision: COMMIT)    |                                |
     | ---- COMMIT ------------->|  apply, release locks          |
     | ---- COMMIT ------------------------------------------------->| apply, release locks
 Any NO or timeout in phase 1 → coordinator sends ABORT to all.
 Coordinator crashes after PREPARE? Shards are stuck holding locks ("in doubt") until it recovers.

Why large systems avoid 2PC on hot paths:

• Blocking. A participant that voted YES can't decide on its own — if the coordinator dies, rows stay locked until it comes back.
• Latency. At least two round trips plus forced log writes on every participant; across regions that's hundreds of ms.
• Availability multiplies down. The transaction needs every participant up; with 3 shards each 99.9% available, the transaction is ~99.7% available.
• Throughput. Locks are held across network round trips, so contention rises sharply.

(Consensus-replicated systems like Spanner and CockroachDB make 2PC non-blocking by making each participant a replicated Raft/Paxos group, but the latency cost remains.)

What to do instead:

1. Design operations to be single-shard — the main reason for co-locating data by shard key.
2. Sagas: a sequence of local transactions, each publishing an event that triggers the next; if a later step fails, run compensating actions (e.g. release a reserved username).
3. Transactional outbox: in the same local transaction as the business change, insert an event row into an outbox table; a relay publishes it to Kafka. This guarantees "the DB change and the event either both happen or neither", without 2PC between the DB and Kafka.
4. Idempotent, retryable consumers so at-least-once delivery converges to the right state.

10. Rebalancing

Rebalancing moves data and load between nodes when nodes are added, removed or overloaded. Goals: afterwards load is fair; during the move the database keeps serving reads and writes; and no more data moves than necessary.

10.1 Moving a shard without downtime

1. Copy: snapshot the logical shard to the target server (or start a streaming replica of the source).
2. Catch up: stream ongoing changes (replication or CDC) until the target lags by only milliseconds.
3. Cut over: briefly block writes to that shard (seconds), let the target apply the final changes, flip the mapping in the config service, and let routers pick it up.
4. Verify and clean up: compare row counts/checksums, then delete the shard from the source.

Practical concerns: throttle copies so they don't starve production traffic; prefer human-approved rebalancing plans (fully automatic rebalancing combined with automatic failure detection can cascade — a slow node is declared dead, its data is moved onto other nodes, which become overloaded…); and make sure routers handle "shard moved" errors by refreshing their mapping.

10.2 Request routing

Someone must know the mapping: (a) smart clients that cache the map (Instagram-style app-level routing, Cassandra drivers, Redis Cluster clients following MOVED redirects); (b) a routing tier (MongoDB mongos, Vitess vtgate, Citus coordinator); or (c) any node that forwards to the right one (Cassandra coordinators, gossip). The authoritative map usually lives in a consensus store (ZooKeeper, etcd, a config server).

11. How Instagram uses sharding

11.1 A request through the shards

1. Open profile @bob. The global usernames index (sharded by username) resolves bob → user_id 1872….
2. logical = user_id mod 4096 = 1337; mapping says logical shard 1337 lives on pg-db-27; a nearby read replica of pg-db-27 serves Bob's profile row and his newest 12 posts from schema shard_1337 — one shard, one index range scan.
3. Tap a post. The post ID's middle 13 bits say shard 1337 → same server. Like count and comments come from post_id-partitioned stores and caches.
4. Alice likes it. The like is written to the shard of the post; a copy keyed by Alice's user_id ("posts I liked") is written asynchronously via an event.
5. Feed. Bob's new post is fanned out by workers to each follower's timeline, keyed by the follower's user_id — touching many shards, but asynchronously and in batches.

Rough numbers (assumptions)

Assume 2 B users, ~1 KB of user row + indexes each       → ~2 TB of user data
Logical shards: 4,096                                      → ~490,000 users and ~0.5 GB per logical shard
Physical servers: 64 primaries (each + 2 replicas)          → 64 logical shards / server, ~32 GB / server
Peak user-DB writes 40,000/s ÷ 64 primaries                 → ~625 writes/s per primary (comfortable)
Growth to 128 servers: move 32 logical shards off each server (~16 GB each), no re-hashing.

Scatter-gather tax: a query without the shard key at 4,096 logical shards
  = up to 64 physical queries per request; at 1,000 req/s → 64,000 queries/s just for that feature.

12. Trade-offs & common pitfalls

Common pitfalls

• Sharding too early. Vertical scaling, read replicas, caching and query tuning go a long way. Shard when a single primary's write load or data size is truly the bottleneck — but design IDs and keys so sharding later is possible.
• Sharding too late. Migrating a live, overloaded monolith database is far harder than sharding a healthy one.
• A shard key that doesn't match the hot queries → everything becomes scatter-gather.
• Sequential keys with range sharding → all writes on one shard.
• hash mod N on physical servers → adding a server moves most of the data. Use logical shards or consistent hashing.
• Auto-increment IDs per shard → collisions across shards. Use globally unique IDs (shard-embedded or Snowflake-style).
• Ignoring per-key hotspots — celebrities, viral posts, default/empty keys (user_id = 0 for anonymous events all landing on one shard).
• 2PC everywhere to preserve monolith-style transactions → latency and availability collapse.
• Schema migrations forgotten on a few shards → shards drift. Automate migrations across all shards and verify.

13. Real-world technologies

14. Interview corner

15. Check your understanding

Related lessons

• Consistent Hashing — the other way to add nodes without moving most keys
• Replication — each shard is itself a replicated group
• Unique ID Generation — Instagram's shard-embedding 64-bit IDs
• Databases — PostgreSQL vs Cassandra partitioning models
• Indexing — how secondary indexes work inside a shard
• News Feed — fan-out across shards and the celebrity problem
• Social Graph — storing both directions of follow edges on different shards
• Message Queues & Kafka — events and the outbox pattern instead of 2PC