Write-heavy

Direct sharding is the most intuitive write-heavy move: split the keyspace across N database primaries, route each write to the correct shard, and let each shard handle its fraction of the total load. If one Postgres primary tops out at 12K writes/sec, four shards give you roughly 48K writes/sec with linear scaling — in theory.

The reality is messier. The shard key determines everything. A good key has high cardinality (millions of distinct values), even distribution (no single value dominates), and query alignment (the most common queries filter on the shard key, avoiding cross-shard scatter). user_id is the classic good key. country_code is a classic bad key — the US shard absorbs 40% of traffic while 150 other shards idle. timestamp is even worse for write-heavy: all current writes funnel into the single "now" partition.

When you detect skew, the fix depends on the shape. For moderate skew, composite keys work: shard on user_id + date to spread a single user's writes across days. For extreme skew (a celebrity post getting millions of likes), you need sharded counters — split the counter into N sub-keys (post:1:likes:0 through post:1:likes:15), route each increment to a random sub-key, and SUM on read. The write contention drops by N; the read cost increases by N lookups, which is almost always an acceptable trade because reads on a write-heavy path are infrequent.

Cross-shard queries are the operational tax. Any query that spans shards requires scatter-gather, which is slower and harder to paginate. The discipline is to design the schema so that 95%+ of queries are shard-local. For the remaining 5%, build a separate read model (a search index, a denormalised aggregate table) that is updated asynchronously.

Resharding — growing from N to M shards — is the hardest operational moment in a sharded system. The proven approach is dual-write: configure the application to write to both old and new shard maps, run a backfill job to copy historical data to the new map, verify consistency, then cut reads over. This is an online migration; the system stays live throughout, but it requires careful coordination and a verification step that compares rows across old and new shards.

The log-first variant decouples write acceptance from write processing. Producers append events to Kafka (or Kinesis, Pulsar, Redpanda) — a sequential, partitioned, durable log. Consumers read events at their own pace and materialise whatever views the query layer needs. The log is the source of truth; views are disposable projections.

The storage engine behind most write-optimised stores is the LSM tree (Log-Structured Merge tree). Cassandra, ScyllaDB, RocksDB, LevelDB, and HBase all use variants of it. The idea is elegant: writes go to an in-memory buffer (the memtable) and a sequential WAL for durability. When the memtable fills, it flushes to an immutable sorted file on disk (an SSTable). Background compaction merges SSTables to reclaim space and keep read amplification bounded.

The LSM trade-off is explicit: writes are fast (sequential append, no random I/O), reads are slower (may need to check multiple SSTables + bloom filters), and compaction consumes background CPU and I/O. For write-heavy workloads where reads are aggregate or infrequent, this trade-off is ideal. For workloads with heavy point-read traffic on recent data, you may need to pair the LSM store with a cache or a B-tree index for the hot slice.

The operational model is different from RDBMS. Schema changes are cheap (add a column to a wide-column store without locking), but data modelling is query-driven: you design tables around your access patterns, not around entity relationships. Denormalisation is the norm, not the exception. If you need a new query shape, you add a new consumer that reads the log and builds a new table — no migration, no downtime, just replay.

The failure modes are consumer-centric. Consumer lag means the read model is stale; monitor lag as a primary SLO metric and autoscale consumers on it. Poison messages block a partition; use a dead-letter queue after N retries. Hot partitions in Kafka mirror the hot-shard problem in databases: pick a partition key with high cardinality, and if no natural key exists, add a random suffix or time-slice component.

This variant shines for event-sourced systems, analytics ingestion, CDC pipelines, and any workload where the write shape (append) differs from the read shape (aggregate, time-range, or entity-centric lookup). It is the backbone of modern data platforms at companies like LinkedIn, Uber, and Netflix.

Batching is the oldest trick in the write-heavy playbook and remains one of the most effective. The insight is that per-write overhead — network round-trip, WAL fsync, index update, replication — dominates at low batch sizes. A single Postgres INSERT takes ~1ms including commit. A batch of 100 INSERTs in a single transaction takes ~5ms total — a 20x throughput improvement from trivially grouping writes.

Batching can happen at three tiers, each with different latency-throughput trade-offs:

Tier 1 — Client-side buffering. The Kafka producer's linger.ms setting is the canonical example: instead of sending each record immediately, the producer waits up to N milliseconds to accumulate a batch, then sends them all in one request. This reduces network round-trips and enables compression. The trade-off is added latency equal to the linger window. For analytics events, 100ms of linger is invisible; for payment confirmations, it is unacceptable.

Tier 2 — Application-level micro-batching. The application server accumulates writes in a local buffer and flushes every 50ms or every 200 records, whichever comes first. This is common in metrics pipelines: instead of one INSERT per metric point, the app builds a batch and does a single multi-row INSERT or COPY statement. The complexity is managing the buffer lifecycle — flush on shutdown, flush on timeout, handle partial failures.

Tier 3 — Database group commit. Postgres commit_delay and commit_siblings tell the WAL writer to wait briefly for additional commits before fsyncing, amortising the fsync cost across multiple transactions. MySQL's innodb_flush_log_at_trx_commit = 2 is similar. This is invisible to the application but requires trusting that a brief power loss window (the commit delay) is acceptable for the durability guarantees you need.

The three tiers are composable. A well-tuned write-heavy pipeline uses client-side linger (Kafka producer), application-level batching (Flink micro-batches), and database group commit (Cassandra's commitlog sync period) simultaneously. Each tier removes a multiplier from the per-write cost, and the cumulative effect is dramatic: 100x throughput improvement is routine.

The anti-pattern is unbounded batching. If the batch grows without a size or time cap, a slow consumer accumulates an enormous batch that overwhelms the database on flush. Always cap batches by both count and wall-clock time, and monitor batch sizes as a key operational metric.

When the write volume is so large that even Kafka partitions would be overwhelmed by per-event processing, the answer is to reduce the event count before it reaches the durable layer. Hierarchical aggregation builds a tree of stateless (or soft-state) aggregators that merge partial results at each level, reducing cardinality by 10–100x per tier.

The classic example is a live viewer count for a streaming platform. During a major event, millions of viewers send heartbeat pings every 10 seconds. Writing each ping individually to a database would require millions of writes per second — far beyond what any single store can handle. Instead:

• Edge aggregators (one per PoP or data center) count heartbeats locally over a 5-second window. 100 edge nodes each seeing 10K pings/sec produce 100 partial counts every 5 seconds — a 500x reduction.
• Mid-tier aggregators (one per region) merge the edge partials. 10 mid-tier nodes produce 10 regional counts — another 10x reduction.
• The root aggregator sums the regional counts and writes a single global number to the database. The write rate to durable storage: one write every 5 seconds, regardless of the viewer count.

The properties that make aggregation composable are associativity and commutativity: SUM, COUNT, MIN, MAX, HyperLogLog unions. If your metric has these properties, you can aggregate at any level in any order and get the correct answer. Metrics that require exact ordering (median, percentile) need more sophisticated sketches (t-digest, DDSketch) that are still mergeable but approximate.

The trade-off is temporal resolution. Each aggregation window adds latency: if the edge window is 5 seconds and the mid-tier window is 10 seconds, the global count is up to 15 seconds stale. For a live viewer counter this is fine — users cannot perceive the difference between 1,234,567 and 1,234,892. For a real-time bidding system, it is not. Match the window size to the staleness budget.

Failure handling in a tree is graceful: if one edge node dies, the global count is slightly low for one window, then recovers. Compare this to a centralised counter where the single node's death means total blindness. The tree architecture is inherently fault-tolerant because each node is stateless (or holds only one window of soft state).

Start with a single Postgres or MySQL primary on NVMe storage. Tune the WAL: set commit_delay to 10–50 microseconds to batch fsyncs, consider synchronous_commit = off for workloads where losing the last few transactions on crash is acceptable (metrics, analytics), and set sync_binlog=0 on MySQL for the same effect. Use bulk INSERTs or COPY instead of row-at-a-time inserts. Disable unused indexes on write-hot tables — every index is a write amplifier.

With these tunings, a single modern server (64 cores, NVMe, 256 GB RAM) sustains 10K–15K write TPS for typical row sizes. This is the right architecture for most startups and many mid-stage companies. The bottleneck that forces you to v2 is either CPU saturation on the primary, WAL throughput hitting the NVMe ceiling, or replication lag to followers becoming unacceptable.

When vertical tuning is exhausted, shard the database. Pick a shard key with high cardinality and even distribution — user_id for social apps, device_id for IoT, ad_id for click tracking. Hash the key modulo N to select the shard. Each shard is a full primary with its own WAL, its own replicas, and its own capacity budget.

The operational cost of sharding is real. Cross-shard queries require scatter-gather. Schema migrations must be coordinated across all shards. Resharding (changing N) requires a dual-write migration: write to both old and new shard maps, backfill historical data, verify consistency, then cut over. Plan the initial shard count for 2–3 years of growth, because resharding is the most expensive operational event in a sharded system.

Monitor per-shard write rates and latency independently. Skew — one shard absorbing disproportionate traffic — negates the scaling benefit. If you detect skew, the fix is either a better shard key or sharded counters for the hot keys.

At this stage, writes go to Kafka partitions — append-only, horizontally scalable, with configurable retention (7 days is a common default, with archival to S3 for longer). The database is no longer in the write path; it is in the consumer path, receiving pre-processed, batched, and possibly pre-aggregated data from consumer workers.

Consumer groups materialise purpose-built views: an LSM store (Cassandra, ScyllaDB) for point lookups by entity + time, an OLAP engine (Druid, Pinot, ClickHouse) for aggregate dashboards, and S3 Parquet files for offline analytics and compliance. If a view design turns out wrong, add a new consumer and replay the Kafka topic — no migration, no downtime.

The SLO metric shifts from database latency to consumer lag: how far behind the consumer group is from the log head. Alert on lag exceeding your staleness budget (e.g., 30 seconds for a metrics dashboard). Autoscale consumers based on lag, not CPU — lag is the signal that matters.

For global-scale write-heavy systems, each region gets its own Kafka cluster. Producers write to the local cluster with single-digit-millisecond latency. MirrorMaker 2 (or Confluent Cluster Linking) replicates topics across regions asynchronously. Each region's consumers read from the local cluster, so consumer processing is also region-local.

The trade-off is cross-region consistency. Events written in US-East are visible to EU consumers after the replication lag (typically 50–200ms inter-region). For most write-heavy workloads (metrics, telemetry, clickstream), this is invisible. For workloads that require causal ordering across regions (multi-player games, collaborative editing), you need additional coordination — vector clocks, hybrid logical clocks, or a global sequencer.

The disaster recovery story is strong: if one region fails, the other regions have a near-complete copy of the log. Failover means redirecting producers to a surviving region and letting consumers catch up from the replicated topic. Data loss is bounded by the replication lag at the moment of failure — typically under one second.

The storage engine is the single most consequential choice in a write-heavy system. B-trees (Postgres, MySQL InnoDB) and LSM trees (Cassandra, RocksDB, ScyllaDB) make opposite trade-offs, and picking the wrong one costs you an order of magnitude in throughput.

B-trees are read-optimised. Every write updates the tree in place: find the leaf page, modify it, write the dirty page back to disk. This means writes incur random I/O — seeking to the correct page, reading it, modifying it, writing it back. For a table with 10 indexes, a single row insert triggers 11 random writes (the heap page + 10 index pages). The upside is that reads are fast: one tree traversal finds the exact row, with no need to merge results from multiple files.

LSM trees are write-optimised. Writes go to an in-memory buffer (memtable) backed by a sequential WAL. No random I/O at write time — just an append to the WAL and an insert into a sorted in-memory structure. When the memtable fills (typically 64–256 MB), it flushes to an immutable SSTable file on disk. Background compaction merges overlapping SSTables to reclaim space and keep the number of files bounded.

The trade-offs are quantifiable:

• Write amplification: B-trees write each page at least twice (WAL + data file), plus index updates. LSM trees write each record once to the WAL + memtable, then again during each compaction level. Typical write amplification: B-tree ~10–30x, LSM ~10–30x (leveled compaction) or ~2–4x (tiered compaction). LSM with tiered compaction wins handily for pure append workloads.
• Read amplification: B-trees read one page per tree level (~3–4 for typical tables). LSM trees may need to check multiple SSTables across levels; bloom filters cut most false checks, but worst-case reads touch more I/O. For point lookups, B-trees are faster. For range scans on sorted data, LSM trees are competitive because SSTables are sorted.
• Space amplification: B-trees keep ~50% page utilisation after many updates. LSM trees have temporary space amplification during compaction (up to 2x for leveled). Neither is clearly better; it depends on the workload's update vs append ratio.

When to pick LSM: append-heavy workloads (timeseries, events, metrics), write throughput > 10K TPS, reads are mostly range scans or aggregate queries, you can tolerate background compaction consuming 10–20% of disk I/O. When to pick B-tree: mixed read-write workloads with heavy point lookups, need for strong transactional semantics (SERIALIZABLE isolation), operational team is experienced with Postgres or MySQL, write throughput < 10K TPS.

The hybrid approach is increasingly common: use Postgres (B-tree) for the transactional core (accounts, orders, inventory) and Cassandra or ScyllaDB (LSM) for the high-volume append path (events, metrics, audit logs). Connect them with CDC so the LSM store stays in sync without the application managing dual writes.

A hot write key is the write-heavy equivalent of a hot cache key. When millions of users like the same post, all increments target the same row — post:1:likes. The database serialises those updates, and throughput collapses to whatever one row lock can sustain (typically 1–5K updates/sec on Postgres, less on Cassandra due to read-before-write for counters).

The sharded counter pattern spreads the write load across N sub-keys. Instead of one key post:1:likes, create N keys: post:1:likes:0 through post:1:likes:15 (for N=16). Each increment picks a random sub-key (or hashes the user ID to a sub-key for idempotency). Write contention drops by N because updates are distributed across N independent keys.

The read path sums all sub-keys: GET post:1:likes:0 through GET post:1:likes:15, then sum. This is N reads instead of one, but for write-heavy counters, this trade-off is almost always correct — the write bottleneck is the limiting factor, and N reads are cheap when N is small (16–64).

Choosing N. Start with N = number of database shards or Redis nodes, so each sub-key naturally lands on a different node. If you are on a single node, N = 16 is a reasonable starting point — it reduces contention by 16x while keeping the read fan-out manageable. Monitor per-sub-key write rates; if one sub-key is still hot (because of hash collisions), increase N.

Idempotent increments. If you need to prevent double-counting (a user liking a post twice), the sub-key selection must be deterministic: hash(user_id) % N selects the sub-key, and the sub-key stores a set of user IDs that have already incremented. This is more expensive (read-modify-write instead of blind increment) but necessary for correctness on engagement metrics.

Periodic compaction. Over time, you can compact the sharded counter back into a single value. A background job reads all sub-keys, sums them, writes the total to a post:1:likes:total key, and resets the sub-keys. This makes reads cheaper (one key instead of N) at the cost of a brief inconsistency window during compaction. Run compaction during off-peak hours.

Twitter uses this pattern for like and retweet counts, with N=16 sub-keys per tweet. At 300K+ likes per second during peak events, a single counter row would be a permanent bottleneck. The sharded counter reduces contention to ~19K writes per sub-key per second, well within what a single Cassandra partition can handle.

Resharding — changing the number of shards from N to M — is the most operationally complex event in a sharded database's lifecycle. The naive approach (stop writes, rehash all data, restart) requires downtime proportional to data size, which is unacceptable for any production system doing thousands of writes per second.

The online approach has four phases:

Phase 1 — Prepare. Provision the new shard set (M shards). Configure the application to compute both old and new shard assignments for every write, but continue writing only to the old shards. Verify that the new shard assignment logic is correct by shadow-logging where each write would go.

Phase 2 — Dual-write. Enable dual-write: every new write goes to both the old shard (using the old hash function) and the new shard (using the new hash function). Reads still come from the old shards. This phase ensures that all new data exists in both shard sets. The write throughput cost roughly doubles, but it is temporary.

Phase 3 — Backfill. A background job scans the old shards and copies historical data to the new shards. This runs at throttled throughput to avoid impacting production writes. For a 10TB dataset at 100MB/sec throttled, the backfill takes roughly 28 hours. During this phase, dual-write continues, so any data modified during the backfill is handled by the dual-write path.

Phase 4 — Verify and cutover. A verification job compares rows between old and new shards (sample-based for large datasets, full-scan for small ones). Once verification passes, cut reads to the new shards. After a soak period (24–72 hours), disable dual-write and decommission the old shards.

The critical subtlety is ordering during dual-write. If a row is updated twice in quick succession, the old shard sees update A then B, but the new shard might see B then A due to network timing. The backfill job handles this by comparing the final state (not the update sequence) and copying the latest version. For workloads with true concurrent updates to the same row, use a last-write-wins timestamp or a conflict resolution function.

Vitess, Citus, and CockroachDB automate parts of this process, but understanding the four phases is essential for interviews because interviewers frequently ask "what happens when you outgrow your shard count?" and expect the dual-write answer.

Load shedding is the deliberate dropping of low-priority traffic to protect the system's ability to process high-priority traffic. In a write-heavy system, the write path is the bottleneck, and shedding is the last-resort defence when batching, sharding, and queuing are all saturated.

The key insight is priority classification. Not all writes are equal. In an ad-click pipeline, a billable click event is critical (it affects revenue reconciliation), while a debug-level impression event is best-effort. In a metrics pipeline, P0 alerts data is critical, while P2 dashboard data can be dropped for 60 seconds without anyone noticing. Classify your write traffic into 2–3 priority tiers before you need shedding, because doing it during an incident is too late.

Admission control strategies:

1. 1Queue depth threshold. Monitor the depth of the write queue (Kafka consumer lag, in-memory buffer size). When it exceeds a threshold, reject new writes for the lowest priority tier. Simple to implement, but the threshold requires tuning — too low and you shed unnecessarily, too high and you shed too late.

1. 1CoDel (Controlled Delay). Track the time each write spends in the queue. If recent items are consistently waiting longer than a target (e.g., 50ms), start dropping. CoDel adapts to sustained overload and ignores brief bursts, making it more robust than fixed thresholds.

1. 1Token bucket per priority. Each priority tier has its own token bucket with a configured rate and burst. High-priority writes get large buckets (effectively unlimited under normal load). Low-priority writes get smaller buckets that drain quickly under overload. This gives you fine-grained control over the relative admission rates.

1. 1Adaptive shedding with feedback. Monitor the downstream system's latency and error rate. When latency exceeds the target, reduce the admission rate for low-priority traffic proportionally. When latency recovers, gradually re-admit. This is the most sophisticated approach and is used by Google's Doorman and Uber's QALM.

What to do with shed traffic. Never silently drop. Log the dropped event count and type to a shedding counter. For recoverable shedding, write the dropped events to a cold-path store (S3, a separate low-priority Kafka topic) for later replay. For non-recoverable shedding (real-time metrics where stale data is useless), count and move on. The shedding counter is a critical operational metric — if it fires continuously, you need more capacity, not more shedding.

Batching is deceptively simple in concept — group writes and flush together — but the implementation details determine whether you get a 100x throughput improvement or a system that occasionally loses data and always has unpredictable latency.

Where to batch. The three tiers (client, application, database) are not alternatives; they are composable layers. A well-tuned pipeline uses all three:

• Client (Kafka producer linger.ms=50, batch.size=64KB): reduces network round-trips from producer to broker.
• Application (micro-batch every 100ms or 500 records): converts per-record INSERT into multi-row COPY/INSERT.
• Database (Postgres commit_delay=10us, commit_siblings=5): amortises WAL fsync across concurrent transactions.

Each tier reduces the effective cost per write by a different multiplier. Client batching saves network. App batching saves SQL parsing and index updates. DB batching saves fsync calls. Together they compound.

When to flush. Every batch needs two triggers — a time cap and a size cap — and the flush fires on whichever comes first. Time cap alone risks tiny batches under low load (wasted flushes) and enormous batches under high load (memory pressure). Size cap alone risks unbounded latency under low load (the batch never fills). Both together give predictable latency (bounded by the time cap) and predictable memory (bounded by the size cap).

Typical values for a metrics pipeline: flush every 200ms or 1000 records. For a Kafka consumer writing to Cassandra: flush every 500ms or 5000 records. For a real-time dashboard: flush every 50ms or 100 records. The values depend on the downstream system's optimal batch size and the staleness budget.

Error handling. When a batch write fails (network timeout, constraint violation, partial failure), you have three options:

1. 1Retry the entire batch. Simple but risks duplicates unless the writes are idempotent.
2. 2Binary-search the failure. Split the batch in half, retry each half, repeat. Isolates the bad record in O(log N) retries. Used when one poison record in a batch of 1000 should not block the other 999.
3. 3Dead-letter the batch. Write the entire failed batch to a DLQ for manual inspection. Used when the failure is systemic (schema mismatch, downstream outage) and retrying is pointless.

The `commit_delay` trick. Postgres's commit_delay is underappreciated. It tells the WAL writer to wait N microseconds after the first commit in a group before fsyncing, hoping more commits arrive in that window. With commit_delay=10 and commit_siblings=5 (only delay if 5+ transactions are in-flight), a heavily concurrent write workload sees 3–5x throughput improvement at the cost of 10 microseconds of added latency per commit. This is invisible to the application and requires zero code changes — just a GUC setting.

Hierarchical aggregation is the answer to the question "how does YouTube show a live viewer count when 50 million people are watching simultaneously?" The answer is: it does not count 50 million pings. It builds a tree that reduces 50 million pings to one write.

The tree structure. At the leaves, edge aggregators sit close to users (one per PoP, one per data center, or one per server). Each edge node maintains an in-memory counter and flushes a partial count upstream every W seconds (the window size). Mid-tier aggregators (one per region or availability zone) receive partial counts from their children and produce a regional total. The root aggregator sums the regional totals and writes the global count to a durable store.

Cardinality reduction. Suppose 100 edge nodes each receive 10,000 heartbeats per second. Without aggregation, the durable store sees 1,000,000 writes/sec. With a 5-second edge window, each edge node flushes one partial count per window, producing 100 writes per 5 seconds = 20 writes/sec to the mid tier. With 10 mid-tier nodes and a 10-second window, the root sees 10 writes per 10 seconds = 1 write/sec. The reduction factor is (event_rate × window) / nodes_per_tier, compounded across tiers.

Mergeable data structures. The tree works because the aggregation operation is associative and commutative. COUNT and SUM are trivially mergeable. For approximate distinct counts, HyperLogLog sketches are mergeable — each node maintains a local HLL, and the parent merges child HLLs into a combined estimate. For percentiles, t-digest or DDSketch sketches are mergeable with bounded error. For exact distinct counts, you would need to pass the full set upstream, which defeats the purpose — use an approximate structure instead.

Window alignment. Edge windows do not need to be synchronized. If edge A flushes at t=5s and edge B flushes at t=7s, the mid-tier node simply accumulates whatever partials it receives and flushes on its own window. The global count is a moving average with a staleness bound of (sum of all window sizes). For a three-tier tree with 5s/10s/10s windows, the global count lags reality by at most 25 seconds. This is acceptable for viewer counts, ad impression dashboards, and most operational metrics.

Failure handling. If an edge node dies, its partial count is simply missing from the next mid-tier flush. The global count dips slightly for one window and then recovers when the edge restarts (or its traffic is redistributed to other edges). No data loss in the durable store, no cascade failure, no manual intervention. Compare this to a centralised counter: if the single counter node dies, you have zero visibility until it recovers. The tree architecture trades exact accuracy for resilience, which is the correct trade-off for all but financial accounting.

Implementation pattern. The simplest implementation uses Kafka topics as the communication layer between tiers. Edge nodes produce to a counts-edge topic (partitioned by region). Mid-tier Flink jobs consume, window-aggregate, and produce to a counts-regional topic. The root Flink job consumes regional counts and writes the global aggregate. Each tier is independently scalable, and Kafka handles backpressure and durability between tiers.