Scaling writes

Reads can be copied. Writes cannot, because every copy must see them. When a single primary is full, the options change character: you either do less work per write, smooth the bursts, or split the data.

Crux: reduce and smooth writes first. Split the data only when one machine truly cannot cope, and then the partition key is the decision that matters.

The problem

Signs that writes are the limit:

• the primary's CPU, disk or write throughput is saturated, while replicas sit idle
• bursts (a sale, a game launch, a viral post) push latency up
• the data no longer fits one machine

Ask: what is the peak write rate, how big is each write, how long is the data kept, and can a write be late or dropped? A like count can be a little late. A payment cannot.

Options, from simple to complex

Do less per write

• Remove indexes you do not need. Each index is extra work on every write.
• Keep rows small and avoid rewriting large fields.
• Use a store built for the write pattern: an append-only log or a write-optimised database (often LSM-tree based) copes with more writes than a general relational one, at the cost of other abilities. Name the property you need before naming a product.

• A bigger machine (faster disks, more cores and memory) is the cheapest first move and needs no redesign. It has a ceiling and costs more per unit of capacity as you climb, so use it to buy time while you do the real work.

Batch writes

Combine many small writes into one larger one. One insert of 500 rows is far cheaper than 500 inserts.

• Use when: data can wait a moment: events, clicks, metrics, logs.
• Careful: a batch held in memory is lost if the process dies. Decide how much loss you accept, or buffer it durably.
• Aggregate before writing: for counters, add up in memory (or in a stream processor) and write the total every second or so. Ten thousand clicks on one link become one update, not ten thousand.

Aggregate in layers. When thousands of sources report the same kind of number (page views from many web servers), each source adds up its own counts locally, a middle layer adds up groups of sources, and only the combined totals reach the database. Every layer cuts the write rate by the size of its group. The price is delay (each layer waits for a window) and loss on a crash unless a layer is durable.

Buffer with a queue

Put a queue or log between the producers and the database. The queue accepts bursts, and consumers write at a steady rate the database can take.

• Use when: load is bursty and a short delay before the data is visible is fine.
• Careful: a queue does not add capacity; it only moves the problem in time. If the average rate exceeds what the database can write, the queue grows forever. The caller also no longer learns immediately whether the write succeeded.

Shed load

When the system is overloaded, choose what to drop rather than letting everything slow down. Reject or defer low-value writes (analytics, telemetry) to protect important ones (orders, payments). Return a clear "try later" response and make clients back off.

Vertical partitioning

Split by kind of data: put users, orders and logs into separate databases. Each has its own write capacity.

• Use when: the data naturally separates and rarely needs joining.
• Careful: joins across databases are gone, and a transaction across them is hard.

Sharding (horizontal partitioning)

Split one table across many databases, each holding a shard: a slice of the rows chosen by a partition key.

Ways to map key to shard:

• Hash partitioning: hash(key) mod N, or better a consistent hash ring, which moves only a small share of keys when you add a shard. Spreads load evenly. Range queries need every shard.
• Range partitioning: shard 1 holds A to F, and so on. Range queries are cheap. Recent data or sequential IDs pile onto one shard.
• Directory (lookup table): a map says which shard owns each key. Flexible, but the directory is one more thing to scale and protect.

What a good partition key does:

• Spreads writes evenly (high cardinality, no single dominant value).
• Keeps data that is read together on one shard (all of one user's rows).
• Matches the main access pattern, so most requests touch one shard.

Costs to say out loud:

• Hot shards and hot keys. One celebrity user or one popular product takes all the load of its shard. Mitigate by adding a small random suffix to the key and combining the pieces on read, or by giving the hot key its own shard.
• Cross-shard queries and transactions are slow or hard. Design the key so the common ones stay on one shard.
• Resharding (moving data when you add shards) is operationally heavy. Consistent hashing or many small virtual shards from the start make it easier.
• Unique IDs can no longer come from one auto-increment counter.

A worked example: counting clicks

Assumption: 10,000 clicks per second at peak, each one recorded against its short code (an assumption for the exercise).

• Naive: one row update per click is 10,000 writes per second on the hottest rows. Popular links become hot rows (see contention).
• Step 1, buffer: each click goes to a log. Producers never wait on the database.
• Step 2, aggregate: consumers add up clicks per code per second in memory. If in one second 10,000 clicks hit 800 distinct codes, we write at most 800 updates, not 10,000. The exact ratio depends on how clustered the traffic is, which we do not know, so we call it an estimate.
• Step 3, shard if needed: if the aggregated writes still outgrow one database, partition by hash(code). A hot code gets split across a few sub-counters (code#0 to code#3), summed on read.
• Trade-off: counts lag by a second or two, and a crash may lose up to one window unless the log is replayed. For a click counter, that is acceptable. For money it would not be.

Choosing

• First: tune, remove indexes, batch.
• Bursts, not sustained overload: a queue.
• Overload with a priority order: shed load.
• Separable data: vertical partitioning.
• One logical dataset beyond one machine: shard, with a considered key.

Do not shard for scale you cannot justify with numbers. It is much harder to undo than to add.

When to bring this up, and when not to

• Signals in the prompt: a very high event rate (clicks, telemetry, location pings), a launch or sale with a sudden spike, a dataset that grows without bound.
• Not needed when the write rate is a few hundred per second. A single well-indexed database handles that. Do the arithmetic first (see estimation).

Deep dives interviewers use

Adding shards later

With hash mod N, changing N moves almost every key. Better routes:

• A consistent hash ring, where a new shard takes a slice from its neighbours only.
• Many small logical shards (say, 1,000) mapped onto fewer machines. Adding a machine moves whole logical shards, and the key-to-logical-shard rule never changes.
• The move itself: copy the data in the background, send new writes to both places (or replay a change log), check they match, flip the route, then delete the old copy. Plan for reads and writes to keep working throughout.

One key too hot for a shard

A single account, product or live event can exceed one shard whatever the shard count. Two ways to split it:

• Split every key in advance: append a small suffix to all keys (key#0 to key#7), writing to a random one and summing on read. Simple, but every read now costs eight lookups, even for cold keys.
• Split only the hot keys: detect them (a counter of request rate per key, or a list you maintain) and spread just those. Reads of cold keys stay cheap, at the cost of detection logic and a lookup that must know which keys are split.

Either way, the read side becomes an aggregation, so this suits counters and logs and not values that must be read as one exact item.

Which store to pick

Describe the write pattern, then the property. Append-only events and time series suit log-structured or time-series stores. High-volume key lookups suit a partitioned key-value or wide-column store. Needing multi-row transactions pulls you back to a relational database, and then to sharding it yourself or a distributed SQL option. State the trade, not just the product.

Overload and what to protect

Decide the order of sacrifice before the incident: drop telemetry, then delay analytics, then slow browsing, and protect payments last. Shed at the edge (rate limits per client, a cheap rejection) so the rejected work costs little, and tell clients to back off with a retry-after hint.

Common mistakes

• Sharding first, when batching or a queue would do.
• A partition key with low cardinality (a country, a boolean) or one that is monotonic (a timestamp) and funnels all writes to one shard.
• Forgetting hot keys.
• Using a queue as if it were extra capacity.
• Mod-N hashing, so adding one shard reshuffles nearly every key.
• Not saying what happens to queries that span shards.
• Dropping writes silently, with no signal to the client.

Check yourself

• Name four things to try before sharding.
• Why does a queue not fix sustained overload?
• What makes a good partition key?
• How do you deal with one hot key on one shard?
• Why is consistent hashing better than hash mod N when shards change?
• What do you give up when you shard?