Start practising

SYSTEM DESIGN · PATTERNS · 3 OF 7

Scaling Writes: Sharding, Queues, Batching and Write-Optimised Stores

What to do when a system must absorb a flood of writes: the signals of a write-heavy question, why one primary database tops out, and the standard moves (cheaper writes, sharding by a good key, a queue to absorb bursts, batching and pre-aggregation, sharded counters for hot keys, write-optimised stores), with the trade-offs and how to say it in an interview.

  1. 1
  2. 2
  3. 3
  4. 4
  5. 5
  6. 6
  7. 7
1 / 15

Press play to hear the narration, or step through at your own pace.

Notes

The signals

Listen for data that never stops: drivers sending location every four seconds, every ad click recorded, metrics from a hundred thousand servers, likes on a viral post. Do the math. A million active drivers reporting every four seconds is two hundred and fifty thousand writes a second. When the estimate lands above ten thousand writes a second, or bursts ten times the average, this pattern applies.

Signals in the question
  • every device reports every
  • few seconds (drivers, IoT)
  • every click, view or like
  • is recorded
  • logs and metrics ingestion
  • chat at WhatsApp scale
  • estimate shows 10k+ writes/s
  • or bursts 10x the average

The naive design

The naive design writes straight into one primary database. Each commit flushes the write-ahead log and updates every index. A single relational primary handles somewhere in the low tens of thousands of small writes a second, often less with many indexes. Read replicas do not help at all, because each replica replays every write too. Under a burst, locks pile up and requests time out.

Why one primary tops out
  • each commit: write-ahead log
  • flush + every index updated
  • one relational primary: low
  • tens of thousands of small
  • writes/s, often less
  • replicas do not help: they
  • replay the same writes
  • bursts → lock waits, timeouts,
  • lost data if clients give up

Move 1: cheaper writes

Move one: make each write cheaper. Every index is extra work per write, so drop the ones nothing queries. Insert in batches, because one statement with five hundred rows costs far less than five hundred statements. Append new event rows instead of updating a row in place. And storage engines built on LSM trees, like Cassandra or RocksDB, turn random writes into sequential ones. Concepts episode twenty-three explains why.

Make each write cheaper
  • drop indexes nobody queries
  • batch: 1 insert of 500 rows
  • beats 500 inserts
  • append, do not update in
  • place (event rows)
  • async commit where losing
  • the last ms is acceptable
  • LSM-tree stores turn random
  • writes into sequential ones
  • (Concepts episode 23)

Move 2: shard

Move two is the real fix: shard. Split the data by a key, like hashing the driver ID, so each shard owns a slice of the keys and takes only its share of the writes. If one shard safely handles ten thousand writes a second, two hundred fifty thousand needs about twenty-five, so plan thirty-two. Each shard keeps its own replicas for failover. Concepts episode twenty-one covers sharding in depth.

Sharding
shard = hash(driver_id) % N

250,000 writes/s
÷ 10,000 per shard (safe)
= 25 shards, plus headroom
→ plan 32

each shard has its own
replicas for failover

Choosing the key

Everything rides on the shard key. It needs many distinct values so load spreads evenly, and most queries should touch only one shard. Sharding by date is the classic mistake: every write today hits the same shard. Shard by user, device or order instead, and for time series combine device and day. Use consistent hashing so adding shards moves only a fraction of the keys.

A good shard key
  • high cardinality, even spread
  • most queries need only 1 shard
  • bad: date (all writes hit
  • today), country, status
  • good: user_id, device_id,
  • order_id, (tenant, user)
  • time-series: (device, day)
  • resharding: consistent hashing
  • (Concepts episode 22)

Move 3: buffer

Move three: put a durable log in front of the database. The ingest service appends each event to Kafka and acknowledges as soon as the log has it, a few milliseconds. Writer processes consume at their own pace, batch, and insert. When a burst arrives at ten times normal, the backlog grows instead of the database falling over. The cost: data shows up in the database seconds later. See Concepts episode twenty-eight.

Queue in front of the DB
  • ingest appends to Kafka and
  • acks once the log is durable
  • one Kafka partition: tens of
  • MB/s of appends
  • burst of 10x? the backlog
  • grows, nothing is dropped
  • writers consume, batch and
  • insert at a steady rate
  • reads lag by seconds

Move 4: aggregate

Move four: aggregate before you write. If you only need totals, the writer keeps counts in memory and flushes every second. Nine thousand click events for one ad become a single update adding nine thousand. That is a thousand-fold reduction. Keep the raw events in the log or in object storage, so you can recompute if a writer crashes before flushing.

Pre-aggregate
raw: 9,000 click events/s
  for ad 7

writer keeps a map in memory:
  counts[ad_7] += 1
every 1 s: flush
  UPDATE ad_7 SET n = n + 9000

9,000 writes → 1 write
keep raw events in the log
  or object storage for replay

Hot keys

Hot keys break even a sharded design. One viral post gets fifty thousand likes a second, and they all land on one row and one shard. The fix is a sharded counter: split the counter into sixty-four sub-counters, add to a random one on each write, and sum them on read, caching the total. Fifty thousand a second becomes about eight hundred per row.

Hot keys: sharded counters
one viral post: 50k likes/s
one row → lock contention

split it into 64 sub-rows:
  likes:post9:0 ... :63
write: pick one at random
  INCR likes:post9:rand(64)
read: sum the 64 (cache it)

50k/s ÷ 64 ≈ 800/s per row

Move 5: the right store

Move five: choose a store built for writes. Cassandra, Scylla and DynamoDB spread writes across many partitions, and the first two use log-structured storage, so they absorb hundreds of thousands of writes a second. Column stores like ClickHouse ingest events for analytics. And sometimes the right answer is Kafka plus object storage, queried later. But keep payments and inventory in a relational database, where transactions matter more than raw speed.

Move 5: the right store
  • append-heavy, key lookups:
  • Cassandra, ScyllaDB (LSM),
  • DynamoDB (auto-partitioned)
  • metrics and events:
  • ClickHouse, time-series DBs
  • raw firehose: Kafka + object
  • storage, query later
  • keep money and inventory in
  • a relational DB: they need
  • transactions more than speed

Failure modes

Failure handling. A writer can crash halfway through a batch, so commit the Kafka offset only after the database write succeeds. That means some events will be delivered twice, so make writes idempotent, for example with the event ID as a unique key. Alert on consumer lag; if it keeps growing, add partitions and writers. And if a shard fails, the queue keeps accepting while a replica takes over.

Failure handling
  • writer crashes mid-batch
  • → commit Kafka offset only
  • after the DB write
  • → redelivery: make writes
  • idempotent (event_id key)
  • consumer lag keeps growing
  • → add partitions + writers,
  • alert on lag
  • shard down → its replica
  • promotes, queue holds writes

Trade-offs

The trade-offs. Sharding makes cross-shard joins and transactions painful, so design queries around the shard key. A queue smooths bursts but reads are seconds behind. Aggregating loses individual events unless you keep the raw stream. Sharded counters make reads add up many rows. And the fastest write stores offer weaker consistency. Say which of these your product can accept.

Trade-offs
  • sharding: cross-shard joins
  • and transactions get hard
  • queue: durable and smooth,
  • but reads lag by seconds
  • aggregation: lose detail and
  • up to 1 s on a crash
  • (unless raw is kept)
  • sharded counters: reads sum
  • N rows
  • leaderless stores: weaker
  • consistency

Where it shows up

Where it shows up: the ad click aggregator and top K are all about batching and stream aggregation. The likes counter is the sharded counter problem. Uber and food delivery ingest a location every few seconds per driver. Monitoring ingests millions of metrics. WhatsApp handles billions of messages a day. And the unique ID generator exists so writes do not need a central counter.

Problems that use it
  • Ad click aggregator: batching
  • Likes counter: sharded counts
  • Uber: location every 4 s
  • Monitoring: metrics firehose
  • WhatsApp: billions of msgs
  • Unique ID generator: ids for
  • every write without a
  • central counter
  • Top K: stream aggregation

Say it like this

Here is the phrasing. Writes peak at two hundred fifty thousand a second, too many for one primary. The ingest service appends to Kafka and acknowledges. Writers batch into Cassandra, sharded by driver ID so load spreads evenly. Hot counters are split into sub-counters. Every write is idempotent by event ID, and I alert on consumer lag. Each sentence answers a follow-up before it is asked.

In the interview
  • "Writes peak at 250k/s, too
  • much for one primary. Ingest
  • appends to Kafka and acks.
  • Writers batch into Cassandra,
  • sharded by driver_id. Hot
  • counters are split into 64
  • sub-counters. Writes are
  • idempotent by event_id, and
  • I alert on consumer lag."

Interview questions

Why do read replicas not help with write scaling?

Every replica must apply every write, so replicas multiply read capacity but not write capacity; to scale writes you must split the data across independent primaries with sharding.

How do you choose a shard key for a write-heavy table?

Pick a high-cardinality key that spreads writes evenly and that most queries include, such as user_id or device_id. Avoid timestamps or low-cardinality fields, which send all current writes to one shard.

What does putting Kafka in front of the database buy you?

A durable, fast append path that absorbs bursts, decouples producers from the database, and lets consumers batch writes at a steady rate. The cost is that the database lags behind by seconds.

How do you handle a counter that receives tens of thousands of increments per second?

Use a sharded counter: split it into many sub-counters, increment a random one on each write, and sum them on read, usually caching the total. Alternatively aggregate increments in memory and flush once per second.

How do you avoid double-counting when consumers retry?

Commit the consumer offset only after the write succeeds and make writes idempotent, for example by using the event ID as a unique key or by recording processed IDs, so redelivered events have no extra effect.

When would you choose Cassandra over PostgreSQL?

For very high write volume with simple key-based or time-ordered access, multi-region needs and tolerance for eventual consistency. Keep PostgreSQL for relational data that needs transactions, joins and constraints.

What do you monitor in a write pipeline?

Ingest rate, consumer lag per partition, write latency and error rate per shard, shard size and hotness, and dead-letter queue depth, with alerts when lag keeps growing.