The whole design, in writing
Learn system design by building a distributed counter for view counts, likes and rate metrics at massive scale, step by step. An interactive guide covering why a single UPDATE row melts under write contention, sharded sub-counters that trade read cost for write throughput, batching through a stream, caching the total, HyperLogLog for cheap unique counts, idempotent exactly-once increments, and the unhappy paths (hot keys, races, time windows).
Every step of the build above, written out: the problem each piece solves, the option that was taken and the ones that were not, the numbers, and how it fails in production.
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.
What the new pieces do
- Event Sourcesclient
- A firehose of increment events — every video view, every like, every request to rate-limit — from millions of users.
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?
One row, UPDATE count = count + 1 per event, on a viral key. What breaks?
A 64-bit integer counts to billions of billions — magnitude isn't the issue. The problem is contention: every increment must lock and serialize on the same row.
Reading one value is trivial. The bottleneck is writes: all increments target the same row, so they serialize on its lock and can't go faster than one-at-a-time round trips.
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.
Why this piece earns its place
The tempting response is to buy a bigger database, and it is worth being precise about why that fails. The ceiling here is not CPU or disk, it is the time the row's lock is held, times one. If a single increment holds that row for a network round trip plus a durable write, the system does one increment per that interval no matter how many cores sit behind it, because every writer is queued on the same cell. More hardware moves a number that is not the bottleneck. The degradation is not gentle either: as arrivals approach that serialized service rate the queue runs away, so a key does not get slower in proportion to its popularity, it falls off a cliff somewhere in the middle of the launch. And the damage rarely stays local — increments waiting on the lock hold connections out of a shared pool, so one viral key pushes latency onto unrelated queries that never touch it. That blast radius is why this gets fixed structurally instead of tuned around.
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.
Why this piece earns its place
A seam only buys freedom if its interface stops promising things the implementations behind it cannot keep, and there is one specific mistake here that closes every door in the steps that follow: having increment return the new total. It is the obvious signature, callers love it, and it quietly requires the write path to stay synchronous, unsharded and unbatched forever — you cannot answer it from one random sub-counter, and you cannot answer it at all while the increment is still sitting in a buffer. So increment acknowledges and returns nothing, and read is a separate call that advertises what it really is: a recent value, not a transactional one. The same goes for a caller that increments and immediately reads expecting to see itself. That is the one thing this design will never do, so it belongs in the contract on day one rather than being discovered later by a feature team. Deciding what the API is allowed to promise is the work of this step; the box on the diagram is the easy part.
What the new pieces do
- Counter Servicegateway
- Applies increments to a shard and answers reads. Never a single row — it spreads writes across many sub-counters.
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?
How do you let many increments to ONE key run in parallel?
Instead of one row, keep N sub-counters for the key. Each increment picks a shard (random or by writer), so writes spread across N locks and run in parallel; a read sums the N shards. You trade an N-way sum on read for N× write throughput.
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.
Serializing through a queue preserves the single-writer bottleneck — you've just moved the queue. Sharding lets increments truly run in parallel across independent cells.
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.
Why this piece earns its place
N is the knob, and it should not be one number for the whole system. Too low and the contention from the last step comes back divided — each cell is still a lock, and a key hot enough to melt one row will melt sixteen. Too high costs you somewhere easy to miss: N is paid per key, not per hot key, so a uniform shard count multiplies the cells of an enormous long tail of keys that get read once and incremented twice. Most keys are cold. So I would set N by heat, starting small and promoting a key as it warms, which is also why the shard count has to be recorded with the key rather than compiled into the reader. The day you raise N, reads must keep summing the cells the old writers used — a read that only covers the new range silently drops every increment that landed before the change, and that reads as a product bug, not a counter bug. Growing N is safe when reads sum a superset; shrinking it is the operation that needs a deliberate merge.
- N shardsper hot key
- random shardon increment
- sum Non read
What the new pieces do
- Shard Acache
- One of N sub-counters for a key. An increment hits a random shard, so no single row is a write hotspot.
- Shard Bcache
- Another sub-counter. Writes spread across shards; the true count is the sum of all of them.
- Shard Ccache
- More shards = more write throughput for a hot key, at the cost of summing more shards on read.
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?
Sharded, but still one store write per event. How do you cut the write rate further?
Batching doesn't lose data if the buffer is durable (a stream/queue); it just combines many +1s into one +N. Per-event writes are wasteful when you can aggregate first.
You can approximate uniques (step 6), but for a total view count you want to actually add every event — just aggregated, not dropped. Batch, don't discard.
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.
Why this piece earns its place
Aggregating in memory makes the consumer stateful, and that changes the shape of failure. The partial sum held between flushes exists in exactly one place, so a crash or a partition rebalance drops it — which is why the stream offset may only advance after a flush is acknowledged, never before. That ordering converts every loss into a possible double count instead, and duplicates are the failure I would rather have, because the durability step later gives me a way to collapse them. The subtler cost is that the count now sits downstream of a queue, and queues fall behind. A lagging consumer does not raise an error or return something that looks stale; it returns a smaller number, which is indistinguishable from a quieter hour. Nobody files a ticket about a view count that looks perfectly reasonable and is wrong. So consumer lag is the health signal for this whole path, watched directly rather than inferred from the counts, and the flush window is sized against how far behind the product can afford to be, not against how much write volume I would like to save.
- +Nnot N × +1
- streamdurable + replayable
- windowaggregate before flush
What the new pieces do
- Event Streambus
- Buffers the increment firehose so it can be batched and deduplicated before touching the counters.
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.
Why this piece earns its place
Two things go wrong with a number published by a background job. The first is that the job can die while reads stay perfectly healthy: the cache still answers, the count simply stops moving, and a frozen counter looks exactly like a counter nobody is incrementing. So the total is stored alongside the time it was computed, and a reader that finds it older than the refresh window treats it as absent and sums the shards live — slower, but moving. The second is that the published total can genuinely move backwards: two refresh runs can overlap and publish out of order, and a shard that is lost comes back short until the durable rollup rebuilds it. Nobody notices a view count a second behind; everybody notices one that counts backwards, and it reads as data loss long after the shards underneath have caught up. For a quantity that only ever increases I would never publish a total below the one already published, and treat a drop as something to investigate rather than a value to show.
- O(1)cached read
- ~1sstaleness
- decoupledwrite path ≠ read path
What the new pieces do
- Cached Totaldata
- The periodically-summed grand total, cached so a read returns one number instead of summing every shard.
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?
Count UNIQUE viewers without storing every id. How?
HyperLogLog estimates the number of distinct items from the pattern of hash values, using a tiny fixed amount of memory (a few KB) regardless of cardinality, with ~2% error. Perfect for "unique views/visitors" where exactness isn't required.
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.
Dividing by a guessed average is not a real distinct count and is wildly inaccurate. HyperLogLog gives a principled, bounded-error estimate in tiny memory.
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.
Why this piece earns its place
The sketch answers exactly one shape of question, and it is worth naming that before product asks for another. Sketches merge, so any union is cheap — this video plus that one, today plus yesterday, one region plus another. Nothing else is. You cannot subtract one sketch from another to get "people who watched last week but not this week", and you cannot intersect two to get an overlap; the arithmetic that looks available on the numbers is not available on the sketches, and doing it on the estimates compounds the error of both. So I would keep sketches at the finest granularity anyone will ever slice by and merge upward, because merging is the only operation I get. The other thing frozen on day one is the hash: the estimate is a property of one hash function and precision, so changing either makes new sketches unmergeable with every sketch already stored. That is a migration, not a config change. And the error is relative, which puts the uncertainty where the numbers are biggest: two percent of a thousand uniques is nothing, two percent of a billion is a gap wider than most counts ever reach. So the sketch is the right answer for a dashboard and the wrong one for anything that gets reconciled against an exact report.
- ~2%error
- few KBfixed memory
- mergeableunion sketches
What the new pieces do
- HyperLogLogservice
- A probabilistic sketch that counts *distinct* items (unique viewers) in tiny fixed memory, with ~2% error.
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.
Why this piece earns its place
Exactly-once is not a flag, it is a second stateful system, and it is worth pricing before promising it. Deduplicating by event id means remembering ids, and remembering every id forever is the unbounded set this whole design has been avoiding. So in practice you remember them for a window, and that window is the guarantee: a retry inside it is free, a retry outside it counts twice. That matters more than it sounds, because the slowest retries come from clients that were offline and are now draining a queue from hours ago — the case where a duplicate is most likely is the case a short window misses. The seen-set also takes the same traffic the counter does, so it is hot in its own right, and needs its own sharding and durability before it is worth anything. All of which is the real argument for deciding this per use case: for views and likes, drift is cheaper than the machine that removes it; for anything denominated in money, the machine is cheaper than being wrong.
- rollupdurable truth
- idempotency keydedupe retries
- per use caseapprox vs exact
What the new pieces do
- Durable Rollupdata
- The durable record the shards flush into — survives restarts and lets lost shards be rebuilt.
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.
Why this piece earns its place
Windowed counts raise a question the all-time count never had to answer: whose clock decides which bucket an event belongs to. A client timestamp is attacker-controlled and wrong on a surprising number of real devices, so the bucket gets assigned at ingest, on a server. But with a stream in front of the counters, ingest is not arrival — an event can land after the bucket it belongs to has already been read and shown to someone. So "views in the last hour" has to be defined against a watermark: buckets stay open for a bounded lateness, close after it, and anything later is either dropped or booked to the current bucket deliberately rather than by accident. Say which, because the two give different numbers for the same traffic. The other cost is cardinality — per-minute buckets multiply cells by however many buckets you retain, for every key — so fine buckets get a short TTL and are rolled up into coarse ones as they age. Otherwise the windowed feature quietly becomes the largest thing in the store.
You did it
You just designed a distributed counter.
Everything you assembled, in order
- A single UPDATE row serializes every increment on one lock — a hot key melts it.
- A counter service lets you trade a little read cost and freshness for huge write throughput.
- Shard into N sub-counters: increment a random shard (parallel writes), sum N on read.
- Batch increments through a durable stream — flush +N instead of N × +1 — to cut write volume.
- Cache the summed total so reads are O(1), lagging reality by only a second or two.
- Count uniques with HyperLogLog: ~2% error in a few KB, mergeable — no storing ids.
- Flush to a durable rollup and use idempotency keys; choose approximate-vs-exact per use case.