Scaling writes

Scaling writes is the playbook for systems that take in data faster than one database can safely store it: GPS pings, ad clicks, metrics, chat messages, likes. You make each write cheaper, batch them, put a durable log in front for bursts, partition by a key that really spreads, store history in an append-friendly engine, defuse hot counters, and stop writing what nobody reads. Every rung buys throughput with latency, durability or read cost; the skill is naming which.

Picture it first: the post office in Rakhi week

In Rakhi week a town post office takes in thousands of envelopes an hour. One clerk writes each into a register, updates four card indexes, and stamps the register after every line. The office improves step by step:

1. Drop the two card indexes nobody uses, and stop stamping every line.
2. Enter letters in bundles of a hundred, one stamp per bundle.
3. Put a drop box at the door, so the 5 p.m. rush waits there while clerks work at their own pace.
4. Split the counters by PIN code, not by arrival hour, or every clerk but one sits idle.
5. Keep a day-book that clerks only append to; file it into cabinets overnight.
6. Give each counter its own tally sheet for “rakhis to Delhi” and add them up at closing.
7. Keep only the latest “reached Agra” postcard.

Map it back: the register is the primary database, card indexes are secondary indexes, the stamp is a commit, bundles are batches, the drop box is a log, PIN-code counters are hash partitions, the day-book is an LSM tree, tally sheets are sharded counters, and the postcard is last-write-wins.

Spotting the pattern

Listen for these cues:

• “Ingest N events per second”, or a fleet reporting on a timer: phones, meters, cars, servers.
• Append-mostly data rarely read row by row: logs, clicks, metrics, location trails.
• A write:read ratio near or above 1: rows written once and read only in aggregate, or never.
• Bursts tied to a moment: a sale at midnight, a last-over six. Discord could read the 2022 World Cup final’s goals off its message-send graph.
• Counters on popular things: likes and views on one post, votes.
• Fan-out on write: one post copied into 200 followers’ timelines.

It does not apply when:

• Reads dominate. Product pages and feeds belong to scaling reads.
• Many people want one scarce thing. A seat or the last phone in stock is about who wins, not volume: contention and double booking.
• The rate is small. 500 writes per second fits one indexed PostgreSQL node.

The running example is a food-delivery platform’s location pings. Assume 500,000 delivery partners online at the dinner peak, each phone pinging every 4 seconds: 500,000 ÷ 4 = 125,000 writes per second. A ping is about 100 bytes, so bandwidth is only 12.5 MB/s; the number of writes is the problem. If the daily average is 40% of peak, that is 50,000 × 86,400 ≈ 4.3 billion pings a day, about 430 GB raw, or 13 TB for 30 days. Three readers: dispatch and the customer’s map need each partner’s latest position; support and pay need each order’s trail; surge pricing needs pings per zone per minute.

Levels 0 and 1: one database, cheaper writes

Level 0: the version that breaks

The first draft inserts each ping as its own transaction on one PostgreSQL primary, into a table with a surrogate bigserial key and three more indexes (partner and time, order, geohash). Each ping writes a row, four index entries and their WAL, then waits for the commit to reach disk. PostgreSQL’s bulk-loading guide warns that committing each row separately means “a lot of work for each row”.

Use a simple cost model (planning assumptions; benchmark your own): each transaction costs a fixed F ≈ 100 µs of the node’s capacity (round trip, parse, commit, flush wait) plus r ≈ 10 µs per row with four indexes. One node does 1 ÷ 110 µs ≈ 9,000 writes per second, 14 times short. It worsens with time: once indexes outgrow memory, each index insert can cost a random disk read.

Level 1: make each write cheaper

• Index only what the hot path reads. The trail is read by partner and time. Make (partner_id, ts) the primary key and drop the surrogate: one index instead of four, and a replayed ping becomes a no-op (database indexes).
• Append, never update. Insert-only tables avoid row locks and dead tuples (vacuum and bloat).
• Relax the commit for data you can lose. With synchronous_commit = off, PostgreSQL acknowledges before the WAL is flushed. A crash can lose up to three times wal_writer_delay of acknowledged commits, 600 ms at the 200 ms default; the docs stress this is data loss, never corruption, and name event logging as a fair use. If you must stay synchronous, commit_delay lets one flush serve several commits (group commit).
• Partition the table by day. Dropping an expired partition is, in the docs’ words, “far faster than a bulk operation” and avoids the VACUUM work of a bulk DELETE (table partitioning).
• Scale up: NVMe disks and more memory raise the ceiling once (vertical vs horizontal scaling).

Say fewer indexes halve the row cost (r = 5 µs) and asynchronous commit halves the fixed cost (F = 50 µs). You gain 1 ÷ 55 µs ≈ 18,000 writes per second, double. You pay with lookups served some other way and up to 600 ms of pings lost in a crash. It breaks because 125,000 ÷ 18,000 is still about 7: every ping still pays for a whole transaction.

Say: “Pings are append-only, so I keep one natural-key index, partition by day and accept losing half a second of pings in a crash.”

Level 2: batch and buffer

The fixed cost is per transaction, so put many rows in each. A writer service collects pings in memory and flushes every 100 rows or 20 ms, whichever comes first, as one multi-row INSERT or COPY; PostgreSQL’s docs say COPY is almost always faster than INSERT, even batched. Throughput becomes B ÷ (F + B × r) for B rows per transaction.

With 25 writer instances at 5,000 pings per second each, 100-row batches fill in 20 ms, and the database sees 125,000 ÷ 100 = 1,250 transactions per second instead of 125,000.

CREATE TABLE pings (
  partner_id bigint           NOT NULL,
  ts         timestamptz      NOT NULL,
  order_id   bigint,
  lat        double precision,
  lng        double precision,
  PRIMARY KEY (partner_id, ts)        -- the only index; also the idempotency key
) PARTITION BY RANGE (ts);

-- one transaction per batch of 100 rows
BEGIN;
SET LOCAL synchronous_commit = off;   -- may lose the last ~600 ms in a crash
INSERT INTO pings VALUES (...), (...), (...)   -- 100 rows
ON CONFLICT DO NOTHING;               -- a replayed ping changes nothing
COMMIT;

DROP TABLE pings_2026_09_27;          -- retention: drop a day, no DELETE

Brokers batch the same way. A Kafka producer groups records per partition up to batch.size (16 KB by default) and waits up to linger.ms for more; Kafka 4.0 raised that default from 0 to 5 ms, as bigger batches usually cost no extra latency. A producer spread over many partitions sends small batches; a 50 ms linger is invisible next to a 4-second ping interval.

You gain about ten times the throughput, and fewer, larger disk writes. You pay up to one batch window of delay and, if you acknowledge before the flush, up to one window of data per writer in a crash: 20 ms × 5,000 = 100 pings. It breaks at the node’s per-row ceiling, 1 ÷ r = 200,000 rows per second, and at the first burst: when arrivals outrun the flushes, the buffers fill and the phones time out.

Say: “I batch 100 rows per transaction, about 20 ms of delay, which takes one node from 18,000 to about 180,000 rows a second.”

Level 3: absorb bursts with a log

Bursts come from moments, not averages. Say a two-minute mobile-data outage hits one city. Its 100,000 partners each buffer 30 pings (120 ÷ 4) and upload them within 20 seconds of reconnecting: 150,000 more per second, so 275,000 per second arrive at a database that tops out near 182,000. Writes time out, phones retry, and retries add load (retry storm).

Put a durable log between the ingest API and the stores: the queue-based load leveling pattern. The API appends each ping to a Kafka topic and returns; writers consume at a pace the database can take. As Azure’s guidance notes, you then provision consumers for the average load, not the peak (message queues).

The arithmetic: the backlog grows at 275,000 − 160,000 = 115,000 per second for 20 seconds, 2.3 million pings, just 230 MB of log. It drains at 160,000 − 125,000 = 35,000 per second, so it is empty after about 66 seconds. A ping arriving at the peak waits 2.3M ÷ 160,000 ≈ 14 seconds before it is written.

What users see. The phone gets “OK” in milliseconds, once the log has the ping. A map fed from this backlog would freeze for 14 seconds, so the latest position gets its own consumer (Level 6) that skips stale pings while catching up.

Durability is a setting. With acks=all, Kafka’s default, the leader waits for every in-sync replica, and a record survives while one of them does. With acks=1, a leader that dies before followers copy the record loses it. The idempotent producer, also on by default, stops retries from duplicating or reordering records. Consumers still see at-least-once delivery, so writers commit offsets after their database write and let the (partner_id, ts) key absorb replays (idempotent consumers).

Bound everything. A queue hides overload; it doesn’t remove it. If brokers are unreachable, the producer’s 32 MB buffer.memory holds about a minute of one ingest server’s 500 KB/s, then send() blocks for up to max.block.ms (60 s) and fails. Act sooner: tell phones to ping every 8 seconds, halving the load (backpressure and load shedding).

You gain survival of bursts, independent consumers, and a replayable history. You pay with lag, at-least-once duplicates, ordering only within a partition, and a cluster to run. It breaks when the average arrival rate exceeds the drain rate: the backlog never empties.

Say: “Ingest appends to Kafka with acks=all and returns. A burst becomes up to 14 seconds of lag, not an outage.”

Level 4: partition by a key that spreads

When the average outgrows one database, split the writes. The partition key decides whether the split works (database sharding).

Hash a high-cardinality ID. Key the Kafka topic by partner_id: the producer picks the partition from a hash of the key, so each partner’s pings stay in order on one partition. With 48 partitions, each carries 125,000 ÷ 48 ≈ 2,600 pings per second, shared by up to 48 consumers. Shard the trail store by the same hash. Planning at 70% of our model’s 182,000 rows per second, about 126,000 per shard, one shard would sit right at that line at today’s peak; two leave room for a storm, and four cover a doubled fleet, at about 3.3 TB each. Place keys with consistent hashing so adding a fifth shard moves about a fifth of the data.

Never key by time or an auto-increment ID in a range-partitioned store: every new write lands in the newest range. Watch skewed keys too: a city, a celebrity, a hashtag.

Per-partition ceilings make this concrete. A DynamoDB partition delivers at most 1,000 write units per second, a unit being one write of up to 1 KB. Key pings by city, and Bengaluru’s 25,000 writes per second are 25 times that limit; key by minute, and all 125,000 land on one key. The documented fix is write sharding: append a suffix such as 2026-10-27T20:01#1 to #200, and each key takes 125,000 ÷ 200 = 625 writes per second. A random suffix spreads writes evenly; a calculated one, such as hash(partner_id) mod 200, lets you read a single partner’s item back directly. Reading the whole minute then costs 200 queries and a merge.

You gain write capacity that grows with partitions. You pay with ordering only per key, cross-shard queries that fan out, and rebalancing work. It breaks when one key is hotter than one partition, which no number of partitions fixes.

Say: “I partition by hash of partner ID: even load, and per-partner order is preserved. Time or city would put the load on one partition.”

Level 5: use write-optimised storage

A B-tree database updates pages in place: each insert finds the right leaf page in every index, reading it from disk if it isn’t cached, and later writes the page back. With billions of rows, those are random I/Os. An LSM tree, the design behind Cassandra, ScyllaDB and RocksDB, turns writes into appends. In Cassandra each write goes to a sequential commit log and a sorted, in-memory memtable; full memtables flush to immutable SSTables, and background compaction merges them (how databases store and fetch data, write-ahead logging).

The bill moves elsewhere. Write amplification: each ping is written to the commit log, flushed once, then rewritten by every compaction it passes through. Read amplification: a read may check the memtable and several SSTables. And Cassandra blocks writes when memtables fill faster than they flush. Discord’s message store, partitioned by channel and a static time bucket, grew to 177 Cassandra nodes and trillions of messages, with hot partitions and compaction backlogs. After moving to ScyllaDB, a Cassandra-compatible engine without a garbage collector, it needed 72 nodes, and p99 insert latency went from 5–70 ms to a steady 5 ms.

For the trail, use PRIMARY KEY ((partner_id, day), ts). The partition key spreads load and bounds partition size: a 10-hour shift is 36,000 ÷ 4 = 9,000 pings, about 900 KB per partition. Metrics and sensor data have time-series databases built on the same append-and-compact idea.

You gain high write throughput per node and easy horizontal scaling. You pay with costlier reads, compaction tuning, and no joins (Cassandra and DynamoDB). It breaks when compaction falls behind or reads dominate.

Say: “The trail is write-once and read by partner and day: a natural fit for Cassandra.”

Level 6: hot keys and counters

Partitioning spreads keys, not the traffic on one key. Three tactics cover most hot spots.

Overwrite what only matters as “latest”. Dispatch and the map need where a partner is now. A separate consumer keeps one Redis key per partner, loc:<partner_id>, overwritten on each ping. At 125,000 updates per second and a planning figure of about 100,000 operations per second per shard (see scaling reads), four shards give headroom. The trap: storm uploads replay old pings, and a blind overwrite moves a rider backwards on the map. Write only if the ping is newer, and while catching up keep only each partner’s newest ping per batch. If Redis loses a key, the next ping rebuilds it within 4 seconds.

SET_IF_NEWER = """
local cur = redis.call('HGET', KEYS[1], 'ts')
if cur and tonumber(cur) >= tonumber(ARGV[1]) then return 0 end
redis.call('HSET', KEYS[1], 'ts', ARGV[1], 'lat', ARGV[2], 'lng', ARGV[3])
redis.call('EXPIRE', KEYS[1], 60)        -- offline partners disappear
return 1
"""

set_if_newer = r.register_script(SET_IF_NEWER)   # r: a Redis client

def on_batch(pings):                       # one poll from the pings topic
    newest = {}
    for p in pings:                        # conflate: newest ping per partner
        if p.partner_id not in newest or p.ts > newest[p.partner_id].ts:
            newest[p.partner_id] = p
    for p in newest.values():
        set_if_newer(keys=[f"loc:{p.partner_id}"], args=[p.ts, p.lat, p.lng])
    consumer.commit()                      # offsets after writes: at-least-once

A geo set per city is a hot key too: Bengaluru’s 100,000 partners ÷ 4 s = 25,000 updates per second on one key. Key it by geohash cell (proximity search).

Aggregate before you write. Surge pricing counts pings per zone per minute. At dinner, 8,000 partners in one busy zone send 2,000 pings per second, and UPDATE zone_stats SET pings = pings + 1 makes each wait for the previous row lock. At an illustrative 5 ms per lock, one row manages about 200 updates per second, ten times too few (why one hot row tops out). Instead, each of 25 consumers sums counts in memory and flushes once a second: 25 writes per second per zone, 80 times fewer, for up to a second of delay. A stream processor does this with windows (Flink and stream processing).

Shard the counter. When increments must land individually, such as likes on a viral post, split the counter into N rows, add to a random one, and sum on read. Google’s Firestore guidance says write throughput grows linearly with shards, and so does read cost; 16 shards at 200 each take 3,200 updates per second. Show a roll-up total refreshed every few seconds. For distinct counts, such as unique partners per zone per hour, a HyperLogLog sketch takes fixed, small memory however many IDs it sees (count-min sketch and HyperLogLog).

You gain 16 to 80 times less load on a hot key. You pay with counts a second or more behind, reads that sum shards, and timestamp checks. It breaks when a number must be exact the instant it changes; then one row arbitrates, at its lock rate.

Say: “Latest position is overwritten, newest timestamp wins. Counters are summed in memory for a second, or sharded 16 ways.”

Level 7: write less

The cheapest write is the one you skip.

• Send less. Idle partners, with no order and not moving, can ping every 30 seconds instead of 4. If 40% are idle, that removes 200,000 × (1/4 − 1/30) ≈ 43,000 pings per second, about a third of the load.
• Dedupe at the edge. Drop retried pings with an already-seen (partner_id, ts) and repeated positions from a parked phone.
• Roll up and expire. Keep 4-second points for 7 days for disputes, then one point per minute, 15 times fewer rows, and drop everything after 30 days. Storage falls from 13 TB to about 7 × 430 GB + 23 × 29 GB ≈ 3.7 TB.
• Compress batches. Kafka producers can compress whole batches (gzip, snappy, lz4, zstd); bigger batches compress better.
• Sample what is statistical. Web-analytics and debug-log events can be sampled; billable ad clicks and payments cannot.

You gain fewer writes, less storage, lower cost. You pay in detail you can never get back. Say: “Idle partners ping every 30 seconds and history is rolled up after a week: a third less load, over two-thirds less storage.”

Putting it together

Every rung that buys throughput can lose or reorder something. State it per path (delivery guarantees).

Which classic problems use it

Common mistakes

• Keying by timestamp or auto-increment ID. The newest partition takes every write while the rest sit idle.
• Read replicas for a write problem. Every replica replays every write, so replicas add no write capacity.
• Writing several stores from the request: the dual-write problem above.
• Unbounded queues with no backpressure. Lag grows until the data is useless or deleted unread. Alarm on lag; slow or shed producers.
• UPDATE … SET n = n + 1 on one row for likes or views. One row is a queue of lock waits; aggregate or shard.
• Batching without idempotency. A retried batch or a replayed offset double-writes unless rows have a natural key.
• Storing history on the hot path when only the latest value is read.

Cheat sheet

• Throughput with batches = B ÷ (F + B × r); the ceiling is 1 ÷ r.
• Backlog = (arrivals − drain) × burst seconds; drain time = backlog ÷ (drain − normal arrivals).
• Worst lag ≈ peak backlog ÷ drain rate.
• One hot row ≈ 1 ÷ lock time; local aggregation turns N increments into one per flush.

How to say it in the interview

“This is write-heavy: 500,000 partners pinging every 4 seconds is 125,000 writes a second, and a city-wide reconnect can more than double it for twenty seconds. I split the data by how it’s read. The ingest API appends each ping to Kafka, keyed by partner ID with acks=all, and replies once the log has it. The latest position is overwritten in Redis, newest timestamp wins, so dispatch and the map read one key. A trail writer upserts every ping into Cassandra, keyed by partner and day and rolled up after a week. Zone counts for surge are summed in memory and flushed every second. A burst becomes at most about 14 seconds of trail lag, never an outage.”

Why not just use a bigger database, or add replicas? Replicas replay every write, so they add no write capacity. A bigger machine raises the ceiling once: even batched, one node near 180,000 rows a second has no room for a 275,000-a-second storm, a doubled fleet or 13 TB a month.

A trail writer crashes halfway through a poll. What is lost? Nothing. It commits Kafka offsets only after every ping in the poll is written, so a restart replays from the last committed offset, and rows keyed by (partner_id, ts) simply overwrite themselves.

How do you keep one partner’s pings in order? The partner ID is the Kafka key, so a partner’s pings share a partition, and the idempotent producer keeps retries in order. Across partners there is no order, and none is needed; the position store compares timestamps, so a late upload can’t move a rider backwards.

One post gets a million likes a minute. How do you count them? About 16,700 increments a second on one counter. Store each like as a row keyed by post and user, so a double tap counts once; sum increments in memory for a second, add them to one of 16 counter shards, and show a roll-up that may lag a few seconds.

The backlog keeps growing. What do you do? Check whether average arrivals now exceed the drain rate. Add consumers up to the partition count, as far as the store can absorb; collapse position updates to the newest per partner; and shed load by telling phones to ping every 8 seconds. Never let retention delete unread data.

Where to go next

• Scaling reads: the companion playbook for read-heavy systems.
• Contention and double booking: when many writers want the same row.
• Message queues and Kafka: the log that absorbs bursts.
• Backpressure and load shedding: what to do when the backlog won’t drain.
• Database sharding and table partitioning: splitting the write path.
• How databases store and fetch data: B-trees versus LSM trees.
• Cassandra and DynamoDB: partition keys in depth.
• Batch vs stream processing: where aggregation and roll-ups run.

References and further reading

1. PostgreSQL documentation: Populating a databasepostgresql.orgWhy one commit per row is expensive, COPY versus batched INSERT, and the cost of maintaining indexes and foreign keys during large loads.
2. PostgreSQL documentation: Asynchronous commitpostgresql.orgsynchronous_commit = off loses at most three times wal_writer_delay (200 ms by default) of acknowledged commits, never corrupts; commit_delay as group commit.
3. PostgreSQL documentation: Table partitioningpostgresql.orgDropping or detaching a partition is far faster than a bulk DELETE and avoids its VACUUM overhead.
4. Apache Kafka 4.1 documentation: Producer configskafka.apache.orgbatch.size (16 KB), linger.ms (5 ms since 4.0), acks=all versus acks=1, idempotence on by default, key-hash partitioning, buffer.memory and max.block.ms.
5. Azure Architecture Center: Queue-Based Load Leveling patternlearn.microsoft.comA queue as a buffer so consumers run at their own pace; sizing for average rather than peak; queues grow when producers outpace consumers; at-least-once delivery.
6. Amazon DynamoDB Developer Guide: Designing partition keysdocs.aws.amazon.comEach partition delivers at most 1,000 write units per second; one write unit is one write per second of up to 1 KB.
7. Amazon DynamoDB Developer Guide: Using write shardingdocs.aws.amazon.comRandom and calculated suffixes on a hot partition key, and the read-side cost of querying every suffix and merging.
8. Apache Cassandra documentation: Storage enginecassandra.apache.orgWrites go to the commit log and a memtable, flush to immutable SSTables, and block when flushing cannot keep up.
9. How Discord Stores Trillions of Messages (Discord engineering, 2023)discord.comMessages partitioned by channel and time bucket; compaction and hot-partition pain on 177 Cassandra nodes; 72 ScyllaDB nodes with a steady 5 ms p99 insert.
10. Firebase documentation: Distributed countersfirebase.google.comA counter split into shards summed on read; write throughput grows linearly with shards while read cost grows with them too.