Courses › System Design › Question Breakdowns

Design Distributed Counters

Hot-key contention, sharded counters, log-based batching, HyperLogLog and CRDT counters.

Design a service that counts things at enormous scale — likes on posts, views on videos, reactions, API calls — where a single viral item can receive hundreds of thousands of increments per second. It's a short question with a deep core: write contention on hot keys, sharded counters, batching, and the trade-off between exact and approximate counts.

Requirements

Functional

  • increment(counterId, delta) — e.g. a like (+1) or unlike (−1)
  • get(counterId) — read the current count; and batch get for a page of items
  • (Variant) Unique counts — distinct viewers, not total views

Out of scope: who liked what (that's a separate edge store), analytics breakdowns by time and dimension (see Design an Ad Click Aggregator).

Non-functional

  • Scale: ~1M increments/second overall; a single hot counter may get 100k+/second
  • Low-latency reads: counts render on every feed item — < 10ms, ~10M reads/second
  • Eventually accurate: counts may lag by a few seconds, but must converge to the exact value and never lose increments
  • Highly available for increments; a failed write shouldn't be visible to the user

[!TIP] The problem is one row A single database row can take maybe a few thousand locked updates per second. A viral post asks for 100k. Everything in this design — sharding the counter, batching, buffering — is a way of turning many writes to one key into few writes to many keys.

Capacity estimates

Core Entities

  • Counter — ID (e.g. post:123:likes), value
  • Counter shard — counter ID, shard number, partial value
  • Increment event — counter ID, delta, idempotency key, timestamp

API

POST /v1/counters/{counterId}/increment   { "delta": 1, "idempotencyKey": "like:u42:p123" }  → 202
GET  /v1/counters/{counterId}             → { "value": 1840213 }
POST /v1/counters:batchGet  { "ids": ["post:1:likes", "post:2:likes", ...] }

Note the increment returns 202 — accepted for counting, not necessarily applied yet.

High-Level Design

app = client "App Servers" @0,1
inc = service "Counter Service" @1,1
q = queue "Increment Log\n(Kafka, by counter)" @2,0
agg = service "Aggregators\n(batch per counter)" @3,0
db = store "Counter Store\n(sharded rows)" @3,1
cache = cache "Read Cache" @2,2
app -> inc : increment / get
inc -> q : append
q -> agg
agg -> db : one write per counter per second
inc -> cache : get
cache -> db : on miss / refresh
  • Increments are appended to a durable log (Kafka partitioned by counter ID) and acknowledged.
  • Aggregators consume the log, sum deltas per counter in memory for ~1s, and write one update per counter per interval.
  • Reads come from a cache refreshed from the counter store, so read traffic never touches the write path.

Deep Dives

1. Approaches compared

Real systems combine them: buffering for throughput, sharding for the hottest keys, Redis for fast reads.

2. Sharded counters

Split a counter into N sub-counters: post:123:likes#0 … #N-1. Each increment updates a random shard; a read sums all shards.

  • Choose N per counter: 1 for normal items, 64+ for viral ones. Promote a counter to more shards when its write rate crosses a threshold.
  • Reads get more expensive with N — cache the sum.

3. Buffering and batching

The aggregator turns 100k +1s into one +100000 every second. To make it exactly once:

  • Kafka partitions by counter ID, so one aggregator owns each counter.
  • The aggregator writes the batch and the Kafka offset it covers in the same transaction (or makes the write idempotent by recording the last applied offset per partition in the counter store). After a crash it resumes from the stored offset without double-applying.

[!WARNING] "Fast" in-memory counting can lose increments Keeping counts only in app-server memory and flushing periodically loses everything since the last flush when a server dies. Put increments in a durable log first; memory is only for aggregating what's already safe.

4. Idempotency and unlikes

Likes are toggles: the same user liking twice shouldn't count twice. The source of truth for "who liked" is an edge store (user, post unique). Only a state change in that store emits a +1 or −1 to the counter. The counter itself is then a derived, eventually consistent aggregate — and can be recomputed from the edge store if it ever drifts.

5. Reading counts at 10M/s

  • Cache counts in Redis/Memcached with a short TTL (a few seconds) or update the cache from the aggregator after each batch.
  • Batch-get for a feed page (MGET of 20 counters).
  • Display rounding ("1.8M likes") means hot counters don't need second-by-second precision on screen.

6. Approximate and unique counts

  • Unique viewers with exact sets is memory-hungry (a set of user IDs per video). HyperLogLog estimates distinct counts within ~1% using ~12 KB per counter, and HLLs from different shards merge losslessly.
  • Sampling: for very hot counters, count 1 in 10 increments and multiply — only where approximate is acceptable (views, not money).

7. Multi-region

Let each region count locally and replicate per-region partial counts (a G-counter / PN-counter CRDT: the global value is the sum of each region's value). Regions never conflict because each only writes its own slot.

What's expected at each level

  • Mid-level: a counters table, caching reads, and recognizing that one hot row is a bottleneck.
  • Senior: sharded counters and write buffering through a log, exactly-once aggregation via offsets, likes as a derived count from an edge store.
  • Staff: adaptive sharding for hot keys, HyperLogLog for unique counts, CRDT counters across regions, reconciliation from the source of truth.