Distributed Counter

What the interviewer is asking

"Design a distributed counter" usually means the numbers under a video or a post: "1.2M views · 48K likes". For a small site it's one line of SQL, view_count = view_count + 1. The question is what happens at a billion views a day, when every view is a write and one video goes viral.

The interviewer listens for three things: that you notice the writes are as many as the reads, that you spot the hot row, and that you know which counts may be a few off. The walkthrough follows the 8 steps of the interview framework.

1. Scope: what version 1 must do

• Count every view of a video or post.
• Like an item, once per person: a second tap must not add a second like.
• Show both counts on every page, a few seconds behind at most.

That last line is the most useful thing to agree on. Nobody can tell 1,204,331 views from 1,204,345, so the counts may lag a few seconds, and the whole design uses that room. Name the rest and park it: counts that update live on open pages, unique viewers (which means remembering who watched; a sketch such as HyperLogLog estimates it in about 12 KB per video, about 1% off), and breakdowns by country or hour, which are analytics over the same events.

2. Data: two numbers per item, one row per like

table   key            fields
items   item id        view_count, like_count
likes   (user, item)   when it was liked

Each item keeps its two counts on its own row, so a page reads one row instead of adding anything up. Each like is a row keyed by (user, item), which keeps it to one per person and lets you recount likes. A view gets no row of its own: a log of every view would be about 1 billion rows a day, for features that are parked.

3. The API: three calls

POST /items/v42/view
→ 202 Accepted

POST /items/v42/like
→ 200 OK

GET /items/v42/counts
→ 200 OK
{"views": 1204331, "likes": 48210}

The view call is fire and forget: 202 Accepted means "taken, not done yet", so the app never waits for a database write it doesn't need. The like takes its user from the sign-in token, never from the body. Don't let GET /counts count the view as a side effect: a GET must be safe to repeat and to cache, and counts also show in lists and search results, where nobody watched.

4. The busy path: a read and a write on every view

In most designs the busy path is a read: a URL shortener answers about 100 redirects for every link it creates. Here every page view reads the counts and also sends POST /view, so the writes are as many as the reads, and both sides need a plan. Likes are a small share: say 1 like for every 20 views.

5. The numbers

• Views: 1 billion a day ÷ 86,400 seconds ≈ 12,000 a second on average. Dividing by the 1,440 minutes of a day gives views a minute, a common slip.
• Busy hours: about 5× the average, so about 60,000 view writes and 60,000 count reads a second.
• Likes: about 50 million a day, about 600 a second. They're the only thing that grows: about 18 billion rows a year, and at about 50 bytes each about 1 TB a year, before the usual 3 copies.
• A viral video: the interviewer adds that one video, v42, can get about 50,000 views a second.

Say one database server handles about 5,000 simple reads or writes a second: neither side fits on one, and the viral video alone is 10 times that.

6. Services: a counter service and a cache

A counter service takes the three calls, with the counts database and a cache of counts behind it. Reads go to the cache first; on a miss the service reads the row and keeps a copy for about a second (cache-aside). One cache server answers about 100,000 reads a second, and a popular video's row is read about once a second instead of on every page view. Read replicas would be the wrong tool: 60,000 reads would need about 12 of them, each also applying all 60,000 writes a second.

7. Scale: split the counts by item

The writes don't fit either: 60,000 a second at about 5,000 per server means about 12 servers. Split the counts over about 12 shards by item id: a hash of the id picks the shard, so each item's row lives whole on one shard and a read is one lookup.

Then the trap. Every view of v42 carries the same id, so all 50,000 a second land on the same shard and the same row, not 50,000 ÷ 12 ≈ 4,000 per server. Sharding spreads different keys; it can't split one. That's a hot key, here a hot row.

8. Deep dive: the hot row

UPDATE items SET view_count = view_count + 1 WHERE id = 'v42';

Updates to one row take turns: each locks the row until it's saved. Say each holds the lock for about 0.2 ms, so one row takes about 5,000 updates a second, whatever server it's on. Run v42's first viral second:

time    sent     counted   not yet counted
0.1 s   5,000    500       4,500
0.5 s   25,000   2,500     22,500
1 s     50,000   5,000     45,000

45,000 updates wait after one second, and 45,000 more join them every second: a view sent then waits 9 seconds, and the wait grows by 9 seconds every second. A bigger server doesn't fix it, since the lock still lets one update in at a time, and read replicas take no writes.

The fix is sharded counters: split v42's count over many rows, v42#0, v42#1 and so on, on different shards. Each view adds 1 to one row picked at random, and a read adds the rows up. 50,000 ÷ 5,000 = 10 rows at the very least; say 20, about 2,500 updates a second each, to leave room. A small run, item v8 with 4 rows and 10 views:

views   picked            rows now
1–3     #0, #3, #2        1  0  1  1
4–7     #0, #0, #1, #3    3  1  1  2
8–10    #2, #2, #0        4  1  3  2
read    all 4 rows        4 + 1 + 3 + 2 = 10

Random picks never spread exactly evenly, so you can't read one row and multiply: row #1 times 4 says 4 views, row #0 says 16. The price is the read: each read of v42 adds up 20 rows, 1 million row reads a second at 50,000 page views, so the cache keeps the sum for a second. Fewer rows would bring the queue back: with 2, each takes 25,000 updates a second, 5× what a row takes. Only hot items need more than one row.

9. Deep dive: batching

Sharded counters spread v42's writes, but they're still 50,000 writes a second. Batching cuts the writes themselves: each counter server keeps a tally per item in memory and, once a second, writes one += per item it counted, then starts again from 0. Run it small, with 3 servers, 2,500 views of v42 and 6 of v7, a quiet video, in one second:

server   v42 tally   v7 tally   writes when the second ends
A        855         3          2
B        841         2          2
C        804         1          2
all      2,500       6          6

6 writes carry 2,506 views. However busy v42 gets, it costs one write a second per server: with 100 counter servers, 100 writes a second instead of 50,000, far under what one row takes. A quiet video gains little, but quiet videos were never the problem. A queue whose workers write views one by one would still make 50,000 writes a second.

The price is what lives in memory. If Server B crashes halfway through the next second, the 408 views in its tally are gone; its earlier tallies are already in the database. 408 of about 1.25 million is about 0.03%: invisible on a view count, unacceptable for payments or stock.

Where even that loss is too much, each counter server can append every view and like to a stream kept on disk, and separate adders read it, add up in batches and update the counts. Nothing is lost in a crash, but delivery is at-least-once: an adder that crashes starts again from its last saved place and reads a few events twice.

10. Deep dive: exact or approximate

#1  view  v42
#2  like  v42  by ana
#3  view  v42
-- the adder crashes, back to #2
#2  like  v42  by ana
#3  view  v42

Save each like as a (user, item) row, and add 1 to like_count only when the row is new. The second save of (ana, v42) finds the row there and changes nothing: the like count goes from 48,210 to 48,211 once, while the views went up by 3 for 2 real views, which is fine. Keep the like rows on the item's shard, so one transaction saves the row and adds the 1, and if the count ever drifts, a recount from the rows makes it exact again. Exactly-once delivery from the stream wouldn't help, since it can't see your database.

• A few off is fine: views of a video, people watching a live stream right now, the "about 2 million results" line of a search. Batching, streams and estimates are worth it.
• Must be exact: likes (one per person, even if the total lags a second), votes in an election, items left in stock, the money in a wallet. Keep every event as a row and count from the rows, or check at the moment it matters, as stock is at the moment of the sale.

What breaks first

• The hot row: spot the viral item, then use sharded counters, batching, or both.
• A hot sum expiring in the cache: every read misses at once and adds up 20 rows. Let one request refresh it while the others get the old sum.
• Counter servers crashing: each loses up to a second of tallies. Fine for views; the like rows let you recount likes.
• The stream's repeats: each like's write must be safe to repeat, by its (user, item) key.

Summary: the order to follow

1. Scope: views and likes on every page, a few seconds behind at most.
2. Data: view_count and like_count on each item's row; a (user, item) row per like.
3. API: POST /items/{id}/view answered with 202, POST /items/{id}/like, GET /items/{id}/counts.
4. Busy path: every page view reads the counts and writes a view.
5. Numbers: about 60,000 reads and 60,000 writes a second; a viral video at 50,000.
6. Services: a counter service, the counts database, a cache kept a second.
7. Scale: about 12 shards, split by item id.
8. Deep dives: sharded counters for the hot row, the sum cached; batching in memory; likes exact by their rows, views allowed to be a few off.