Part II — Intermediate: Distributed Systems Core

What changes when data and traffic no longer fit on one machine: splitting data, copying it, moving work asynchronously, keeping multi-step operations correct, drawing service boundaries, and staying reliable, observable, and secure. Part II ends with hands-on data modeling practice that uses everything so far.

9. Partitioning & Sharding

9.1 Strategies

Strategy	How	Pros	Cons
Range	Key ranges per shard (A–F, G–M...) or time ranges	Efficient range scans	Hotspots (all new writes go to the "latest" range)
Hash	`shard = hash(key) mod N`	Even distribution	Range queries scatter; changing N moves almost every key
Consistent hashing	Keys and nodes on a ring; virtual nodes	Adding a node moves only ~1/N of keys	Still needs vnodes for balance; hot keys remain hot
Directory / lookup	Mapping table key → shard	Flexible, can move individual tenants	Lookup service is critical and must be cached
Geo / tenant-based	Shard by region or customer	Data residency, isolation, cells	Uneven tenant sizes (whales)

The CDN guide §6 has a runnable consistent-hashing implementation, including a hash-quality bug that's worth reading about.

9.2 Choosing the partition key

The partition key must:

Spread load evenly (high cardinality, no celebrity skew).
Keep together what's queried together (so common queries hit one shard).
Rarely or never change (moving a row between shards is a distributed transaction).

System	Good key	Bad key	Why
Chat messages	`(conversation_id, time_bucket)`	`message_id`	Reading a conversation would scatter across all shards
Orders	`customer_id`	`order_date`	Today's date is a hot shard
Multi-tenant SaaS	`tenant_id` (+ split whales)	`user_id`	Tenant queries and isolation need locality
IoT telemetry	`(device_id, day)`	`device_type`	Low cardinality → few huge shards

9.3 Hot partitions and the celebrity problem

Even with a good key, real traffic is Zipfian: a few keys get most of the load. Mitigations:

Key salting: write to key#0..key#9, and read by merging all 10. Use it only for keys known to be hot.
Split hot tenants onto dedicated shards (directory-based routing).
Cache in front of hot reads; batch or aggregate hot writes (e.g. count likes in memory and flush every second).
Adaptive capacity (DynamoDB does this automatically; know that it exists and that it has limits).

9.4 Secondary indexes on sharded data

Local (document-partitioned) index: each shard indexes its own data. Writes are cheap, but a query by the secondary attribute must ask every shard (scatter-gather), so tail latency grows with shard count (see the fan-out math).
Global (term-partitioned) index: the index is partitioned by the indexed value. Reads go to one place; writes update a remote index, usually asynchronously, so the index is eventually consistent.

9.5 Resharding

Plan it from day one:

Start with many logical shards (e.g. 4,096) mapped to few physical nodes, and move logical shards between nodes. Never rehash keys.
Online move: dual-write or replicate (CDC) to the new location → backfill → verify (checksums) → switch reads → switch writes → clean up.
Vitess (MySQL), Citus (Postgres), and the managed NewSQL databases automate much of this. Saying so is a valid "buy" answer.

9.6 🏗️ Architecture: logical shards, physical hosts, and a routing layer

             App servers
                 │  query with shard key (tenant_id / user_id)
                 ▼
        ┌──────────────────┐     shard map (cached, versioned; source of truth in etcd/config DB)
        │ Routing layer    │◀──  logical_shard = hash(key) % 4096
        │ (proxy or lib:   │     logical_shard → physical cluster
        │ Vitess/DBProxy)  │
        └───┬───────┬──────┘
            ▼       ▼
     ┌──────────┐ ┌──────────┐ ┌──────────┐
     │ Host A   │ │ Host B   │ │ Host C   │   each host = primary + replicas
     │ ls 0-1365│ │ls1366-   │ │ls2731-   │   each logical shard = schema/DB/table group
     └──────────┘ └──────────┘ └──────────┘

 Online move of logical shard 17 from A → C:
   1. Snapshot ls17 on A → restore on C           4. Brief write freeze on ls17 (ms-s)
   2. Stream changes A→C (binlog/CDC)              5. Flip shard map ls17 → C (versioned)
   3. Verify (row counts, checksums)               6. Unfreeze; drain A; delete ls17 on A later

9.7 🌍 In production: how real companies sharded

Company	Scheme	Notable detail
Instagram (2012, "Sharding & IDs at Instagram")	Thousands of logical shards as Postgres schemas spread over a few physical servers	64-bit IDs generated in Postgres: 41 bits ms-timestamp + 13 bits logical shard ID + 10 bits sequence. The ID itself says which shard holds the row, so no lookup is needed.
Pinterest (2015, "Sharding Pinterest")	MySQL, 4,096 virtual shards	64-bit ID = shard ID + type + local ID; objects never move, so the ID is a permanent address
Notion (2021 / 2023)	480 logical shards (schemas), 15 per database on 32 hosts, later re-sharded to 96 hosts	Picked 480 because it divides evenly by 32, 40, 48, 96, ...
Figma (2024)	"Colos": groups of related tables sharded the same way so joins stay local; DBProxy routes queries	Logical sharding first (route as if sharded while still on one DB, using views), then physical sharding. The first split hit primary availability for only about 10 s.
Slack (2020)	Moved from per-workspace MySQL shards to Vitess, with sharding keys per table (e.g. by channel)	Driven by cross-workspace features
Meta Shard Manager (SOSP 2021)	A general framework: apps define shards, and SM handles placement, load balancing, and failover	Planned events (deploys, maintenance) happen ~1,000× more often than unplanned failures, so graceful shard migration matters more than crash handling
Google Slicer (OSDI 2016)	Auto-sharding service that assigns key ranges to tasks and rebalances hot ranges automatically	Moves the key → task mapping, not data; useful for in-memory caches and stateful services

9.8 ⚠️ Corner cases & deep details

Global uniqueness (usernames, emails) when sharded by user ID: the unique constraint only works per shard. Use a separate uniqueness table sharded by the unique value (insert there first; it acts as a reservation).
Cross-shard queries: "top 10 posts across all users" means scatter-gather across N shards, merging per-shard top-10s. Tail latency equals the slowest shard. Precompute it into a separate store instead.
Cross-shard pagination: keep a cursor per shard (or a global sort key) and merge-sort. Offset pagination across shards is O(N × offset).
Cross-shard transactions: design them out (colocate by key, as Figma's colos do), or use sagas or the DB's 2PC (Vitess has limited support).
Changing the shard key (e.g. user → organization when a B2B tier launches) means rewriting every row's location. It's a migration project (§29), not a config change. That's why the shard key is a one-way door.
Consistent backups across shards: each shard's backup is taken at a different moment, so restoring "the whole system to 14:00" isn't consistent across shards. You need coordinated snapshots or logical timestamps, or you accept per-shard point-in-time recovery.
Celebrity tenants: one tenant grows to 40% of a shard. You need directory-based override routing to move a single tenant, which hash-based routing can't do.
Hash choice: hash(key) % N with a poor hash, or with keys that share structure, skews load. The CDN guide §6 has a real hash-quality bug. Use a well-mixed hash (xxHash, Murmur3, or SHA for safety).