Muhammad Ahsan Farooq

Posted on Oct 9

How Databases Scale to Millions of Reads and Writes per Second

#database #systemdesign #backend #scalability

Replicas, caches, sharding, LSM-trees, and the consistency trade-offs that come with each

Every database hits a ceiling. One server can take only so many CPU cores and so much RAM before the next upgrade costs far more than it delivers. To serve millions of operations per second, you stop making one machine faster and start spreading the work across many.

Two terms get mixed up here. Performance is how fast the system answers a single request. Scalability is whether it keeps answering that fast as load grows. A database can be quick at 100 requests per second and fall over at 100,000. This post is about the second problem.

The core idea is to split the read path from the write path, because they fail in different ways and need different tools. Then choose storage engines, caches and consistency models that fit each path, and accept the trade-offs between consistency, availability and latency.

read path and write path · 6 components, 2 lanes

Reads are served from copies: an edge cache, a distributed cache and replicas. Writes are buffered by a queue and split across shards.

In this post:

Part 1: scaling reads with replicas, caches, indexes and denormalization
Part 2: scaling writes with sharding, LSM-trees, queues and connection pools
Part 3: consistency models, CAP, PACELC and distributed SQL
A one-page cheat sheet you can keep

Part 1: Scaling reads

In most web applications, reads outnumber writes by 100:1 or even 1,000:1. That imbalance is good news: reads are easier to scale, because the same data can be copied and served from many places.

1. Primary-replica replication

Send every write to one primary node. The primary streams those changes to one or more read-only replicas, and a load balancer spreads read traffic across the replica pool. Add replicas and read capacity grows almost linearly.

primary-replica replication · 1 primary, 3 replicas

Replicas can lag the primary slightly. The dashed path sends a user who just wrote back to the primary, so they always see their own change.

The catch is replication lag. Replication is usually asynchronous, so the primary does not wait for replicas before acknowledging a write. That keeps write latency low, but a replica can briefly serve stale data.

For cases where staleness is visible to users, such as seeing your own post right after publishing it, use read-your-own-writes. Route that user's reads to the primary (or a replica known to be caught up) for a short window, while everyone else reads from the asynchronous replicas.

2. Multi-layer distributed caching

Sending millions of concurrent reads to a disk-backed database is unsustainable. Put a distributed in-memory cache such as Redis or Memcached in front of it. Large platforms report hit rates above 95-99%. Netflix's EVCache is a well-known example. At those rates the database sees only a small fraction of the real traffic.

To see why this matters, take 1,000,000 reads per second. At a 90% hit rate, 100,000 reads reach the database. At 95%, 50,000 do. At 99%, only 10,000 do.

Two patterns cover most cases:

Cache-aside (lazy loading): the application checks the cache first. On a miss, it reads from the database, fills the cache and returns the result.
Write-through: every update is written synchronously to both the cache and the database, so later reads get fresh data. The price is slightly higher write latency.

cache-aside read · 1 decision, 2 paths

The three common strategies behave differently on writes. Write-behind is covered in Part 2.

Strategy	How it works	Write latency	Main risk	Best for
Cache-aside	The app reads the cache and loads from the database on a miss	Unchanged	Stale entries until the TTL expires or you invalidate them	Read-heavy data that tolerates brief staleness
Write-through	Every write goes to the cache and the database together	Slightly higher	Extra write cost for data that is rarely read	Data that must be fresh on the next read
Write-behind	Writes hit the cache, which flushes to the database later	Lowest	Unflushed writes are lost if the cache node fails	Bursty writes where small loss risk is acceptable

3. Preventing cache stampedes

When a popular key expires, thousands of requests can miss the cache in the same millisecond. If nothing stops them, they all hit the database with the same query. This is a cache stampede, also called the thundering herd, and it can turn one expired key into an outage.

Two defenses work well together:

Request coalescing: the first request for a missing key goes to the database. Duplicate requests wait for that single result and are served from the new cache entry.
Probabilistic early expiration: each request has a small, growing chance of refreshing a key in the background as it nears its TTL, so the key is usually refreshed before it ever expires.

request coalescing · before and after

The database sees one query instead of thousands, and the waiting requests are served from the fresh entry.

Here is a minimal cache-aside read with in-process coalescing in TypeScript:

const inflight = new Map<string, Promise<User>>();

async function getUser(id: string): Promise<User> {
  const cached = await redis.get(`user:${id}`);
  if (cached) return JSON.parse(cached);

  // Coalesce: one database query per key, everyone else awaits it
  if (!inflight.has(id)) {
    const load = (async () => {
      const user = await db.users.findById(id);
      await redis.set(`user:${id}`, JSON.stringify(user), { EX: 300 });
      return user;
    })().finally(() => inflight.delete(id));
    inflight.set(id, load);
  }
  return inflight.get(id)!;
}

This coalesces per process only. Across a fleet of servers, you still get one query per server, so use a distributed lock or a caching proxy for global coalescing. Adding random jitter to TTLs also stops many keys from expiring together.

4. Indexing and denormalization

Indexes: a secondary index (usually a B-tree or B+ tree) turns a full-table scan, O(N), into a logarithmic lookup, O(log N). It is the cheapest read optimization you have.
Denormalization: multi-table JOINs get expensive at scale in both CPU and disk I/O. Duplicating the fields you need into the target record, or into a materialized view, replaces the join with a fast, flat lookup. You pay for it with extra storage and more work to keep copies in sync.

In short: copy the data (replicas), keep hot data in memory (caches), and make each lookup cheaper (indexes and denormalization).

Part 2: Scaling writes

Write-heavy systems such as real-time messaging, IoT telemetry and clickstream logging are harder. Read replicas do not help, because every write still has to land on a primary storage engine. You need to spread the writes themselves.

1. Sharding (horizontal partitioning)

Sharding splits one large dataset across several independent database nodes. Each node, or shard, holds a distinct slice of the data. A shard key, such as user_id or a hash of a composite key, decides which shard stores a given row.

shard key choice · balanced vs hot shard

With range partitioning, the newest IDs always land on the last shard. A hashed key spreads the same writes across every shard.

The key you pick matters more than almost anything else:

Avoid hot keys. With range-based partitioning, sequentially increasing IDs send all new writes to the last shard. The rest of the cluster sits idle.
Prefer high-cardinality, evenly distributed keys. UUIDs or hash-based partitioning spread writes across every node.
Know the trade-off. Hash sharding balances load but makes range scans and cross-shard queries more expensive.

2. Write-optimized storage engines: LSM-trees vs B-trees

Traditional relational databases use B-trees, which update pages in place. Under heavy write volume, those random disk writes become the bottleneck.

Log-structured merge-trees (LSM-trees) avoid this. Wide-column and key-value stores such as Apache Cassandra, ScyllaDB and RocksDB use them. A write goes to an in-memory buffer (the MemTable) and is flushed to disk as an immutable, sorted file (an SSTable). Because the disk work is sequential, these engines sustain millions of writes per second at very low write latency.

LSM-tree write path · memory to disk to compaction

The trade-off shows up on reads. A lookup may check the MemTable and several SSTables, and background compaction has to merge files over time. LSM-trees buy write throughput with read and compaction cost.

	B-tree	LSM-tree
Write pattern	Random, in-place page updates	Sequential appends
Write throughput	Limited by random disk I/O	Very high
Read cost	One tree lookup	May check the MemTable and several SSTables
Background work	Page splits and rebalancing	Compaction
Typical engines	Traditional relational databases	Cassandra, ScyllaDB, RocksDB

3. Asynchronous buffering and event streaming

To absorb bursts, put a durable queue such as Apache Kafka or RabbitMQ between your application and the database. The API accepts the event and returns quickly. Ingestion workers then consume from the queue, group individual writes into batches and bulk-insert them.

asynchronous write buffering · 4 stages

This load leveling protects the database from sudden spikes and prevents thread starvation in the application tier. The cost is that writes become eventually visible, which your product has to be able to tolerate.

A related pattern is write-behind (write-back) caching. The application writes to the cache, and the cache flushes to the database asynchronously. It is very fast, but data not yet flushed can be lost if the cache node fails, so use it only where that risk is acceptable.

4. Connection pooling

Opening a new database connection for every API request costs real CPU and memory. Unclosed connections are worse. They leak, exhaust the pool and eventually starve threads.

Keep a pool of persistent, reusable connections, or put a database proxy in front of the database. This removes most handshake overhead and keeps throughput stable under high concurrency.

In short: split the writes across nodes, use an engine built for appends, and buffer bursts behind a queue.

Part 3: Consistency, CAP and distributed SQL

Once data spreads across nodes and regions, network partitions stop being hypothetical. They happen. The CAP theorem says that during a partition you must choose between consistency and availability. PACELC extends it: even with no partition, you still trade latency against consistency, because coordinating replicas takes time.

Read the decision tree below from the top. First ask whether a partition is happening, then follow the trade-off that applies.

CAP and PACELC decision tree · 1 question, 2 branches, 4 outcomes

Cassandra and DynamoDB sit on the availability and low-latency side. Spanner and Aurora DSQL pay in latency to keep consistency.

AP vs CP systems

AP (availability + partition tolerance): Apache Cassandra and Amazon DynamoDB, in their default setups, keep accepting writes during a partition and synchronize in the background. The result is eventual consistency. Replicas converge, but readers may briefly see different values.
CP (consistency + partition tolerance): MongoDB and HBase favor consistency. If a node cannot reach a quorum, it refuses reads or writes instead of risking stale or conflicting data. You give up availability for the duration of the partition.

Many of these systems are tunable, so treat the labels as defaults, not absolutes. A practical rule: choose AP when a stale read is tolerable (feeds, carts, telemetry), and CP when a wrong answer costs money (payments, inventory, ledgers).

Distributed SQL

Distributed SQL databases such as Google Spanner and Amazon Aurora DSQL aim for both worlds: SQL, ACID transactions and strong guarantees, with multi-region, active-active availability and no hand-written sharding in your application.

Optimistic concurrency control (OCC): Aurora DSQL runs a read-write transaction locally without taking locks up front. At commit time, an adjudicator service checks whether a conflicting transaction wrote to the same keys during that window. If one did, a transaction is rejected and must be retried. OCC works best when conflicts are rare, and your application code must be ready to retry.
Synchronized clocks: Spanner (TrueTime) and Aurora DSQL (Amazon Time Sync) rely on precisely synchronized clocks to give transactions a consistent global order. That lets reads be served from the local region with strong snapshot semantics, without a cross-region round trip for every read.

This is not free. Commits that span regions still pay a latency cost for coordination, which is exactly the latency-versus-consistency trade-off PACELC describes.

In short: pick consistency or availability per feature, and know what each database does when the network breaks.

Cheat sheet

Reads scale by copying data to more places. Writes scale by splitting data across more nodes.

Layer	Read-heavy workloads	Write-heavy workloads
Data partitioning	Read replicas (primary-replica)	Horizontal sharding with evenly distributed hash keys
Caching	Cache-aside or write-through (Redis, Memcached)	Write-behind (asynchronous cache-to-database flush)
Storage engine	B+ tree indexes and denormalized views	LSM-trees
Ingestion	CDN or edge caching, request coalescing	Message queues (Kafka) and batching
Consistency	Eventual consistency, read-your-own-writes	Distributed SQL with OCC, or CRDTs for conflict-free multi-region writes

Conclusion

Scaling to millions of operations per second is rarely one big trick. It is a series of small decisions, made in a sensible order:

Index and denormalize before you buy more hardware.
Add replicas and a cache to absorb reads, and guard against stampedes.
When writes outgrow one node, shard on an evenly distributed key and consider a write-optimized engine.
Buffer bursts with a queue and pool your connections.
Decide, feature by feature, how much staleness you can accept, and choose AP or CP behavior to match.

Every step buys capacity and charges you in consistency, complexity or operational cost. Measure before you add the next layer.

Two topics worth exploring next: write-behind caching with Redis Streams, and conflict-free replicated data types (CRDTs) for resolving conflicts in multi-region AP databases.