The big idea

Why is counting hard?

"Show the view count" seems like the easiest feature ever written. But a viral video gets hundreds of thousands of views per second, and they all want to increment the same number. Every increment is a write to one value — and everyone's fighting over it at once.

A distributed counter makes a single logical count absorb a massive write rate while staying cheap to read. The core trick is to stop treating it as one number: shard it into many sub-counters, spread writes across them, and sum on read. Add batching, caching, and (for uniques) probabilistic sketches, and you can count anything at any scale.

How to read this: Each step opens with a real design decision — make the call before I show you what ships. Watch the design grow, and at the end lose a shard and drop the total cache to see the accuracy/availability trade. Hit Begin.

Step 1 · One row, one bottleneck

UPDATE n = n + 1 melts

The textbook version: a row (key, count), and each event runs UPDATE counter SET count = count + 1 WHERE key = ?. Correct. So why does a hot key destroy it?

Design decision: One row, UPDATE count = count + 1 per event, on a viral key. What breaks?

The call: Every increment locks the same row, serializing all writes into a single contention point. — A read-modify-write on one row must take a lock; under hundreds of thousands of concurrent increments they queue behind each other, throughput collapses, and latency spikes. The single row is a write hotspot.

Every increment is a read-modify-write on the same row, so it must take a lock. Under a viral key, hundreds of thousands of increments per second all serialize behind that one lock — throughput collapses and latency explodes. The number isn't the problem; the single point of contention is. You must spread the writes.

Write contention: Concurrency dies when everyone writes the same cell. A lock (or optimistic-retry) on one row turns parallel traffic into a single-file queue. The whole design is about removing that one shared point that every writer fights over.

Step 2 · A service in front

Own the counting logic

Before sharding, put a counter service between the event sources and storage so the increment/read logic lives in one place. But the service alone doesn't fix contention — what has to change behind it?

Add a counter service that owns two operations: increment(key) and read(key). It hides how the count is stored, so you can change the storage strategy (sharding, batching, caching) without touching callers. The service is the seam; the next steps make what's behind it scale. The key realization: you can trade a little read cost and a little freshness for enormous write throughput.

Trade reads for writes: Counters are almost always write-heavy and read-tolerant of slight staleness (nobody notices if a view count is a second behind or off by a few). That asymmetry is the lever: make writes cheap and parallel, accept a bit more work on read and a bit of lag. Every technique below spends read/freshness to buy write scale.

Step 3 · Shard the counter

Many sub-counters, summed

One row can't take the write rate. How do you let thousands of increments to the same logical key happen in parallel without them fighting over one value?

Design decision: How do you let many increments to ONE key run in parallel?

The call: Retry the single UPDATE faster on conflict. — Optimistic retries still all target one row, so under high contention they mostly collide and retry — throughput still collapses. You must physically spread writes across multiple cells, not retry the same one.

Shard the counter: keep N sub-counters for each key (e.g. views:vid42:#0..#15). Each increment picks a shard — at random or by the writer's id — so writes spread across N independent cells and run in parallel, N× the throughput of one row. A read sums the N shards. You've traded a cheap N-way sum on read for a massive gain in write concurrency.

Sharded counters: This is the core idea (App Engine's "sharded counters", Redis with N keys, etc.). Pick N to match the key's heat — a viral key wants more shards. The only cost is that reading now sums N cells, which the next steps make cheap by caching the sum.

Step 4 · Don't write per event

Batch through a stream

Even sharded, hitting the store on every single event is a lot of tiny writes. Many events can be combined: 500 increments in a 100ms window are just +500. How do you collapse them?

Design decision: Sharded, but still one store write per event. How do you cut the write rate further?

The call: Feed events through a stream; aggregate locally and flush +N periodically. — Events go onto a durable stream (Kafka); consumers aggregate counts per key in memory over a short window and flush a single +N to the shard. The store sees far fewer, larger writes, and the stream gives durability + replay + dedupe.

Route increments through a durable stream and aggregate before writing. Consumers (or the app itself) sum increments per key in memory over a short window and flush a single +N to a shard instead of N separate +1s. The store sees far fewer, larger writes; the stream adds durability, replay, and a natural place to deduplicate. Local pre-aggregation + sharding together tame almost any write rate.

Aggregate then flush: Write-behind aggregation trades a little latency (the flush interval) for a huge reduction in write volume. Combined with sharding, it turns a firehose of tiny +1s into a trickle of batched +Ns — the store barely notices the viral key.

Step 5 · Cheap reads

Cache the total

Reads now sum N shards, and popular counts are read constantly (every viewer sees the number). Summing shards on every read re-introduces load. How do you make reads O(1)?

Cache the summed total. A background job periodically sums the shards and writes the grand total to a single cached value that reads return directly — O(1), one lookup. The cache is refreshed every second or few, so the displayed count lags reality by only that window (invisible for view/like counts). Reads hit the cache; writes hit the shards; the two are decoupled.

Read the sum, not the shards: The cached total closes the loop: sharding made writes fast but reads a sum; caching the sum makes reads fast again, at the cost of a second or two of staleness. Write path (shards) and read path (cached total) now scale independently.

Step 6 · Counting *unique* things

HyperLogLog for cardinality

"Unique viewers" is a different, harder count: you must not double-count the same user. Storing every viewer's id to dedupe would cost gigabytes per popular video. How do you count distinct items cheaply?

Design decision: Count UNIQUE viewers without storing every id. How?

The call: Store every viewer id in a set and take its size. — An exact set of ids costs memory proportional to the number of uniques — gigabytes for a viral video, and it grows without bound. When approximate is acceptable, a sketch is thousands of times smaller.

Use a HyperLogLog sketch. It estimates cardinality (distinct count) from the bit-patterns of hashed items in a fixed few kilobytes — no matter if there are a thousand uniques or a billion — with ~2% error. Add an item and it may update the sketch; ask for the count and it estimates. It even merges (union two sketches to combine days/regions). For "unique viewers/visitors," it replaces gigabytes of ids with a tiny, mergeable summary.

Probabilistic counting: Exact distinct counting costs memory ∝ uniques. HyperLogLog trades ~2% accuracy for O(1) memory by observing that the more distinct hashed values you see, the more likely you are to see rare bit-patterns (like a long run of leading zeros). Mergeable sketches make windowed/segmented uniques cheap too.

Step 7 · Durability & exactly-once

Don't lose (or double) counts

In-memory shards are fast but volatile, and a retried request could increment twice. For view counts a tiny drift is fine — but you still want the count durable and, where it matters, exactly-once.

Flush shards to a durable rollup (the source of truth) so a lost shard is rebuilt, not lost. For exactly-once increments, attach an idempotency key (event id) and dedupe it — at the stream (offset/exactly-once semantics) or with a seen-set — so a retry adds nothing. Decide per use case: view/like counts tolerate at-least-once and slight drift; billing or inventory counts demand exactly-once and strong durability.

Idempotent increments: "Add 1" isn't naturally idempotent — retrying adds again. Make it idempotent by tagging each increment with a unique id and ignoring ids you've already applied. How hard you enforce this is a business call: approximate for vanity metrics, exact for money.

Step 8 · The sharp edges

Hot keys, races & time windows

Reality: one key goes supernova (a hotter hot key), naive read-modify-write still races, and product wants "views in the last hour," not just all-time.

For an extreme hot key, increase its shard count dynamically and add an edge/CDN counter that pre-aggregates before the service even sees it. Avoid races by using atomic increments (INCRBY) or the stream-aggregate path — never a manual read-then-write. Serve time-windowed counts with per-bucket counters (per-minute/hour sub-counters you sum over the window) or a decaying counter. And pick consistency deliberately: eventual + approximate for scale, strong + exact only where correctness is money.

Design for the unhappy path: Supernova key → more shards + edge pre-aggregation. Race → atomic INCRBY, not read-modify-write. "Last hour" → time-bucketed counters. Money → exact + durable. The sharded, cached, batched core flexes across all of these by tuning N, the window, and the consistency you demand.

You did it

You just designed a distributed counter.

• A —
• A —
• S — h
• B — a
• C — a
• C — o
• F — l