Design a Top-K Heavy Hitters Service

Rank the top 1,000 videos by views over the last minute, hour, day and all time, from a million views a second.

Last updated 2026-10-26. Difficulty: hard. Patterns: heavy-hitters, count-min-sketch, sliding-windows, stream-processing, hot-keys. Reported at Google, Meta, Amazon, TikTok, X, LinkedIn.

Functional requirements

• Top K items for a window. K up to 1,000, for the last minute, the last hour, the last 24 hours and all time. Each entry is an item id with its count, in rank order; a request for the top 10 or 100 is a slice of the same list.
• Windows slide, they do not reset. “Last hour” at 10:42 means 09:42 to 10:42, not 10:00 to 10:42. The 1-minute list moves every 10 seconds and the 1-hour list every minute; the 24-hour list is minute-fresh at the head and drops its tail an hour at a time.
• Exact counts for the long windows. 24-hour and all-time counts must agree with a full recount of every view, except for the hour or so the recount has not reached yet, and the list says how big that part is.
• Every list says how fresh and how exact it is. as_of, the hour up to which counts are exact, and the largest possible over-count on the rest. Ranking, reporting and analytics teams consume these lists too, and they need to know what they are reading.
• Any countable item, not only videos. Hashtags from posts use the same pipeline with the key (type, id). They have far more distinct values than videos, and spammers can mint millions of new ones a minute.
• Removed videos drop out within seconds. A takedown or a video going private removes it from every list without waiting for its counts to age out. A video at number one all time would otherwise never leave.
• Out of scope. Deciding what counts as a view (watch-time rules, bot and fraud filtering happen upstream; we receive validated view events), personalised trending, per-country lists (a follow-up), and the view count on the watch page.

Non-functional requirements

• Ingest rate (1 M views/s peak). About 30 B views a day. Popularity is heavily skewed: one video can be 5 % of all traffic during a live moment, and most videos get a handful of views.
• Freshness of the short windows (≤ 1 min). A 1-minute list that is five minutes old is a different product. This drives the hardest tradeoff in the design: at this rate, fresh and exact do not come from the same computation, so there are two paths.
• Accuracy (exact through the last recounted hour). Short windows may over-count by a stated bound and never under-count, so a true heavy hitter cannot be hidden. Long windows are exact except for the open hour or two, which is bounded the same way.
• Read latency (p99 < 50 ms). A read returns a list that already exists. Nothing is counted, merged or ranked on the read path.
• Read volume (~50 k reads/s, spiky). The trending shelf is on the home page. Almost every read is for one of eight lists, identical for every user, which is what makes reads cheap.
• Counting state bounded (independent of distinct items). Memory must not grow with the long tail or with a spam wave of a million junk hashtags. This is what pushes the fast path towards a sketch.
• Availability of reads (99.99 %). A list a minute old is fine; an empty shelf is not. When counting stalls, serve the last good list with its as_of rather than an error.

Back-of-envelope estimates

• View rate: ~350 k/s avg · 1 M/s peak. 30 B views/day ÷ 86 400 ≈ 350 k/s. Peak is about 3× the average, 1 M/s, so a peak minute is 60 M views and an average hour 1.25 B.
• Raw event volume: ~3 TB/day raw · ~30 TB kept. Event id, item id, timestamp, session hash and a few attributes: ~100 B. 30 B × 100 B = 3 TB/day raw, about 1 TB compressed in columnar files. Thirty days of raw views is ~30 TB, enough to recount any hour of the last month.
• Exact state for a sliding hour: ~15 GB. Assume ~5 M distinct videos in a peak minute at ~50 B per hash-map entry: 250 MB per minute bucket. A sliding hour keeps 60 buckets: 15 GB. A sliding day at minute resolution would be 360 GB. Affordable for an hour, not for a day, and unbounded once hashtag spam arrives.
• Count-min sketch size: 272 k × 5 · ~5.4 MB. For ε = 10⁻⁵ and δ = 1 %: width ⌈e/ε⌉ = 271 829, depth ⌈ln(1/δ)⌉ = 5. 1.36 M counters × 4 B = 5.4 MB per bucket, however many distinct items there are. 60 minute buckets for the hour: ~330 MB, about 45× less than exact.
• Sketch error bound, peak minute: ≤ 600 views. Over-count ≤ εN with probability 1 − δ: 10⁻⁵ × 60 M = 600 views. If popularity is roughly Zipf, the 1,000th video in a peak minute has ~4 k views, so the bound is ~15 % at the bottom of the list and negligible at the top, where the leaders have hundreds of thousands.
• A viral video after pre-aggregation: ~32 records/s. A video at 5 % of peak is 50 k views/s. With 64 pre-aggregators each emitting one combined count per item every 2 s, its counter partition receives 64 ÷ 2 = 32 records/s instead of 50 000. The hot key disappears; the total volume barely changes.
• Exact 24-hour recount: ~720 M rows. Each closed hour collapses 1.25 B views into ~30 M (item, hour) rows. Recomputing the 24-hour totals reads 24 × 30 M = 720 M rows: a few minutes of batch, cheap enough to recompute every hour instead of trusting a running total through late corrections.
• Origin reads for the lists: ~500/s. 8 lists (2 types × 4 windows) × 3 allowed sizes of k = 24 cacheable responses. With ~200 CDN PoPs and a 10 s TTL, the origin sees at most 200 × 24 ÷ 10 ≈ 480/s, whether users send 5 k or 500 k reads a second. Read volume is not a design driver.

Components

• Video player: Web, mobile and TV players. When playback qualifies as a view, the player adds a view event with a client-generated event id to a batched beacon, and retries with the same id if the beacon fails. Playback never waits on it.
• Post service (hashtag uses): Another team’s system. Emits one event per hashtag per published post, which enters the same pipeline under the key (hashtag, tag). Hashtags have far more distinct values than videos and a spam problem, which is one reason counter memory is bounded.
• View ingest (edge · stateless): Accepts beacons at edge PoPs, checks signature and schema, and appends events to the view log in batches. Keys each event by session hash, not by video, so traffic spreads evenly whatever is popular. Holds no state, so it scales with requests and never sees a hot key.
• View log (Kafka · 256 partitions by session): The durability boundary and the fork between the two paths: the fast path reads it live, and an archive sink copies it to object storage. Keyed by session hash, so partitions stay balanced during a viral spike. Kept 3 days, so a broken consumer can rewind instead of losing views.
• Pre-aggregator (combine 2 s, then keyBy item): 64 stream tasks, each reading four log partitions, that sum views per (item, 10-second bucket) in a hash map and flush every 2 seconds. Only then is anything routed by item. The combine is what turns a video at 50 k views a second into 32 records a second at its counter partition.
• Window counters (count-min sketch + min-heap): 64 partitions, each owning a hash range of items, so an item’s whole count lives in one place. Per bucket, a count-min sketch and a min-heap of the partition’s top 1,000: six 10-second buckets for the sliding minute, a ring of 60 minute buckets for the sliding hour, and 24 hour sketches for open hours of the day. About 8 MB per partition, so a checkpoint every 10 s restores in about a second.
• Raw archive (object store · Parquet by hour): Every view event, written by a sink on its own consumer group and partitioned by event hour. An hour is closed 10 minutes after it ends, to let late beacons land. Kept 30 days: the input to every exact count and to any recount after a bug.
• Exact batch (Spark · hourly): For each closed hour: drop duplicate event ids, count per item with a map-side combine, and overwrite that hour’s partition of exact counts. Then recompute the 24-hour totals from 24 hourly partitions, apply the hour to the all-time totals, and write the exact candidates for both lists. Usually done by 20 past the hour.
• Exact counts (wide-column · keyed by item): Hourly counts per (item, hour), the exact 24-hour and all-time totals per item for point reads, and the exact top 3,000 candidates per window with the hour they are exact until. Written only by the batch, read by the list builder. Hourly partitions are overwritten whole, so a rerun replaces rather than adds.
• List builder (merge · stitch · publish): Merges the 64 partitions’ top lists into the global 1-minute list every 10 seconds and the 1-hour list every minute. Once a minute, builds the 24-hour and all-time lists by adding exact closed hours to the sketch’s estimate for the open ones. Hydrates titles, drops removed videos, and publishes each list as a new immutable version.
• Trending page: The home-page shelf, the trending tab, and internal consumers such as ranking and reporting jobs. Shows “updated 20 s ago” from as_of, and asks for k = 10, 100 or 1,000 so that every response is cacheable.
• CDN (10 s / 30 s TTL): Caches the 24 cacheable responses at every PoP: 10 seconds for the short windows, 30 for the long ones, with stale-while-revalidate. Origin traffic is set by PoPs and TTL, not by users: about 480 requests a second at worst. Takedowns purge the affected keys.
• Top-K API (stateless · two reads): Reads the current version pointer and the blob, slices the first k, and returns it with as_of, exact_until and max_overcount. No counting, merging or ranking happens here, which is what keeps p99 under 50 ms at any read rate and keeps the shelf up when counting stalls.
• Top-K lists (Redis · versioned blobs): One blob per (type, window, version), about 100 KB for 1,000 hydrated entries, plus a pointer to the current version of each list. The builder writes the new version and then moves the pointer, so a reader never sees a half-built list, and rollback is moving the pointer back.
• Video metadata (titles · status events): The video catalogue, owned by another team. The builder batch-reads titles and thumbnails for each list and subscribes to status changes, so a takedown or a video going private is filtered out of every list within seconds.

User flows

1. One view, from the player to the 1-minute list. The fast path. Note what is not on it: no per-view database write, nothing keyed by video until counts have been combined, and no ranking at read time.
   1. The player sends a batched beacon containing a qualified view and its event id. The player decides a view happened, by rules upstream of us, and mints the event id, so a retried beacon carries the same id and the exact path can drop the duplicate. The beacon is fire-and-forget: a slow ingest must never stall playback.
   2. Ingest appends the events to the view log, keyed by session. Keying by session rather than by video is deliberate: a partition’s load then follows the number of watchers, which is smooth, rather than the popularity of one video, which is not. From here the view is durable and both paths can read it.
   3. Hashtag uses from the post service enter the same way, under their own item type. Everything downstream counts an opaque key (type, id), so a new item type is a new producer, not a new pipeline. Hashtags differ in the way that matters: millions of distinct values a day, most used once, many minted by spammers. Counter memory must not grow with them.
   4. A pre-aggregator sums views per item and 10-second bucket for 2 seconds, then routes each sum to the item’s counter partition. This combine is what makes keying by item safe. A popular video seen 1,500 times in 2 seconds by one pre-aggregator leaves it as one record, not 1,500. The bucket is part of the key, so a late view is summed into its own bucket and nothing straddles two.
   5. The counter partition adds the count to that bucket’s sketch and offers the item to the bucket’s heap. An add increments one counter in each of 5 rows; the estimate is the minimum of the 5. If the estimate beats the heap’s smallest entry, the item replaces it, an O(log 1,000) step. Memory is the sketch plus 1,000 entries, however many items pass through.
   6. Every 10 seconds, each partition reports its own top 1,000 for the last 60 seconds, and once a minute for the last hour. A partition reports its own top 1,000, not 1,000 ÷ 64. Because every item lives on exactly one partition, the global top 1,000 is guaranteed to be inside the union of these lists, so the merge is exact with respect to the estimates.
   7. The builder merges the 64 lists, hydrates titles, drops removed videos and publishes a new version. The merge waits for all 64 reports for the same as_of. Titles come from one batch read of 1,000 ids; status changes arrive as events and feed a local deny set. The version is written before the pointer moves, so readers see the old list or the new one, never half of each.
   8. The trending page reads the 1-minute list through the CDN. A view reaches the list after about 2 s of combine, up to 10 s until the next report and up to 10 s of CDN TTL: well under the one-minute target. The response carries its as_of, so the page can say how fresh it is instead of pretending to be live.
2. Build and serve the 24-hour list. The read side picks per hour bucket: exact where the batch has finished, the sketch’s estimate where it has not. The work happens once a minute in the builder, never per read, and a takedown is applied at the same place.
   1. Once a minute, the builder reads the exact 24-hour candidates and the hour they are exact until. The batch writes the top 3,000 by exact count over the closed hours of the window, three times K, so items the open hour pushes up or down still have their exact base. exact_until is the first hour not yet recounted: the current hour, or the previous one until about 20 past.
   2. It asks the counters for open-hour estimates of those candidates, and takes their own top lists for the open hours. Two kinds of candidate: items with a big exact base, and items that took off in the open hour and may have almost no base. Missing either set gives a wrong list. Estimate queries are routed by the same hash as the views, so each partition answers for its own items.
   3. For open-hour risers that are not among the exact candidates, it point-reads their exact 24-hour base. A video uploaded this morning that is now viral has a small exact base and a large estimate. A multi-get of a few hundred ids keyed by item takes milliseconds, which is why the 24-hour totals are stored by item and not only as a ranked list.
   4. It scores every candidate as exact closed hours plus the open-hour estimate, and keeps the top 1,000. Only the open part can be wrong, and only upwards. With up to 80 minutes open at 1.25 B views an hour, the bound is about 17 k views, roughly 1 % of the 1,000th video’s daily count and nothing at the top. That number goes into the list as max_overcount.
   5. It hydrates, filters and publishes the version with exact_until and the bound. The list is a statement about a moment: built at 10:42, exact until 10:00, over-count at most 17 k after that. A consumer that needs exactness, such as a daily report, waits for a list whose exact_until has passed the period it cares about.
   6. A client asks for the top 100 over 24 hours, and the CDN usually answers. Everyone asking for the same window gets the same bytes, which is the property that makes this list cheap. The long windows get a 30 s TTL because they change only once a minute.
   7. On a miss, the API reads the pointer and the blob and returns the first 100. Two key reads and a slice. If the builder is behind, the API still returns the last version and its as_of; an old list beats an error on the home page.
   8. A takedown arrives; the builder republishes every list without the video and purges cached copies. Counting goes on as before: views are facts, and visibility is a filter applied when lists are published. Republishing is instant because the builder holds more candidates than it publishes, so the 1,001st item moves up without a recount.
3. A video goes viral at 50 k views a second. The scale-breaking case. Keying by item is what makes the merge correct, and it is also what would melt one partition. Combining before the shuffle is the fix.
   1. A video goes from 100 to 50 k views a second in five minutes. That is 5 % of peak traffic on one key. Ingest does not notice: it is stateless, spread across PoPs, and routes by session, so the spike is smeared across every partition of the log.
   2. The view log stays balanced because nothing upstream is keyed by video. Had the log been keyed by video, one of 256 partitions would take 50 k events a second against an average of about 4 k, its consumer would lag, and every other video on that partition would go stale with it.
   3. Each of the 64 pre-aggregators sees about 780 of its views a second and folds them into one running count. The combine is a hash-map increment, the cheapest operation in the pipeline. Its cost per view does not depend on popularity, so the viral video costs no more per view than any other.
   4. The video’s counter partition receives 32 partial counts a second instead of 50 k views. 64 pre-aggregators × one record per 2 s. Pre-aggregation barely changes the total record count, because most items are seen once per flush anyway; it changes the maximum per key, which is the only number a hot key cares about.
   5. Its estimate passes the heap minimum, enters the bucket’s heap and climbs. Each partial count is one sketch update and one heap update, so the partition does 32 updates a second for this video. The sketch never under-counts, so collisions cannot hide a true heavy hitter; at worst a borderline item rides slightly high.
   6. It is in the next 1-minute list within seconds, and climbs the 1-hour list as minutes accumulate. The 1-hour list needs the views to accumulate, which is correct: a video that is huge for one minute is trending, not yet the most viewed of the hour. The 24-hour list picks it up through the open-hour estimate before any batch has seen it.
   7. Trending traffic triples as people come to look; the CDN absorbs it. Origin load is PoPs × keys ÷ TTL, about 480 a second, however many users there are. A viral moment for the counting tier is a non-event for the read tier.
4. A counter partition dies mid-minute. The failure that tests the freshness promise. Lists must stay correct while they briefly stop moving, and recovery has to take seconds, which is one reason the counting state is kept small.
   1. The worker holding counter partition 17 dies; its counts since the last checkpoint are gone from memory. Partition 17 owns 1/64 of all items. Its state is about 8 MB: six 10-second, 60 minute and 24 hour sketches at 85 KB each, plus heaps. Checkpoints run every 10 s, so at most 10 s of counts are lost from memory, and none from the log.
   2. The job restores the partition from its last checkpoint, together with the log offsets that produced it. State and offsets are snapshotted together, so after a restore the job is exactly where it was at the checkpoint. A few MB restores in about a second; 15 GB of exact hash maps would take minutes, which alone would break the one-minute promise.
   3. The pre-aggregators rewind to those offsets and replay about 10 seconds of views. The restored sketches do not contain those views, so replay applies them exactly once. That matters more than usual: a sketch, unlike an upsert of absolute totals, has no way to tell that it has already applied an increment.
   4. Meanwhile the builder refuses to publish a merge that is missing a partition. A merge of 63 lists would silently drop every item on partition 17, perhaps including the current number one. A list one report old is less wrong than a complete-looking list missing 1/64 of the catalogue, so the builder waits, up to a limit, and then marks the list stale.
   5. The API keeps serving the last complete version with its as_of. The trending page shows “updated 25 s ago”. The availability target is about serving something correct and labelled, not something new.
   6. Partition 17 reports again, and the next merge is complete. Total disruption: one or two report intervals of a frozen list. The exact path never noticed, because the archive sink reads the log on its own consumer group.
5. An hour closes and becomes exact. The slow path. Raw events are recounted with deduplication and no approximation, the long lists swap an estimated hour for an exact one, and the fast path is graded against the result.
   1. The archive sink writes the hour’s views to object storage and closes the hour 10 minutes after it ends. Partitioned by event time, not arrival time, so a beacon that arrives 6 minutes late still lands in its own hour. The 10-minute grace is a choice: longer counts more late views and delays exact_until.
   2. The batch job drops duplicate event ids and counts views per item, combining on the map side. Two shuffles: by event id to remove retried beacons (event ids are uniform, so no skew), then by item to count. The second would put the viral video’s 180 M views for the hour on one reducer; map-side combining sends one partial count per mapper instead, the same idea as the pre-aggregator.
   3. It overwrites hour 09 in the exact counts store. Overwrite, not insert: a rerun after a bug or a late correction replaces the hour. Every downstream total is derived from hourly partitions, so fixing an hour fixes everything built on it at the next recompute.
   4. It recomputes the 24-hour totals from the last 24 hourly partitions and applies the hour to the all-time totals. The 24-hour totals are recomputed (720 M rows, a few minutes) rather than maintained by adding the new hour and subtracting the expired one, so a corrected old hour cannot leave a running total permanently wrong. All-time cannot be recomputed cheaply, so it is incremented, but by the difference from what was last applied for that (item, hour), so a rerun is safe.
   5. It writes the exact top 3,000 candidates for both windows and moves exact_until to 10:00. All-time counts only grow, so the new all-time top list is the old one plus any item touched this hour whose total passed the 3,000th: an exact incremental update with no sort over a billion rows. The 24-hour window loses an hour at its tail, so its candidates come from the recomputed totals.
   6. The builder sees the new exact_until and rebuilds the long lists with one fewer estimated hour. max_overcount drops from about 17 k to the open hour’s bound. Ranks near 1,000 may shift slightly as estimated counts become exact; ranks near the top do not move, because the gaps there are far larger than any error.
   7. The builder grades the fast path: the 1-hour list it published at 10:00 against the exact hour 09. At 10:00 the sliding hour covered exactly 09:00 to 10:00, so the two are directly comparable. Recall at 100 and at 1,000, the largest over-count against the bound, and the largest rank displacement in the top 100. Recall at 100 under 99 % or an over-count above the bound is an alert: a hashing bug, a mis-sized sketch, or a partition silently dropping input.

Deep dives

The theory behind it

• Count-min sketch: an interactive playground. Count a million hashtag mentions in a few kilobytes with a count-min sketch, change its width and rows, and see whether it still finds the true top 10.
• Bloom filters, HyperLogLog and count-min sketch: Probabilistic data structures for system design interviews: Bloom filters for membership, HyperLogLog for counting distinct items, count-min sketch for frequencies and heavy hitters, with sizes, error rates and where each is used.
• Counting at scale: How to count likes, views, clicks and unique visitors at high volume: exact versus approximate counts, sharded counters, batching through streams, idempotent counting, HyperLogLog for uniques, windowed counts and reconciliation.
• Hot keys, hot partitions and data skew: Why real traffic is never uniform, how a single hot key or partition takes down a sharded system, and the fixes: key splitting, local caching, request coalescing, write buffering, salting, adaptive rebalancing, and skew in joins and stream processing.

Related

• Design a Distributed Cache
• Design Search Autocomplete
• Design an Ad Click Aggregator
• Design Venmo
• All 83 system design interview questions
• Run this question for a friend: the interviewer kit