DISTRIBUTED SYSTEMS / SYSTEM CONCEPT BRIEF

Distributed counters

A distributed counter counts events (likes, views, inventory, rate-limit hits) across many servers.

IntermediatePhase 05 / Topic 8 of 17RequirementsTrade-offsFailure modes
01

Overview

A distributed counter counts events (likes, views, inventory, rate-limit hits) across many servers. A single row incremented by everyone becomes a hot spot: every increment competes for the same lock or key, capping throughput. Distributed counters solve this by splitting the count into shards, buffering increments, or using data structures that merge without coordination.

The design depends on accuracy needs. Exact counts for money or inventory need atomic operations and careful sharding. Approximate counts for views and likes can use buffered batches, eventual consistency, or probabilistic structures like HyperLogLog for unique counts.

Counting votes at many polling stations

Instead of every voter walking to one central box, each station counts its own votes and reports the total periodically. The national count is the sum of station counts.

02

When to use it

  • Likes, views, and reactions on popular content.
  • Rate limiting across a fleet.
  • Inventory or quota tracking.
  • Unique visitor counts at massive scale.
03

Where it shows up in interviews

Hot counters

Recognize it when: a viral post gets millions of likes.

  • Design a like button for Facebook
  • Design YouTube view counts
Global limits

Recognize it when: enforce a quota across many servers.

  • Design a distributed rate limiter
  • Design an API quota system
04

Where it is used in real software

YouTube view counts

Views are aggregated in batches and validated, which is why counts sometimes lag.

Redis INCR and HyperLogLog

Atomic counters and approximate unique counts (0.81% error in 12 KB).

CRDT counters

Riak and Redis Enterprise offer conflict-free counters that merge across regions.

05

Key terms

Hot key
A single key receiving a large share of traffic.
Sharded counter
Count split into K sub-counters summed on read.
Write buffering
Aggregate increments in memory and flush periodically.
G-Counter / PN-Counter
CRDT counters that merge per-node counts.
HyperLogLog
Probabilistic cardinality estimator for unique counts.
06

How it works, step by step

  1. 1
    Decide accuracy

    Exact (inventory) or approximate (views).

  2. 2
    Shard the counter

    Increment a random sub-counter among K.

  3. 3
    Buffer increments

    Aggregate in memory or a stream, flush every second.

  4. 4
    Read by summing

    Cache the sum briefly for display.

  5. 5
    Reconcile

    Periodically recompute from the event log if needed.

07

Counter strategies

Viral post receiving 50,000 likes per second

Step 1 / 4
StrategyWrite throughputAccuracyRead cost
Single DB row~1-5k/s (lock contention)Exact1 read
Redis INCR~100k/s on one keyExact (unless failover)1 read
K sharded countersK x singleExactK reads (cache the sum)
Buffered batches via KafkaMillions/sEventually exactPrecomputed

NOWStrategy: Single DB row | Write throughput: ~1-5k/s (lock contention) | Accuracy: Exact | Read cost: 1 read

Combine: record like events durably (for dedupe and exactness), aggregate asynchronously, and display a cached, slightly delayed count.

08

Implementation

const SHARDS = 16; // Spread increments across 16 keys to avoid a single hot keyexport async function incrementLikes(postId: string) {  const shard = Math.floor(Math.random() * SHARDS);  await redis.incr(`likes:${postId}:${shard}`);} export async function getLikes(postId: string) {  const cached = await redis.get(`likes:${postId}:total`);  if (cached) return Number(cached);  const keys = Array.from({ length: SHARDS }, (_, i) => `likes:${postId}:${i}`);  const values = await redis.mGet(keys);  const total = values.reduce((sum, v) => sum + Number(v ?? 0), 0);  await redis.set(`likes:${postId}:total`, String(total), { EX: 2 }); // show a 2-second-old count  return total;}
09

Complexity and performance

Sharded incrementO(1)

Random shard.

Sharded readO(K)

Cache the result.

HyperLogLog memory12 KB

For billions of uniques.

10

Trade-offs

Accuracy vs throughput

Buffering and approximation scale enormously but show slightly stale or approximate numbers.

Shard count

More shards spread writes but make reads more expensive.

11

Variants and related techniques

Adaptive sharding

Only split counters that become hot.

Count-min sketch

Approximate frequency counts for many keys in fixed memory.

12

Common mistakes

  • Counting without deduplication.

    Fix: Store (user, post) like records so a user cannot like twice; the counter is derived.

  • Losing buffered counts on crash.

    Fix: Buffer in a durable stream or accept bounded loss for non-critical counts.

13

Interview questions

How do you design a like counter for a post with millions of likes per minute?

Store each like as an idempotent record (user, post), publish like events to a stream, aggregate counts in batches into sharded counters, and serve a cached total. The UI shows the user's own like immediately (optimistic update).

How do you count unique visitors across billions of events?

Use HyperLogLog per page and time window; it estimates cardinality with about 1% error in kilobytes of memory and can be merged across servers.

14

Practice problems

ProblemDifficultyWhat it trains
Implement a sharded counterEasyHot keys.
Design YouTube view countingHardStreaming aggregation and fraud checks.