Scaling Writes

Reads are easy to spread across copies. Writes are not, because every copy has to apply every write. When write volume outgrows one database, you first make each write cheaper, then smooth out spikes, and only then split the data across machines. This note works through that ladder, from the cheapest fixes to sharding, which is the most expensive.

Start with the cheap wins#

Work through these in order, easiest first:

Async side effects#

An order flow shows the difference. On the left every step runs before the user gets a response. On the right only the critical path does, and background workers then handle email, inventory, analytics, warehouse and recommendations.

LSM trees and append-only logs#

• B-tree (most relational databases): find the right page on disk and update it in place. That's random I/O.
• LSM tree (Cassandra, RocksDB, and many NoSQL stores): write to an in-memory buffer (memtable) plus a sequential log, acknowledge, and flush sorted files to disk in the background. Writes are much faster; the price is extra read work and background compaction.
• Append-only log (e.g. Kafka): always write to the end of a file. That's sequential I/O and very high throughput. Reads have to find the latest entry for a key, and storage grows until compaction cleans it up.

Write buffering with queues and batches#

When traffic spikes to 50,000 writes/s and the database handles 10,000, put a message queue in between. Producers enqueue at any rate; workers drain the queue at a rate the database can handle and write in batches.

During the spike the queue grows by about 40,000 messages per second; after the spike it drains back to zero. The database never sees more than its 10,000/s. The user gets an instant "Order placed!" while the actual write lands a few seconds later.

Batching multiplies the gain. 100 single-row inserts at ~5 ms each cost about 500 ms; one 100-row insert might cost ~10 ms. Workers that pull 50 messages and write them in one call get both spike absorption and fewer round trips.

Choosing a queue#

Delivery guarantees#

The common production choice is at-least-once delivery plus idempotent processing. Give each message a unique request_id. The worker records which ids it has processed and skips repeats, or uses a conditional write that only succeeds once. Then duplicates are harmless.

def handle(message):
    # Conditional insert: only the first delivery
    # of this request_id succeeds.
    inserted = db.insert_if_absent(
        "processed_requests", message.request_id
    )
    if not inserted:
        return  # duplicate delivery, already handled
    apply_order(message.payload)

In a real system, put the dedup record and the business write in the same transaction so a crash between them can't leave you half-done.

Sharding#

Sharding (horizontal partitioning) splits rows across several databases, each owning a slice of the data and the writes for that slice. 100M users on one database capped at 10K writes/s becomes four shards of 25M users at about 2,500 writes/s each, and you can add more shards as you grow.

Choosing a shard key#

The shard key decides where each row lives. Get it wrong and one shard does all the work.

1. High cardinality. user_id has millions of values. is_active has two, and country has about 200 with very uneven traffic.
2. Even distribution. hash(user_id) spreads load evenly. Sharding by date sends all current writes to one shard.
3. Query isolation. Most queries should hit one shard. Ask which column appears in the WHERE clause of 90% of queries. That column is probably the key.
4. Co-locate related data. A user's profile, orders and settings share user_id, so "show the user's dashboard" stays on one shard.

Before committing, check: will one value get far more traffic than others (hotspot)? Will you need JOINs across different keys? If so, rethink the model.

Hotspots: the celebrity problem#

Even a good key can melt one shard. A celebrity posts, and millions of likes and comments arrive with the same post_id, all landing on one shard. Layered fixes:

1. Queue all likes and comments so the spike never hits the database directly.
2. Batch-aggregate: workers turn "1,234 likes this second" into one write.
3. Sharded counters: split like_count into, say, 10 sub-counters on different shards. Each write picks one at random, and reads sum them.
4. Key salting: store comments under post_id + random(1..10) so one post spreads across 10 partitions. Reads then fan out across those partitions.

The cost of sharding#

Before sharding you have one database where JOINs and transactions just work, and backups and debugging are simple. After sharding:

• you have N databases to run,
• cross-shard JOINs are impractical,
• cross-shard transactions need protocols like two-phase commit,
• non-key queries become scatter-gather,
• debugging starts with "which shard?", and
• resharding means migrating live data.

So exhaust simpler options first, in this order:

1. Better indexes and queries.
2. A bigger machine.
3. Read replicas (if reads are part of the load).
4. Caching.
5. Then shard.

Real signals that you need it:

• Write throughput exceeds the biggest single machine ("we need 50K writes/s; the largest instance does 10K").
• Data doesn't fit on one machine.
• The next bigger machine costs far more than the performance it adds.

Partitioning strategies#

Once you shard, how do you map keys to partitions?

The hash + range combination is how DynamoDB and Cassandra model data. With partition key user_id and sort key created_at:

• "All orders for user 1": one partition. ✓
• "User 1's orders in January": one partition, range scan. ✓
• "All orders in January, all users": every partition. ✗

Consistent hashing#

With hash(key) % N, changing N remaps most keys. Going from 4 to 5 shards moves about 80% of them. For a cache that's a burst of misses; for a database it's data you must physically migrate.

Consistent hashing places shards and keys on a ring (think 0° to 360°). Each key belongs to the first shard clockwise from its position.

Say the shards sit at A = 90°, B = 180°, C = 270° and D = 330°:

Now add shard E at 150°. Only keys in (90°, 150°] move, from B to E. Everything else stays put.

Adding a shard moves only about $1/N$ of the keys. Placing each physical shard at many points (virtual nodes, often 100–200 per shard) evens out how much of the ring each one owns, so one shard doesn't end up with 40% while another has 10%.

Moving data is still real work for a database: copy, verify, switch traffic, delete the old copy, all while serving requests. Managed stores like DynamoDB do this automatically when they split partitions.

Multi-leader replication#

With a single leader, every write travels to one region: a Tokyo user writing to a Virginia leader pays about 200 ms. Multi-leader lets each region accept writes locally and replicate to the others asynchronously.

The catch is write conflicts. Two users edit the same document title in different regions at the same time, both succeed locally, and then replication delivers two different answers. Ways to resolve them:

Rule of thumb: if a single leader works, use it. Reach for multi-leader only when you need low-latency writes in several regions, must accept writes while a region is down, or support offline clients that sync later.

Key takeaways#

• Make writes cheaper first: async side effects, fewer indexes, batching, write-optimized storage.
• Queues absorb spikes. Pair at-least-once delivery with idempotent consumers.
• Shard last. Pick a high-cardinality key that matches your main access pattern and keeps related data together.
• Hash partitioning spreads load but breaks range scans; hash + sort key gives you both within a partition.
• Consistent hashing moves only about $1/N$ of keys when shards change. Virtual nodes keep the load balanced.
• Multi-leader buys regional write latency at the cost of conflict resolution.