Engineering Atlas › System Design Interviews

Counting accurately across many servers

13 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

How it works

When the pattern shows up in interviews

Like and view counts, follower counts, unique visitors, ad clicks for billing, trending hashtags and top-k, and rate limiter counters. Interviewers usually ask "how do you show the like count?" and then add a celebrity post with millions of likes in an hour.

Classify before designing

Likes are deduplicated per user, must reflect the user's own action immediately, and can lag for others. Views can be approximate and minutes stale. Unique visitors are distinct and approximate. Ad clicks must be exact, deduplicated and auditable, but can finalise hours later. Saying this out loud is half of a strong answer.

The hot row

UPDATE posts SET like_count = like_count + 1 WHERE id = $1 is correct but serialises on the row lock. As with any hot row, throughput is bounded by how long each transaction holds the lock, a few hundred to a few thousand per second on one row depending on commit latency. A viral post exceeds that easily.

Sharded counters

Split the counter into N rows: (post_id, shard) with a random shard per increment. Writes spread over N locks; reads sum N rows, which is cheap when cached. Choose N from the peak write rate per counter divided by what one row sustains, and grow it only for hot keys, so ordinary posts keep a single row. Google's Firestore documents the pattern with a deliberately low per-document ceiling: its sample assumes each shard takes about one increment per second, so write throughput grows linearly with the number of shards (Firestore docs).

Write-behind aggregation

Instead of writing every increment, each app server keeps an in-memory map post_id -> delta and flushes it every few seconds as one INCRBY or one batched SQL statement. A million increments per second on a single viral post become one write per server per flush interval. The cost is that a crash loses the unflushed deltas, acceptable for views and not for billing. A durable variant writes raw events to Kafka and lets stream processors aggregate per key per window; the event log stays the source of truth, so totals can be recomputed if a bug is found. Netflix's counter service offers both as named modes: a best-effort counter on its EVCache layer, and an eventually consistent one that stores every increment as an immutable event with an idempotency key and rolls events up in time windows in the background; it served about 75,000 count requests per second at single-digit millisecond latency when described in 2024 (InfoQ).

Deduplication and the source of truth

For likes, the real data is the set of (user_id, post_id) pairs with a unique constraint; the count is a derived, cached aggregate that can be rebuilt. For billing clicks, each event carries an ID and the aggregator deduplicates within a window before counting.

Approximate counting

Exact distinct counts need memory proportional to the number of distinct items. HyperLogLog hashes each item, uses some bits to pick a register, and records the longest run of leading zeros seen in the rest; rare long runs imply many distinct items. Redis's implementation uses 16,384 six-bit registers (12 KB) with a standard error of about 0.81 percent (Redis docs), and two HLLs merge by taking the per-register maximum, so daily sketches combine into weekly ones. Query engines expose the same idea: Presto's APPROX_DISTINCT has about 2.3 percent standard error by default, and at Facebook a weekly distinct-visitor count that would take days and terabytes of memory ran in 12 hours in under 1 MB (Meta Engineering). For heavy hitters, a count-min sketch keeps a few rows of counters indexed by different hashes; the estimate is the minimum across rows, which never undercounts (Cormode and Muthukrishnan). Redis Open Source includes count-min sketch (CMS.* commands) and Top-K types; its Top-K uses the HeavyKeeper algorithm, a decaying counter table plus a min-heap of the K largest (Redis docs).

Counting across regions

If each region accepts increments locally, a single global row would need cross-region coordination on every write. A G-counter gives each replica its own slot that only it increments; the value is the sum, and merge takes the element-wise maximum, which is commutative, associative and idempotent, so replicas converge no matter how often or in what order they gossip. A PN-counter pairs two G-counters for increments and decrements.

Down the stack

Layer 1The contract

State the guarantee per counter. Likes: the user's own like is reflected immediately, the total converges within seconds, and the count never double-counts one user. Views: monotonic non-decreasing totals, within a few percent, lagging by under a minute. Unique visitors: an estimate with stated relative error. Billing: exactly-once per event ID, final within hours, reconcilable against raw logs. CRDT counters: every replica that has received the same set of updates reports the same value, regardless of delivery order or duplication.

Layer 2The mechanism

Redis executes commands on a single thread, so INCRBY is atomic without locks and one instance handles on the order of 100,000 simple operations per second. Its HyperLogLog (PFADD, PFCOUNT, PFMERGE) uses a sparse encoding for small cardinalities and switches to the dense 12 KB form as it grows; PFCOUNT caches the last estimate in the header. In Postgres, each counter update writes a new tuple version and a WAL record (a HOT update if no indexed column changes), leaving dead versions for vacuum. A stream processor such as Flink keeps per-key window state in RocksDB and checkpoints it with Kafka offsets so that recovery replays from a consistent point: each snapshot records a position in every source partition together with every operator's state (Flink docs). A PN-counter is two maps P[replica] and N[replica]; value is sum(P) - sum(N); merge sets each entry to max(local, remote).

Layer 3The machine

Contention appears at every level. On one machine, many cores doing atomic increments on the same 64-byte cache line force that line to bounce between cores, each transfer costing on the order of 100 nanoseconds, so a shared atomic counter scales negatively with cores. Linux's percpu_counter and Java's LongAdder solve it the same way as sharded rows: per-core slots summed on read, trading memory for much higher throughput under contention (Java docs). In a database, the bounce is a row lock plus a WAL flush; group commit amortises the fsync, but lock hold time still serialises one row. Over the network, one round trip per increment is the dominant cost, so batching 1,000 increments into one request cuts load by roughly a thousandfold. A 12 KB HLL fits in a few hundred cache lines, so PFADD is a hash plus one register update, well under a microsecond.

Why it's built this way

  1. One memory location, row or key can only be updated serially, so hot counters are split into slots that are updated independently and summed on read.
  2. Per-increment I/O is expensive, so increments are aggregated in memory and written as deltas, trading durability of the last few seconds for orders of magnitude less load.
  3. Exact distinct counting costs memory proportional to cardinality, so sketches trade a small, bounded error for constant memory.
  4. Regions cannot coordinate on every write without paying cross-region latency, so each region owns its own slot and replicas merge with an idempotent max.
  5. Derived counts drift after bugs, so a durable event log or a fact table remains the source of truth for recounts.

Diagram

flowchart LR
    C["Clients: view events"] --> A1["API server 1: in-memory deltas"]
    C --> A2["API server 2: in-memory deltas"]
    A1 -->|"flush every 5 s"| K["Kafka: view events and deltas"]
    A2 -->|"flush every 5 s"| K
    K --> SP["Stream aggregator (per video, per minute)"]
    SP --> R[("Counter store: sharded rows")]
    SP --> H[("HLL per video per day")]
    R --> CA["Cache: rounded counts"]
    CA --> RD["Watch page reads"]

Worked example

A video service records about 2 billion views per day: around 23,000 per second on average and perhaps 100,000 at peak. A newly viral video takes 30,000 views per second by itself.

Instead of one write per view, each of 400 API servers aggregates views in memory and flushes every 5 seconds. For the viral video that is 400 delta writes per 5 seconds, 80 per second, instead of 30,000. Across all videos, each flush carries the distinct videos viewed on that server in 5 seconds, so total counter writes fall to a few thousand per second.

Deltas go to Kafka keyed by video ID, and a stream job sums them per minute into the counter store, recording processed offsets with the totals so a restart does not double-apply a minute. The watch page displays rounded values ("1.2M views"), so the cache can hold a count for 30 seconds without users noticing.

For unique viewers, the team keeps daily HLLs only for videos with over 1,000 views that day; at 12 KB each, a million active videos need about 12 GB. Monetised views are counted separately from the raw event log, deduplicated by view ID, filtered for fraud and published hours later; displayed and paid counts may differ.

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
Single row or key incrementExact, simple, immediateHot-key ceilingLow-traffic counters
Sharded counterSpreads writes, still exactReads sum N shards, more rowsHot keys needing exact totals
Write-behind batchingHuge write reductionLoses unflushed deltas on crash, lagViews, impressions, analytics
Stream aggregation from a logDurable, recomputable, windowedPipeline to operate, minutes of lagBilling, analytics, large scale
HyperLogLog / count-min sketchConstant memory, mergeableApproximateDistinct counts, top-k
CRDT counterMulti-region writes without coordinationMemory per replica, no exact global snapshotActive-active counters

Pitfalls in production

Leadership lens

Check yourself

  1. Why does sharding a counter into 16 rows help writes but cost reads?

    Show answer

    Increments spread across 16 independent locks, but every read must fetch and sum all 16 rows, so reads are usually cached.

  2. Why is merging G-counters with element-wise max safe under duplicate or reordered gossip?

    Show answer

    Each slot only grows and only its owner increments it, so max always picks the newest value for that slot, and applying the same state twice changes nothing.

  3. How does write-behind batching reduce load for a viral item, and what does it risk?

    Show answer

    Each server sends one aggregated delta per interval instead of one write per event, but a crash loses the deltas accumulated since the last flush.

  4. When would you prefer an exact set of user IDs over a HyperLogLog?

    Show answer

    When counts are small or exactness matters, such as likes where you must also answer "has this user already liked it?".

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →