One database is fine when you start. But as your app grows, that single database becomes the part that everything waits on — the bottleneck. There are two main ways to fix this, and they solve different problems. Mixing them up is a classic interview mistake.
The animation on the right runs in two parts. First replication: writes go to one main database, copies flow to the other databases, and reads are spread across the copies (you will also see the small delay called "lag"). Then sharding: a router sends different groups of users to different databases.
Replication — make copies of the data (handles more reads)
Replication
Keep full copies of the database on several machines. One machine, the primary, takes all the writes (saving new data). The other machines, the replicas, get a stream of those same changes and answer read requests (fetching data).
Most apps read data far more often than they write it. So spreading reads across many copies is a big win. It also gives you a backup: if the primary dies, you can promote a replica to take its place.
The downside is replication lag. The copies are always a tiny bit behind the primary. So if a user saves something and then reads it right away, the copy might not have it yet, just for a moment. (This "you'll see it soon" behavior is called eventual consistency.)
Ways to set up replication (topologies)
| Setup | How it works | The catch |
|---|---|---|
| One primary, many replicas | All writes go to the primary; replicas are read-only | Simple, but the primary still handles every write alone |
| Many primaries (active-active) | Several machines accept writes and sync both ways | Hard: two writes can clash and you must resolve the conflict |
| Chain replication | Primary → Replica A → Replica B (changes flow down the chain) | Spreads the load well, but long chains add more delay |
Dealing with replication lag
Sometimes a user must see their own writes right away (they update their profile and expect it to show). Here are your options:
- Read from the primary right after a write — always correct, but you lose the benefit of spreading reads.
- Read from the same replica the write went to — works, but the routing logic gets tricky.
- Sticky reads — the user always reads from the same replica until the lag clears.
- Monotonic reads — each new read goes to a replica that is at least as up-to-date as the last one.
A short line to memorize: "Replication handles more reads and gives you a backup; sharding handles more writes." If the interviewer digs into sharding, talk about how you pick the shard key and how to avoid "hot" shards — that is where the real design thinking is.
Sharding — split the data (handles more writes)
Sharding (partitioning)
Split the data across several databases so each one holds only a slice. For example: users 1–100 on Shard A, users 101–200 on Shard B. Each shard handles its own reads and its own writes.
Replication cannot handle more writes, because every write still hits the one primary. Sharding can, because writes get spread across many shards. The price is real complexity: questions that need data from several shards are painful, and picking a shard key that spreads the load evenly (so no single shard gets overloaded) is the hard part.
Ways to split the data (sharding strategies)
Range-based sharding
Put rows into shards by a range of the key (for example, user IDs 1–1M → Shard 1, 1M–2M → Shard 2). Easy to understand, and it makes range questions simple. Risk: hot shards — if most new traffic uses the newest IDs, one shard gets hammered.
Hash-based sharding
Run the key through a hash function and use hash(key) % N to pick the shard. This spreads the load
evenly. Weakness: if you change N (the number of shards), almost every key moves — which consistent
hashing (Chapter 3) is designed to fix.
Directory-based sharding
Keep a lookup table that maps each key to a shard. This is the most flexible — you can move data between shards without changing any formula. Cost: the lookup table itself can become a bottleneck or a single point of failure unless you cache it heavily.
Picking a good shard key
A bad shard key creates "hot shards" — one shard drowning in traffic while the others sit idle. Avoid these:
- Timestamps — every new write lands on the most-recent-time shard.
- IDs that always go up (auto-increment) — same problem, all new writes pile onto one shard.
- Fields with few values — a
statusfield with only 3 possible values can never give you more than 3 shards.
Good shard keys have many different values and spread evenly: user_id, customer_id, or a UUID.
The cross-shard problem
Once data is split up, any operation that needs more than one shard gets expensive:
- JOINs across shards — you have to fetch from both shards and combine the results in your app code.
- Transactions across shards — you need distributed transactions (like two-phase commit) or you accept eventual consistency.
- Counting / totals —
SELECT COUNT(*) FROM usersmeans asking every shard and adding up the answers.
This is why you put off sharding as long as you can, and why the shard key choice matters so much.
The celebrity (hotspot) problem
Even a shard key with lots of different values can melt one shard. Say you shard a social network by user_id. The number of users per shard is even — but the traffic is not. Katy Perry and Justin Bieber get vastly more activity than a normal account. If they both land on the same shard, that shard maxes out while the rest stay quiet. This is a hotspot key, and even hashing does not fix it, because the imbalance is in how people use the data, not in the key itself.
| Symptom | Cause | Fix |
|---|---|---|
| One shard at 100% CPU, the rest idle | A few keys take most of the traffic | Give each celebrity their own shard |
| Read times spike for one user | All their followers hit the same shard | Cache that user's data heavily (Chapter 4) |
| One shard outgrows the size limit | Data keeps piling onto one key (one user's events) | Use a combined key (user_id + time bucket) |
The practical fix is to handle the heavy hitters separately: give each celebrity their own shard (or their own set of replicas) so they do not drag anyone else down.
Denormalize to avoid cross-shard JOINs
The cleanest way to "fix" a cross-shard JOIN is to never need one. Denormalize — copy the fields you would have JOINed against directly into the row — so a read touches just one table on one shard. You trade extra storage and extra write work for simpler, faster reads, which is the right trade when reads dominate. This is why NoSQL data is shaped around the question you'll ask ("store the data the way you'll read it"), not around tidy textbook tables.
When the interviewer says "but user X has 100M followers", bring up the celebrity problem on your own and suggest a dedicated shard plus a read-through cache. Showing that the bottleneck is the usage pattern, not the hash function, is a strong senior signal — junior candidates just say "add more shards," which does nothing for a single hot key.
CAP theorem — what you give up when the network breaks
Once data lives on more than one machine, the network will break sometimes (a switch reboots, a cable is cut, a node freezes). CAP tells you what you are allowed to keep when that happens.
Consistency (CAP)
Every read sees the most recent write, no matter which machine answers. All clients see the same data at the same moment.
Availability (CAP)
Every request to a working machine gets a real (non-error) answer — even if some machines are down.
Partition tolerance (CAP)
The system keeps working even during a partition — when machines cannot talk to each other.
A distributed system cannot guarantee all three at once. And here is the key point: partitions are not optional — networks really do fail. So you do not actually get to "pick 2 of 3." You only get to pick what to give up when a partition happens: consistency or availability.
| Type | Keeps | Gives up | What it does during a partition | Example |
|---|---|---|---|---|
| CP | Consistency + Partition tolerance | Availability | Block or reject requests rather than serve old data | Bank balances, inventory, locks |
| AP | Availability + Partition tolerance | Consistency | Keep answering, accept maybe-old reads, fix up later | Shopping cart, social feed, DNS |
| CA | Consistency + Availability | Partition tolerance | — | Cannot exist in the real world |
To make it concrete: say you have three copies n1, n2, n3 and n3 gets cut off. A CP store blocks writes to n1/n2 so nobody reads conflicting data — the system goes down until the network heals (a bank returns an error rather than show a wrong balance). An AP store keeps accepting reads and writes on n1/n2, serves whatever it has (maybe a little stale), and syncs n3 once the network comes back. Dynamo and Cassandra are AP by default.
CA is a trap answer. Saying "I'll build a CA system" tells the interviewer you think the network never fails. In any system that spans more than one machine, partition tolerance is a must — the real choice is CP vs AP, and it should follow the business: money and locks lean CP, feeds and carts lean AP.
Treat CAP as a per-operation business decision, not one label for the whole system: "Checkout's inventory count is CP — I'd rather fail the purchase than oversell. The product reviews are AP — a slightly old review count is fine, so keep serving." Showing that CAP can differ by data type inside one product is a senior move.
Quorum consensus — tuning consistency with N, W, R
AP systems do not throw consistency away entirely; they make it adjustable. The dials are three numbers, managed by a coordinator node that sits between the client and the copies.
N, W, R
N = how many copies (replicas) you keep of a key. W = the write quorum: a write succeeds once W copies say "got it." R = the read quorum: a read waits for R copies to answer.
W = 1 means one acknowledgment, not one copy. The data is still sent to all N machines — W=1
just means the coordinator reports success after the first "got it" instead of waiting for the rest.
Same idea for R.
The headline rule: if W + R > N, you get strong consistency. That is because the set of copies you read from and the set you wrote to are guaranteed to share at least one copy that holds the newest value. The classic balanced setup is N=3, W=2, R=2 (2 + 2 > 3).
| Setup | What you get | Why |
|---|---|---|
W + R > N | Strong consistency | The read set and write set always overlap |
W + R ≤ N | No strong-consistency guarantee | The read set might miss the newest write |
R = 1, W = N | Fast reads, slow writes | Read any one copy; every write waits for all N |
W = 1, R = N | Fast writes, slow reads | One write "got it" is enough; reads must check all N |
N = 3, W = 2, R = 2 | Balanced strong consistency | Survives one slow or down copy on each path |
The trade-off is speed vs consistency: a bigger W or R waits on the slowest copy in the group, so more consistency costs you slower responses. With W=1 or R=1 you return as soon as any copy answers.
Memorize "W + R > N gives strong consistency; N=3, W=R=2 is the default." Then add the nuance: "R=1, W=N is best for reads; W=1, R=N is best for writes — I'd pick based on the read/write ratio." That one inequality is the most quotable line in distributed-systems interviews.
Consistency models
Your quorum settings give you one point on a range of consistency models — the promise about what a read is allowed to return.
| Model | The promise | Used by |
|---|---|---|
| Strong | A read always returns the most recent write; clients never see old data | SQL databases, CP stores |
| Weak | A read might not see the most recent write | — |
| Eventual | A kind of weak: given enough time with no new writes, all copies catch up and match | Dynamo, Cassandra |
Strong consistency usually works by refusing reads and writes until every copy agrees — which blocks operations and hurts availability. That is why highly-available stores like Dynamo and Cassandra pick eventual consistency: they let mismatched values into the system and push the job of sorting out conflicting writes onto the client at read time (next section).
Conflict resolution with vector clocks
Eventual consistency means two clients can write the same key at the same time — johnSanFrancisco on one copy, johnNewYork on another. Now there are two versions and no built-in way to tell which is "newer." A vector clock lets you tell apart one came after the other (safe to overwrite) from they happened at the same time (a real conflict).
Vector clock
A list of [server, version] pairs attached to a piece of data, written D([Sx, 2], [Sy, 1]).
Writing D on server Si adds 1 to vi if [Si, vi] is already there, otherwise it adds
[Si, 1].
Comparing two clocks tells you the relationship:
- One came first (no conflict): version X came before Y if every counter in Y is ≥ the matching counter in X.
D([s0,1],[s1,1])came beforeD([s0,1],[s1,2])→ Y simply wins. - They clash (conflict): if Y has any counter lower than X's matching counter, the two happened at the same time.
D([s0,1],[s1,2])vsD([s0,2],[s1,1])→ neither came after the other, so the client must merge them (for example, combine two shopping carts) and write back the merged result.
Two real downsides:
- Extra work for the client — your app, not the database, has to write the merge logic.
- The list keeps growing — the
[server, version]list gets longer with every server involved. The fix is to cap the length and drop the oldest pairs; this can blur the history a little, but Amazon says it has not caused problems in practice.
Tell a vector clock apart from a plain timestamp: "Picking the latest by wall-clock time quietly throws away one of two simultaneous updates; a vector clock can detect that two writes happened at the same time and hand the conflict to the client to merge." Knowing when last-write-wins loses data is the senior distinction.
Failure detection with gossip
Before you can route around a dead machine, you first have to agree it is actually dead — and one machine's opinion is not enough. The simple approach, everyone pings everyone (all-to-all), is O(N²) and floods the network at scale.
Gossip protocol
A spread-out way to detect failures. Each machine keeps a membership list of (member ID, heartbeat counter), regularly bumps its own counter, and regularly sends its list to a few
random machines, which pass it along. A member whose heartbeat has not moved past a timeout is
marked down — and that conclusion spreads the same way.
Gossip spreads like a rumor (or a virus): the news reaches the whole cluster in about O(log N) rounds, with each machine talking to only a few others. That is why it scales where all-to-all pinging falls apart. Requiring several machines to confirm the "down" verdict keeps one flaky connection from declaring a healthy machine dead.
Handling failures: temporary vs permanent
How you respond depends on whether the machine is coming back.
Temporary failures — sloppy quorum + hinted handoff
A strict quorum blocks writes when too many copies are briefly unreachable. A sloppy quorum keeps the system running instead: skip the down machines and use the first W healthy machines on the ring for writes (and the first R healthy for reads).
Hinted handoff
When a copy is temporarily down, a stand-in machine accepts its writes and keeps a hint. When the original machine comes back, the stand-in hands the data back, restoring the intended set of copies. It pairs with sloppy quorum to keep data safe during short outages.
Permanent failures — anti-entropy with Merkle trees
If a copy is gone for good (a replaced disk, a rebuilt machine), you have to resync it from a peer without shipping the entire dataset. Anti-entropy compares copies and repairs the differences; a Merkle tree makes that comparison cheap.
Merkle tree
A tree of hashes where each leaf hashes a bucket of keys and each parent hashes its children. Two copies compare their root hashes first: if they match, the data is identical and you stop. If not, you only go down into the branches whose hashes differ — so the amount of data you transfer matches the difference between the copies, not their total size.
A common setup is one million buckets for one billion keys (about 1000 keys per bucket), so a few changed keys touch only a handful of buckets instead of forcing a full scan.
Split failure handling by how long it lasts: "For a machine that'll be back in seconds, sloppy quorum plus hinted handoff keeps us available. For a machine that's gone for good, anti-entropy with Merkle trees resyncs it — and because we compare hashes top-down, we only ship the buckets that actually differ." Naming both halves shows you know availability and durability are separate problems.
Storage engine internals (LSM / SSTable)
How does a single machine actually store all this data on disk? Cassandra-style engines use a log-structured merge (LSM) design that turns scattered, random writes into neat sequential ones.
Write path (saving data):
- Add the write to a commit log on disk (so it survives a crash).
- Apply it to the memtable, a fast, sorted structure kept in memory.
- When the memtable fills up past a limit, flush it to disk as an SSTable (Sorted String Table — a sorted list of
<key, value>on disk). SSTables are never changed once written.
Because writes only ever append (to the commit log and to new SSTables), they are sequential and fast — no jumping around the disk for each write.
Read path (fetching data):
- Check the memtable first; if the key is there, return it.
- Otherwise the key might be in any of many on-disk SSTables. Check a Bloom filter for each SSTable to skip the ones that definitely do not have the key.
- Read from the SSTable(s) the filter says might have it, and return the answer.
Bloom filter
A tiny structure that answers "is this key in the set?" It says either "definitely not here" or "maybe here" — it never misses a key that is present, but it can occasionally give a false "maybe." It lets the read path skip disk reads into SSTables that cannot hold the key, which is what keeps LSM reads fast even though data is spread across many files.
If asked why writes are fast in Cassandra/RocksDB: "It's log-structured — every write is a sequential append to a commit log and an in-memory memtable, then flushed to immutable SSTables. Reads use a Bloom filter per SSTable to skip files that can't contain the key." Linking "fast writes" to "sequential appends + files you never edit" shows you understand the engine, not just the API.
Resharding
When a shard grows too big or gets too hot, you have to split it. Resharding means:
- Moving half the data from the hot shard to a new shard.
- Updating the routing tables (or redeploying the consistent hash ring).
- Doing all of this while the system stays live.
Consistent hashing (Chapter 3) makes resharding less painful: only the keys in the affected slice move, not the whole dataset.
Estimating the shard count
"How many shards?" is a sizing question, and the interviewer wants to see you reason from throughput and a per-record size to a real number — not pull "let's say 100 shards" out of thin air. The recipe: bytes per day → total bytes over how long you keep the data → divide by what one shard can hold.
Say the system takes 100M writes/day, each row is about 1 KB, and you keep 5 years of history:
Writes/day = 100,000,000
Bytes/record ≈ 1 KB
Bytes/day = 100M × 1 KB = 100 GB/day
Days retained = 365 × 5 ≈ 1,825 (round to ~1,800)
Total raw = 100 GB × 1,800 = 180,000 GB ≈ 180 TBNow divide by what one shard can comfortably hold. A common ceiling is ~2 TB/shard (keep machines small enough to rebuild and back up quickly):
Shards = 180 TB / 2 TB = 90 shardsSo ~90 shards as a starting point. Then pad for reality — you never run a shard at 100%:
+ replication (×3 copies) → 270 TB stored, but shard COUNT does not change
(replicas hold the same 90 partitions, not new ones)
+ headroom (aim for 60% full) 90 / 0.6 ≈ 150 shards
+ indexes/overhead (×1.3 on disk) call it ~180–200 shards to be safeThe headline answer is "on the order of 100–200 shards," and the senior move is saying why the raw 90 is not the number you actually deploy. Note that replication multiplies storage, not shard count — a 3x replica factor means 270 TB on disk but still the same 90 logical partitions.
This ties straight back to shard-key choice: those 180 TB only spread evenly across 90 shards if the key has many more than 90 distinct values and spreads evenly. A key with few values or one that always goes up (timestamps, auto-increment IDs) collapses those 90 shards into a few hot ones — you provisioned 90 but really use 3. Size for the shard count, but the shard key decides whether that count is real.
Don't stop at the raw division (180 TB / 2 TB = 90). State your per-shard capacity assumption out
loud, then pad for replication (storage, not count), 60% headroom, and index overhead to land on a
deploy number — and immediately flag that the estimate only holds if the shard key spreads evenly.
Showing the gap between "math says 90" and "I'd provision ~150" is the senior signal; juniors quote
the bare division.
NoSQL databases and sharding
Many NoSQL systems (Cassandra, DynamoDB, MongoDB) come with sharding and replication built in:
| System | How it shards | How it replicates |
|---|---|---|
| Cassandra | Consistent hashing (virtual nodes) | Tunable (1 to all copies) |
| DynamoDB | Hash partitioning (managed for you) | Automatic across availability zones |
| MongoDB | Config-server lookup table | Primary/secondary sets |
| Vitess | Range/hash sharding of MySQL | MySQL replication |
These take away the manual work of sharding, but you still have to choose the partition key wisely.
The operational cost is real. Sharding is not a switch you flip — it is a migration project that takes weeks, needs careful data moving, and permanently changes how you query. Start with a big primary plus read replicas. Add caching (Chapter 4). Only shard once those two are used up.
When asked "how would you scale a database for 10M users?" walk through the steps: 1) add read replicas to take reads off the primary, 2) add a cache layer (Redis) to cut DB reads by about 90%, 3) if write throughput is still the bottleneck, then shard. This step-by-step answer shows you don't reach for complexity too early.
Practice
Check your understanding of replication, sharding, and consistency.
1. What problem does replication mainly solve?
2. Why does sharding help with writes when replication does not?
3. Which makes a GOOD shard key?
4. During a network partition, what does a CP system do?
5. Why does the rule W + R > N give strong consistency?