Design top-k heavy hitters

Design a top-k most viewed videos system for 10 billion views a day. Start with one counting server, grow it with a split by video, Count-Min Sketches for sliding windows and an exact daily batch, then stress it with the sketch error, late views, one viral video, regions and the batch beside the stream, with code.

Senior28 min core · 47 min full depth

Read first: Estimation Bloom filters and sketches
Scaling patterns: Hot keys and hot partitions Multi-region and active-active
Building blocks: Queues and streams Partitioning Bloom filters and sketches Caching
Related: Search autocomplete Ad click aggregator Leaderboard

“Show the 100 most viewed videos of the last hour, and of the last day, and keep the list fresh.”

It sounds like ORDER BY views DESC LIMIT 100. At ten billion views a day, it is a streaming problem with a counting problem inside it. You cannot keep an exact count for every video in every window without paying for a long tail of videos nobody will ever rank. So you take shortcuts, and each shortcut gives up some exactness. The senior part of this problem is knowing exactly how much.

So this page starts with one counting server, and lets the numbers grow it. Then it tests the result against four hard questions. How wrong may a count be? What does “the last hour” mean when views arrive late? What about one viral video? And how do you merge lists that were counted in different places?

The answer in one minute

If you only have a minute, this is the design. The rest of the page explains each part.

Split the counting by video. Every view of a video goes to the same counting machine. Then each machine’s top list can be merged into the global one exactly.
Count the long windows with a sketch. A Count-Min Sketch is a small grid of counters that estimates any video’s count. It can only count too high, never too low, and we know by how much.
Slide windows by subtraction. Keep one sketch per minute and a running total. When a minute leaves the window, subtract its sketch.
Combine views before Kafka. For a viral video, each ingest server sends one message per second with a count, not one message per view.
An exact batch for the official chart. The live lists are estimates. A daily job counts the archive exactly, removes duplicates, and publishes the chart that pays artists.

Scope it

Four questions decide the design: over which windows, how big k is, how fresh, and whether counts must be exact.

Interviewerminute 0

Design a system that shows the top k most viewed videos. Think YouTube trending, but by raw views.

What they are testing

Whether you pin down the windows, k, freshness and exactness, because those four decide the whole design. Three minutes is right. Drawing Kafka before asking “top over what time window?” is the classic red flag.

These four questions matter most:

Which windows? The last 5 minutes, the last hour and the last day, each moving forward every minute.
How big is k? 100 on the page, up to 1,000 through the API.
How fresh? A list is at most 2 minutes behind real time.
Exact or approximate? The live lists may be approximate. The official daily chart must be exact.

From these answers we make five promises:

Fresh. A view shows up in the live lists within 2 minutes.
Honest counts. A live count is never below the true count. And it’s at most one view in a million of the window’s total above it. Videos nearly tied at rank 100 may swap.
Fast. A top-k read takes under 50 ms at p99, at 10,000 reads a second.
Always up. Reads work 99.9% of the time. A list that is a minute old beats an error.
Never lost. A view we accepted is never lost before the daily batch has counted it.

We leave out three things. Personalized trending is a ranking problem that uses these counts. Fraud scoring of single views is out too: we only rate-limit and cap. And billing per view is the ad click aggregator.

Go deeperMore questions, and what each changes1 min

Question	We assume	If the answer were different
“All time” too?	No	A running counter per video and a sorted index; no windows, no sketches
How many events?	10 billion views a day	100 million a day fits one server with exact counts
k of 100,000?	No	A full ranking: exact counts per video and a sorted store, not a heap
Does the count drive money?	Payouts use the batch counts	Billing-grade live counts need exactly-once processing per event
Count views or distinct viewers?	Views, with a cap per user against abuse	Distinct viewers need a HyperLogLog per candidate, not a counter
History needed?	Yes: “what was trending at 3 pm yesterday?”	Without history, published lists can live only in a cache
Per-country lists?	Later	Each extra dimension multiplies the counting state

The numbers that decide it

Now we work out the numbers. In this problem, memory shapes the design more than disk does.

Assumptions

View events a day: 10 billion
Peak over average: 3×
Bytes per event: 100 bytes
Distinct items viewed in an hour: 50 million
Distinct items viewed in a minute: 3 million
Aggregator partitions: 32
Allowed overcount per million views: 1
The 100th item's views in a day: 1 million

Events120,000/s350,000/s at peak

10 billion views a day ÷ 86,400 seconds, and about three times that in the evening. One process that reads and counts events manages about 100,000 a second.

So we split the counting by video from the start.

Distinct videos3Ma minute; 50M an hour

About 3 million videos get a view in any minute, 50 million in an hour and 200 million in a day. The top 100 of a day have about a million views each. Most of the 200 million have fewer than ten.

Exact state88 GB

A day that moves every minute needs a count per video for each of 24 hours, and for each of the 60 minutes of the current hour. At about 64 bytes per entry: (24 × 50 million + 60 × 3 million) × 64 bytes ≈ 88 GB.

So we exact counting fits on 32 machines, but nearly all of it would be spent on videos that will never be near rank 100.

Sketch state4.7 GB

To keep the error under one in a million, each of 32 partitions needs about 85,000 counters per row, 5 rows, and 87 sketches per partition.

So we use sketches for the long windows: one twentieth of the memory, sized by the accuracy we ask for, not by the number of videos.

Sketch error10,000views a day, worst case

One in a million of a day’s 10 billion views is 10,000 views: about 1% of the 100th video’s million.

So we accept the sketch for live lists: only near-ties at the cutoff can swap.

Raise the views tenfold and two decisions flip. One Kafka message per view stops fitting, so we must combine at ingest. And the error grows to 10% of the 100th video, so the sketch must widen. Try it in the calculator.

Change an assumption

View events a day10 billionPeak over average3×Bytes per event100 bytesDistinct items viewed in an hour50 millionDistinct items viewed in a minute3 millionAggregator partitions32Allowed overcount per million views1The 100th item's views in a day1 million

Events, average: 120,000/s
Events, peak: 350,000/s; Past the 100,000 events/s one counting process takes: partition by item id.
Ingest bandwidth, peak: 35 MB/s; Not a design driver: a handful of Kafka brokers take it.
Raw events a day: 1 TB; Kept 7 days in Kafka and archived to object storage for the daily batch.
Exact state, sliding hour and day: 88 GB; Far more RAM than a top 100 needs, almost all of it for long-tail items nobody ranks: use sketches for the hour and day.
Sketch width per partition: 85,000; Counters per row. Each partition sees 1/parts of the views, so it needs 1/parts of the width.
Sketch state, all windows: 4.7 GB; 5 rows of 4 byte counters, 87 sketches per partition (60 minutes, 24 hours, 3 sums). Set by accuracy, not by item count.
Worst overcount, day window: 10,000; Holds for each item with probability 99.3% (5 rows).
Day overcount, % of the 100th item's views: 1; Small enough that only near-ties at rank 100 can swap.

Build the design

We start with the simplest design that is correct. Then we let the numbers break it, one at a time. Each step fixes only what broke, adds only the boxes that fix needs, and says what the fix costs. Throughout, follow one scenario: views come in, and someone asks for the top 100 of the last hour.

Step 1. One counting server

The simplest version is one server that keeps a map from video to count for the current minute, and sorts it when someone asks.

Top-k heavy hitters · architecture

Stage 1: One counting server

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Events per second: 350,000

Stage 1 of 6

One counting server

Why this stage

Nothing yet.

This is the smallest design that returns a top 100.

What we add

One server that keeps a hash map from item to count for the current minute, and sorts it when someone asks.

What it costs

One server takes about 100,000 events/s, a crash loses every count, and there is no window longer than a minute.

At 350k events per second, Counter server is 350% busy. Stage 2 is the fix.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

It has three flaws. One process counts about 100,000 events a second, a third of the peak. A crash loses every count. And there is no window longer than a minute.

Step 2. Split the counting by video

Ingest servers check each batch of views from the apps, then write each view to

Kafka, keyed by video. Kafka hashes the key to one of 64 partitions, so every view of a video lands on the same partition, and so on the same aggregator. Each of 32 aggregators counts its own videos. When a minute ends, each sends its own top 1,000 to a merger, which takes the best 1,000 overall and writes them to Redis. The read API serves from Redis.

Why is that merge exact? Because a video’s whole count lives on one aggregator. A video in the global top 100 must also be in its own aggregator’s top 100. Nothing on that machine can outrank it without outranking it everywhere. So the union of the local top lists contains the true answer.

Split by user instead, and a video’s views are spread over every machine. Then a video that is 150th everywhere can be 1st overall, and no local list contains it. A stress test comes back to this.

Top-k heavy hitters · architecture

Stage 2: Partition by item

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Events per second: 350,000

Stage 2 of 6

Partition by item

Why this stage

The peak is 350,000 events/s, three and a half times one server.

Counts must also survive a crash.

What we add

Ingest servers, Kafka keyed by item id, exact per-minute counters per partition, a merger that unions their top lists, a Redis copy of each list and a read API.

What it costs

Only tumbling one-minute windows, and one viral item lands on one partition.

At 350k events per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

Step 3. Count the windows with a sketch

So far every window is one fixed minute. The last hour and the last day need to forget old views, and forgetting needs to know what to subtract. Exact counts would mean a count per video per minute: the 88 GB from the estimates.

Instead each aggregator keeps one Count-Min Sketch per minute and one per hour, plus a running sum for each window. A view is added to its minute’s sketch and to the sums. When a minute leaves the window, its sketch is subtracted from the sum. Sketches add and subtract cell by cell, so the result is exactly the sketch of the minutes that remain.

A sketch can estimate any video, but it cannot list its biggest ones. So each window also keeps a heap of about 1,000 candidates. The first stress test explains both, with the code.

Top-k heavy hitters · architecture

Stage 3: Sliding windows on sketches

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Events per second: 350,000

Stage 3 of 6

Sliding windows on sketches

Why this stage

Product wants the last hour and the last day, sliding every minute.

Exact per-key buckets for that need about 88 GB of RAM, almost all of it for items nobody ranks.

A requirement and a memory bill force this stage, not traffic.

What we add

A Count-Min Sketch per minute and per hour, running sums that subtract the bucket that leaves, a heap of candidates per window, and event-time windows with a watermark.

What it costs

Counts are estimates: up to about 1 view in a million of the window total too high, so near-ties at the cutoff can swap.

At 350k events per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

Step 4. An exact batch for the official chart

The live day list is an estimate, and it counts retried batches twice. So it cannot pay artists. A connector copies every Kafka partition to object storage, as hourly files. A daily Spark job runs at 06:00 for the day before. It removes duplicates by event ID, drops views flagged by abuse rules, applies the cap per user, counts exactly, and publishes the official top 1,000. Until it lands, the daily list is marked provisional.

Top-k heavy hitters · architecture

Stage 4: Exact daily lists

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Events per second: 350,000

Stage 4 of 6

Exact daily lists

Why this stage

The official daily chart must be exact, deduplicated and include phones that upload views hours late.

A sketch on a stream cannot promise that.

What we add

Raw events archived to object storage, and a daily batch job that counts exactly and replaces the provisional daily list.

What it costs

Two code paths that must agree on what counts as a view.

At 350k events per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

Follow one read and one view through the design:

Top-k heavy hitters · architecture

Trace a request: Read the hourly top 100

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasionalHop in motionHops doneResponse

Read the hourly top 100: step 1 of 3

Users to Top-k API

GET /v1/topk?window=1h&k=100. The load balancer terminates TLS and picks any API server. Most requests stop here: each server keeps every list in memory for 10 seconds.

2 ms of a 50 ms budget

Users → Top-k API2 ms
Top-k API → Top-k cache1 ms
Response

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

Both paths take a few milliseconds. What takes time is the pipeline between them: 35 to 107 seconds from a view to a published list, inside the 2 minute promise.

This is enough for a mid-level answer. Split by video, sketches with a candidate heap for the windows, a merger, and an exact daily batch. The stress tests are what make it a senior answer.

Go deeperWhere the two minutes go1 min

Step	What happens	Delay
App to Kafka	The app batches views; ingest writes them with full replication	0 to 5 s
Kafka to aggregator	The aggregator adds each view to its minute and the sums	Under 1 s
Minute closes	The clock of events passes the minute’s end plus 30 s	30 to 90 s
Aggregators to merger	Each sends its top 1,000 per window	Under 1 s
Merger to Redis to readers	Merge, rank, write; API servers cache lists for 10 s	Up to 11 s
Total	Inside the 2 minute promise	35 to 107 s

Go deeperThe API2 min

POST /v1/events
Authorization: Bearer <device-token>

{ "events": [
  { "eventId": "e7c1...", "itemId": "v_9HJw3SB", "eventTime": "2026-10-28T19:04:11.200Z",
    "country": "BR", "watchMs": 31000 }
] }

202 Accepted

Apps batch up to 50 views, and send them every few seconds. 202 means “safely in Kafka”, not “counted”. The eventId is made on the phone, so a retried batch can be removed later. Views can arrive hours late. (A phone offline on a plane uploads on landing.) That’s why windows use the time on the event. 429 means a device went over its rate limit. The app keeps its batch, and backs off.

GET /v1/topk?window=1h&k=100

200 OK
Cache-Control: public, max-age=10
{ "window": "1h", "asOf": "2026-10-28T19:05:00Z", "provisional": true,
  "items": [ { "itemId": "v_9HJw3SB", "count": 4812004, "maxOvercount": 430 }, ... ] }

window is 5m, 1h, 1d, or day=2026-10-27 for the official chart. asOf says how current the list is. maxOvercount states the sketch’s error honestly: the count can be that much too high, never too low. GET /v1/topk/history?window=1h&at=... returns the list published nearest a time.

Go deeperThe data, and why counting stays in memory1 min

There are three very different stores. A log for events (Kafka, 7 days). Memory for counting (the stream job’s state). And a small key-value store for results (Redis, plus a column store such as ClickHouse for history and daily counts).

event in Kafka, keyed by item_id
  event_id    16 bytes   from the device, for removing duplicates
  item_id     ~12 bytes  the Kafka key
  user_hash   8 bytes    salted hash, for caps per user
  event_time  8 bytes    when it happened on the device; windows use this
  ingest_time 8 bytes    when we received it

Redis   topk:{window}:{scope}              -> JSON list, asOf, provisional
Table   topk_history (window, scope, as_of) -> list
Table   daily_counts (day, item_id)         -> views

The aggregator’s state (sketches, sums, heaps) is saved every minute to object storage, together with its Kafka positions. So state and input always match after a restart.

Option	Problem at this scale
A database row per video	350,000 writes a second, the hottest rows on one machine
Redis `ZINCRBY` on each view	Same write rate; a sorted set per minute; a union of 1,440 for the day
Counting in job memory	No network call per view; a restart replays up to a minute of Kafka

Redis sorted sets are the right answer at a few thousand events a second, and in the leaderboard problem: exact scores, few updates per key.

Stress it

The design is complete. Now we test it where top-k is hard. How wrong is the sketch? Windows and late views. One viral video. Regions that each count part of a video. And the batch beside the stream. The viral video and the regions each add a few boxes.

Exact counts or a sketch

Interviewerminute 18

You said Count-Min Sketch. How does it work, how wrong can it be, and why not something else?

What they are testing

Whether you state the error correctly: a share of the window’s total, with a probability. Whether you know the sketch alone cannot produce a top list. And whether you can name an alternative. Saying the sketch is “1% accurate” for every video is the most common error in this problem.

How the sketch works. It is d rows of w counters, and each row has its own hash function. Adding a video adds its count to one cell in each row. Other videos land in the same cells, so each cell can only count too high. The estimate is the smallest of the video’s d cells.

How wrong it can be. At most ε times the window’s total N too high, with a chance set by the number of rows. Take ε at one in a million, and 5 rows. An estimate is at most 10,000 views high on a day of 10 billion, 99.3% of the time.

Why it needs a heap. The sketch answers “how many for this video?” and nothing else. It does not remember which videos it has seen. So each window keeps a min-heap of candidates. After each update, the video’s estimate is compared with the weakest candidate, and replaces it if larger. The heap holds 1,000 for a list of 100, so a video near the cutoff is still tracked when the window moves. Lists are re-scored against the sketch when published.

TopK.javacount_min_sketch.pyCountMinSketch.cscount_min_sketch.gotop-k.ts

// A Count-Min Sketch: depth rows of width counters. An item adds to one cell per row, and other
// items share those cells, so every row can only overcount it. The smallest of its cells is the
// estimate: never below the true count, and at most (e / width) x the window's total above it,
// with probability 1 - e^-depth for each query.
final class CountMinSketch {
    private final int width; // a power of two, so a mask replaces a modulo on the hot path
    private final int depth;
    private final long[][] cells;

    CountMinSketch(int width, int depth) {
        if (Integer.bitCount(width) != 1) throw new IllegalArgumentException("width: power of two");
        this.width = width;
        this.depth = depth;
        this.cells = new long[depth][width];
    }

    void add(String item, long count) {
        long h = fnv1a(item);
        for (int row = 0; row < depth; row++) cells[row][index(h, row)] += count;
    }

    long estimate(String item) {
        long h = fnv1a(item);
        long best = Long.MAX_VALUE;
        for (int row = 0; row < depth; row++) best = Math.min(best, cells[row][index(h, row)]);
        return best;
    }

    // The sketch is linear: adding two sketches cell by cell gives exactly the sketch of both
    // streams together, so minutes merge into hours and partitions into regions with no extra error.
    void merge(CountMinSketch other) { combine(other, 1); }

    // Linearity also runs backwards: subtracting a minute that left the window leaves exactly
    // the sketch of the minutes that remain. This is why we use plain updates, not
    // "conservative update", which is more accurate but cannot be subtracted.
    void subtract(CountMinSketch other) { combine(other, -1); }

    private void combine(CountMinSketch other, int sign) {
        if (other.width != width || other.depth != depth)
            throw new IllegalArgumentException("only sketches of the same shape can be combined");
        for (int row = 0; row < depth; row++)
            for (int col = 0; col < width; col++) cells[row][col] += sign * other.cells[row][col];
    }

    // Double hashing: row i uses h1 + i * h2. For a sketch this is as good as depth
    // independent hash functions, and it costs one hash of the item instead of depth.
    private int index(long h, int row) {
        int h1 = (int) h;
        int h2 = (int) (h >>> 32) | 1; // odd, so rows never collapse onto the same column
        return (h1 + row * h2) & (width - 1);
    }

    // FNV-1a over the UTF-8 bytes, not String.hashCode: every aggregator in every region, in
    // any language, must put an item in the same cells, or their sketches cannot be merged.
    static long fnv1a(String s) {
        long h = 0xcbf29ce484222325L;
        for (byte b : s.getBytes(StandardCharsets.UTF_8)) {
            h ^= (b & 0xff);
            h *= 0x100000001b3L;
        }
        return h;
    }
}

// The candidates for one window. The sketch counts every item but cannot list its biggest
// ones; it only answers "how many for this item?". The heap remembers which items to ask about.
final class TopK {
    private record Entry(long estimate, String item) {}

    private final CountMinSketch sketch;
    // About 10 x the k we publish: an item just below the cutoff is still tracked when the
    // window slides and the items above it lose their old minutes.
    private final int capacity;
    private final Map<String, Long> candidates = new HashMap<>(); // item -> latest estimate
    // A min-heap with lazy deletion: when an item's estimate grows we push a new entry and
    // leave the old one, skipping it when it reaches the top. No O(k) search on every event.
    private final PriorityQueue<Entry> heap =
            new PriorityQueue<>(Comparator.comparingLong(Entry::estimate));

    TopK(CountMinSketch sketch, int capacity) {
        this.sketch = sketch;
        this.capacity = capacity;
    }

    void add(String item, long count) {
        sketch.add(item, count);
        long est = sketch.estimate(item);
        if (candidates.containsKey(item) || candidates.size() < capacity) {
            track(item, est);
        } else {
            Entry weakest = peekLive();
            // Most of the long tail stops here: one comparison, no allocation.
            if (est > weakest.estimate()) {
                heap.poll();
                candidates.remove(weakest.item());
                track(item, est);
            }
        }
        if (heap.size() > 4 * capacity) compact(); // bounds the stale entries a hot item leaves
    }

    // Estimates are fetched again when we publish: the cached ones only go stale upward,
    // because other items keep landing in the same cells after we last looked.
    List<Map.Entry<String, Long>> top(int n) {
        return candidates.keySet().stream()
                .map(item -> Map.entry(item, sketch.estimate(item)))
                .sorted(Map.Entry.<String, Long>comparingByValue().reversed())
                .limit(n)
                .toList();
    }

    // Called after the window slides. Every estimate fell by the minute that left, each by a
    // different amount, so the heap order is rebuilt from fresh estimates.
    void rescore() {
        candidates.replaceAll((item, est) -> sketch.estimate(item));
        compact();
    }

    private void track(String item, long est) {
        candidates.put(item, est);
        heap.add(new Entry(est, item));
    }

    // Pops stale entries (an older estimate, or an evicted item) until the top is current.
    // A current entry always exists, because every candidate has one.
    private Entry peekLive() {
        while (true) {
            Entry top = heap.peek();
            Long current = candidates.get(top.item());
            if (current != null && current == top.estimate()) return top;
            heap.poll();
        }
    }

    private void compact() {
        heap.clear();
        candidates.forEach((item, est) -> heap.add(new Entry(est, item)));
    }
}

Why not something else? Space-Saving is the better pure top-k algorithm, with the list built in. But its summaries can’t be subtracted, and windows subtract the minute that left.

Go deeperThe four options, and sizing the sketch2 min

N is the total count in the window. m is the number of counters kept.

Approach	Memory	Guarantee	Lists the top?	Add and subtract
Exact hash map	One entry per video	Exact	Yes	Add yes; sliding needs a map per bucket
Count-Min Sketch + heap	w × d counters + a heap	Never under; over by at most εN with probability 1 − δ	Only with the heap	Both exact
Space-Saving	m counters	Every video above N/m is kept; each over by at most N/m	Yes	Add with bounded error; no subtract
Misra-Gries	m counters	Each under by at most N/(m + 1); every video above that is kept	Yes	Add with bounded error; no subtract

The bound. On average, one cell holds an extra N ÷ w. Set w to e ÷ ε. Then the chance one row is more than εN too high is at most 1 in e. The chance all d rows are is e to the power −d.

Sizing. Each partition sees one 32nd of the views, and its error is relative to its own total. So w = e ÷ (ε × 32) ≈ 85,000. The code rounds widths up to a power of two (131,072), so a mask replaces a division. That lowers the error, and raises fleet memory to about 7.3 GB. That’s still a twelfth of exact state.

Conservative update. On an add, raise each cell only as far as the video’s new estimate. It cuts overcounting a lot. But the sketch can then no longer be subtracted safely, so it suits windows that never slide, not our running sums.

Why not a HyperLogLog per video? A HyperLogLog counts distinct users, not views, and one per video brings back the per-video cost we are avoiding. It belongs on the candidates only, if the product ranks by distinct viewers.

Windows, and views that arrive late

What exactly is “the last hour”, and what happens to a view that arrives ten minutes late?

“The last hour” here means a hopping window: one hour long, moving forward one minute at a time. The hour and 5-minute windows are built from minute sketches. The day is built from hour sketches, so its old edge moves an hour at a time and it covers 24 to 25 hours.

A view belongs to the minute it happened, on the phone, not the minute we received it. The aggregator tracks a watermark: the newest event time it has seen, minus 30 seconds of allowed disorder. When the watermark passes the end of a minute, that minute is closed, the window moves, and lists are published. Ingest replaces device times in the future with its own clock, so one phone with a wrong clock cannot close every minute early.

Late views. Sketches are sums. So a late view whose minute is still inside a window is just added to that minute, and to the running sum. The next publish includes it. Only views older than the whole window cannot be added, because their minute was already subtracted. Those are counted as late, and the daily batch counts them from the archive.

A late view is added if its minute is still inside the window

Scroll or pinch to zoom, drag to pan

Color key
Service
Background worker
Queue or stream

SlidingWindow.javasliding_window.pySlidingWindow.cssliding_window.gosliding-window.ts

// One sliding window, for example the last hour, as one sketch per minute plus their running
// sum. An event updates its minute and the sum. When the watermark passes a minute boundary,
// the minute that fell out is subtracted, so a read is one lookup on the sum, never sixty.
// The day window works the same way with 24 hour sketches, each the merge of 60 minutes.
final class SlidingWindow {
    private final int buckets; // 60 one-minute buckets for the hour
    private final int width, depth;
    private final TreeMap<Long, CountMinSketch> byMinute = new TreeMap<>();
    private final CountMinSketch sum; // all buckets added together: what reads ask
    private final TopK top;           // built on sum, so TopK.add updates the sum too
    private long closed;    // newest minute the watermark has passed, in minutes since the epoch
    long late;              // too old for this window: only the daily batch will count these

    SlidingWindow(int buckets, int width, int depth, int capacity) {
        this.buckets = buckets;
        this.width = width;
        this.depth = depth;
        this.sum = new CountMinSketch(width, depth);
        this.top = new TopK(sum, capacity);
    }

    // Keyed by event time, not arrival time: a backlog replayed after an outage lands in the
    // minutes it happened in, instead of looking like a sudden spike now.
    void add(String item, long count, long eventMinute) {
        if (eventMinute <= closed - buckets) {
            late++; // its minute has already been subtracted; adding it now would never expire
            return;
        }
        // A late event whose minute is still inside the window is simply added: the sketch is
        // a sum, so order does not matter. Minutes after the watermark are kept the same way.
        byMinute.computeIfAbsent(eventMinute, m -> new CountMinSketch(width, depth)).add(item, count);
        top.add(item, count);
    }

    // The watermark says no more events older than `minute` are expected (bar a few late ones).
    // Everything at or before minute - buckets leaves the window.
    void advanceTo(long minute) {
        if (minute <= closed) return;
        closed = minute;
        var expired = byMinute.headMap(closed - buckets, true);
        if (expired.isEmpty()) return;
        expired.values().forEach(sum::subtract);
        expired.clear();
        top.rescore(); // every candidate lost a different amount
    }

    List<Map.Entry<String, Long>> top(int k) {
        return top.top(k);
    }
}

Go deeperOther window types1 min

Window type	What it is	Fit here
Tumbling	Fixed, not overlapping: 19:00 to 20:00	The official daily chart (a calendar day)
Hopping	One hour long, moving every minute	Live lists
Truly sliding	Exactly the last 3,600 s at every instant	Never worth it for a list read every 10 s
Exponential decay	Old views count less, no edge	Trending by rate against a baseline, as in typeahead

A longer watermark delay than 30 seconds buys little: views from healthy phones arrive within seconds, and late adds cover the rest. Flink’s documentation explains the model.

One viral video

Traffic grows ten times, and one music video takes 15% of all views on release night. What breaks?

At 3.5 million views a second, two things break. Kafka carries one message per view, more than twice what it was built for. And the aggregator that owns the viral video gets its normal share, plus 15% of all views. That’s 630,000 events a second, for a process that takes 100,000. Adding partitions does nothing for that one video, because its key still hashes to one place.

The fix is to combine before Kafka. Each ingest server keeps a small map of (video, minute) to count, and sends it every second. A viral video with 525,000 views a second becomes about 120 messages a second, one per ingest server. The long tail barely shrinks, but overall, messages drop to about a quarter of views.

Top-k heavy hitters · architecture

Stage 5: Pre-aggregate at ingest

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Events per second: 350,000

Stage 5 of 6

Pre-aggregate at ingest

Why this stage

At 10 times the traffic, 3.5 million events/s overload the Kafka cluster, and the aggregator holding a viral item (15% of all views) needs six times its capacity.

What we add

Ingest servers sum counts per item for one second and send (item, minute, count) instead of single events.

Raw batches go to a separate topic, one record per batch, for the archive.

What it costs

One more second of delay, and per-event deduplication moves entirely to the daily batch.

At 350k events per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

A viral item before and after ingest pre-aggregation

Scroll or pinch to zoom, drag to pan

Color key
Background worker
Queue or stream
Service

The sketch already takes a count with each update, so nothing after Kafka changes. But a combined message has no event IDs, and the batch needs them. So ingest also writes each app batch, whole, to a raw topic that only the archive reads: one message per 50 views.

Go deeperOther fixes for a hot key1 min

Fix	How it works	Cost
More partitions	Spreads videos over more aggregators	Nothing for one video: it still hashes to one partition
Salting the key	`video#0` to `video#7` on 8 partitions	One video’s count is split again, so the merge needs two phases
Combine at ingest	Sum per video for 1 s before sending	1 s more delay; duplicate removal moves to the batch
A separate path for hot videos	Detect hot videos and count them apart	Detection lag, and two code paths for one count

Regions, and merging split lists

Now there are three regions, and each sends its top 100. Can we add them up and take the top 100?

One region is one failure domain. And copying every view across an ocean costs money for nothing. So each region runs its own ingest, Kafka and aggregators. It sends only summaries to a global merger: its closed minute sketches, and its top candidates. Every region uses the same partitions and hash functions. So the merger adds the sketches cell by cell, into one global sketch. That is about 1.4 MB a second per region.

Top-k heavy hitters · architecture

Stage 6: One pipeline per region

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Events per second: 350,000

Stage 6 of 6

One pipeline per region

Why this stage

Viewers are on three continents, one region is one failure domain, and at 10 million events/s one region is past what it can take.

Raw events should not cross oceans.

What we add

The ingest, Kafka and aggregator path in every region.

A global merger combines regional sketches and candidates in two phases.

What it costs

An item is now split across regions, so merging top lists is no longer exact without the second phase; the global list lags a few seconds more.

At 350k events per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

The catch: a video’s count is now split across regions, so the union of regional top lists is no longer exact. Take the top 1 from two regions:

Video	Region A	Region B	Total
a	10	0	10
d	0	10	10
c	8	8	16

Each region’s top 1 is a or d. The merge says they tie at 10. The true winner, c with 16, never left either region.

The two-phase merge fixes this, and can prove it missed nothing. It comes from the TPUT algorithm (Cao and Wang, 2004), here with m regions:

Each region sends its top k. Summing gives a lower bound on each video’s total, so the k-th largest sum, τ, is at or below the true k-th total.
A video whose total reaches τ must have at least τ ÷ m in some region. That’s because m numbers that are each below τ ÷ m add up to less than τ. So each region sends every video at or above τ ÷ m. No true top-k video can be missing now.
The merger scores these candidates against the global sketch, and ranks them.

The two-phase merge across three regions

Scroll or pinch to zoom, drag to pan

Color key
Service

GlobalMerger.javaregion.pyGlobalMerger.csregion.goregion.ts

// What the global merger can ask each region. With sketches, the counts are the region's
// estimates; the argument below holds for any non-negative counts, estimated or exact.
interface Region {
    Map<String, Long> top(int k);                 // its own top k
    Map<String, Long> atLeast(long threshold);    // every candidate with a local count >= threshold
    Map<String, Long> counts(Set<String> items);  // local counts for these items, 0 if unseen
}

// The exact global top k when one item's views are split across regions. Taking each region's
// top k and summing is not enough: an item ranked 150th everywhere can be first in total.
// This is the three-phase TPUT algorithm (Cao and Wang, 2004).
final class GlobalMerger {
    static List<Map.Entry<String, Long>> exactTopK(List<Region> regions, int k) {
        // Phase 1: each region's own top k. A partial sum never exceeds an item's real total, so
        // the k-th largest partial sum, tau, is a floor under the real k-th total.
        Map<String, Long> partial = new HashMap<>();
        for (Region r : regions) r.top(k).forEach((item, c) -> partial.merge(item, c, Long::sum));
        long tau = partial.values().stream()
                .sorted(Comparator.reverseOrder())
                .skip(k - 1)
                .findFirst()
                .orElse(0L); // fewer than k items seen: tau 0 asks for everything, still correct

        // Phase 2: an item whose total reaches tau must have at least tau / m in some region,
        // because m numbers that each fall short of tau / m add up to less than tau. So asking
        // every region for all items at or above tau / m cannot miss a true top-k item.
        long threshold = tau / regions.size(); // rounding down only adds candidates
        Set<String> candidates = new HashSet<>(partial.keySet());
        for (Region r : regions) candidates.addAll(r.atLeast(threshold).keySet());

        // Phase 3: full totals for the candidates only. The paper also drops candidates whose
        // upper bound cannot reach tau before this step; left out to keep the idea visible.
        Map<String, Long> total = new HashMap<>();
        for (Region r : regions)
            r.counts(candidates).forEach((item, c) -> total.merge(item, c, Long::sum));
        return total.entrySet().stream()
                .sorted(Map.Entry.<String, Long>comparingByValue().reversed())
                .limit(k)
                .toList();
    }
}

Go deeperWays to merge, compared1 min

Fix	How it works	Exact?
Send more candidates	Each region sends its top 10,000 instead of k	No: misses become rare, never impossible
Send candidates plus a bound	Also send each region’s k-th count, an upper bound on the rest	Tells you when the answer may be wrong
Two phases (TPUT)	A threshold that can’t miss, then sum for candidates only	Yes, for counts that are never negative
Add sketches, re-score	Score the union of candidates on the global sketch	Counts yes; the candidate set still not

A region answers step 2 from its candidate heaps, since a sketch cannot list videos. So the proof holds only while each region’s weakest candidate is below τ ÷ m. The region reports that floor, and the merger flags the list when it isn’t.

When a region goes dark for ten minutes, the merger publishes with the regions it has and marks the list partial. When the region recovers, it catches up and resends the affected minute sketches, and the hour and day windows heal on their own. Regional lists, “top in Brazil”, become nearly free.

The batch beside the stream

A stream and a batch now compute the same daily counts. Isn’t that two systems that will disagree? Why not one?

Running a batch beside a stream is called the lambda shape. Running one stream for everything, and replaying Kafka to fix it, is called kappa. Here, one exact stream would need the 88 GB of exact state, plus weeks of Kafka for very late views. A batch pays for exactness once a day, over files that never change, and can simply run again.

So we use lambda for one output only: the official daily chart. Two code paths do drift apart, so one shared library decides what counts as a view, for both. And each morning, the batch’s exact top 100 is compared with the stream’s. When they share fewer than 95 videos, someone is paged. Kappa wins when exact counts are needed live and the state is affordable, as in the ad click aggregator.

When things break

Interviewerminute 40

An aggregator crashes mid-minute. What do users see, and are any views lost or counted twice?

What they are testing

Whether you know state and Kafka positions are saved together, so a replay counts each view once.

The rule: reads never fail because counting stopped. Readers get the last published list, with its asOf time.

An aggregator saves its sketches, heaps and Kafka positions in one checkpoint every minute. After a crash, it loads the checkpoint and reads Kafka again from the saved position. Every event after the checkpoint is counted once, into state that didn’t include it. Its videos freeze for about a minute, and nothing is lost.

An aggregator restarts without losing or double counting

Scroll or pinch to zoom, drag to pan

Color key
Database
Background worker
Queue or stream

Click any box to break it and see what users would notice:

Top-k heavy hitters · architecture

Break it: what fails, who notices, what users see

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Break

Failure: click any box to break it

An aggregator goes down

What breaks

Its partitions stop being counted, so its items freeze in the lists.

How we notice

Consumer lag on its partitions and missing checkpoints.

What users see

Lists stay up but go stale for those items for about a minute.

What we do

The job restarts from its last checkpoint (sketches, heaps and Kafka offsets together) and replays from those offsets, so nothing is lost or counted twice.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.

Go deeperEvery failure, and what users see1 min

Part	Failure	What users see	What we do
Ingest server	Dies	Its last second of counts missing live	Apps retry unacknowledged batches; the batch removes duplicates
Kafka broker	Dies	Nothing	A new leader; full replication loses nothing
Aggregator	Crashes	Its videos freeze for about a minute	Restore the checkpoint and replay
Merger	Dies	Lists stop updating; old ones served	A standby takes over; it holds no state
Redis	Primary dies	Seconds of errors after the 10 s cache	Promote a replica; lists are republished every minute
Archive	Writes fail	Nothing, for up to 7 days	Resume from positions; page long before Kafka forgets
Daily batch	Late or fails	The daily chart stays provisional	Run it again; it reads files that never change
Region	Outage	Its views are missing from global lists	Lists flagged partial; apps buffer and upload later

Output	Guarantee	What users can see
Live counts	Never under; over by at most one in a million of the window	Hundreds too high on the hour, up to 10,000 on the day
Live membership	Complete over the candidate heaps	The 101st shown as 100th when nearly tied
Late views, inside window	Counted on the next publish	A past minute’s count rises slightly later
Official daily chart	Exact, no duplicates, views up to 48 h late	Can differ from yesterday’s live list

The checkpoint is what Flink calls exactly-once state. It says nothing about duplicates the apps send: a retried batch is in Kafka twice. Removing those needs an ID per event, which is the batch’s job.

Go deeperRunning it: alerts, abuse, deploys and cost1 min

Wake someone up when:

The newest list is older than 3 minutes.
An aggregator is more than 60 seconds behind.
Ingest success drops below 99.9%.
The morning comparison fails.
Views too late for any window pass 0.5% of traffic. That means a watermark or app release is wrong.

Abuse is the real work. A trending list is a target. A few thousand scripted devices can push a video into the top 100. The defenses: rate limits per device at ingest, a cap on views per user per video per hour, and holding new videos back from the global list until their viewers look varied. A block list, checked at publish time, lets the trust team remove a video within a minute.

Deploys. Changing the sketch’s shape or hash seed makes old state impossible to merge. So the new version replays a day of Kafka in a parallel job before the merger switches to it. Adding Kafka partitions moves videos between aggregators mid-window, so it is a new topic, run side by side for a day.

Cost is Kafka brokers and the batch’s daily scan of 1 TB, not the sketches.

Wrap up

The whole design

Here is everything we built, in the order we built it. Use the buttons to step through it.

Top-k heavy hitters · architecture

The full design, stage by stage

Color keyServiceDatabaseCacheQueue or streamBackground workerNew in this stageAsync or occasional

Stage

Events per second: 350,000

Stage 6 of 6

One pipeline per region

Why this stage

Viewers are on three continents, one region is one failure domain, and at 10 million events/s one region is past what it can take.

Raw events should not cross oceans.

What we add

The ingest, Kafka and aggregator path in every region.

A global merger combines regional sketches and candidates in two phases.

What it costs

An item is now split across regions, so merging top lists is no longer exact without the second phase; the global list lags a few seconds more.

At 350k events per second, every part has headroom. Raise the traffic until something turns red.

Pinch or tap the zoom buttons to read small labels.Drag any box to move it. Reset puts every box back.The whole design. Step through it: one counting server, a split by video, sliding windows on sketches, exact daily lists, combining views at ingest, then one pipeline per region.

The numbers behind the busy bars

What one machine can do

One counting process handles about 100,000 events a second.
One ingest server takes about 50,000 events a second, because clients send batches.
The Kafka cluster, as provisioned, takes about 1.5 million messages a second.
One API server answers about 5,000 reads a second.

The hot item

The aggregator bar shows the busiest instance: the one holding a viral item at 15% of all views.

What saves work

Pre-aggregation cuts messages to about a quarter.
The largest region takes about 40% of traffic.

The decisions, and what each one costs

Decision	We chose	We gave up	We’d choose differently if
Splitting	By video	Even load under a viral video	Counts were per user (split by user, two-phase merge)
Live counting	Count-Min Sketch + candidate heap	Exact live counts	Distinct videos were few (exact maps)
Windows	Hopping by the minute, event time, 30 s watermark	30 s of freshness	Trending needed no edge at all (exponential decay)
Hot videos	Combine at ingest for 1 s	One second; duplicate removal live	Lists had to be under a second old (salting, two phases)
Regions	Regional pipelines, two-phase global merge	A few seconds on global lists	One region served everyone
Exact daily	A batch for the official chart only	A single code path	Exact counts were needed live, with affordable state

How to spend 45 minutes

0 to 4. Windows, k, freshness, exact or approximate. Accuracy as a number.
4 to 8. 350,000 views a second at peak, 88 GB exact against 4.7 GB of sketches.
8 to 18. Build it: split by video and why the merge is exact, sketches for the windows, the daily batch. A mid-level answer can stop here.
18 to 30. The stress tests: the sketch’s error, then late views, then a viral video.
30 to 40. Regions and the merge, then stream against batch.
40 to 45. An aggregator crash, then your own questions.

Never skip splitting by video, the error bound, or why merging split top lists can be wrong.

Follow-up questions

Interviewer asksNow product wants the top 100 per country, for 200 countries. What changes?

A full-size sketch per country multiplies state by 200, and most countries are small. One shared sketch keyed by country and video makes every error relative to the global total, too coarse for a small country. Better: size each country’s sketch by its traffic, or keep a Space-Saving summary per country.

Interviewer asksHow do you know the sketch is still sized right six months from now?

The morning comparison measures it: the batch’s exact counts give the true error of every published estimate. Alert when the error for top videos nears the bound, which happens as traffic grows. Widening is a deploy with fresh state, at a day boundary.

Go deeperMore follow-up questions1 min

A bot farm pushes one video into the top 10. How do you notice and stop it? Signals: views from few accounts, new devices, one network, or watch times near zero. The cap per user removes repeats, and the batch cleans the next morning’s chart. For the live list, the block list at publish time removes it within a minute.

How would typeahead’s trending queries use this? The same pipeline, keyed by the normalized query, with a short window. Typeahead ranks by the rise against a baseline, so it adds a ratio on top of these counts. See the typeahead page.

Why not a Redis sorted set with ZINCRBY on every view? That is 350,000 writes a second, with the hottest videos on one machine. And forgetting old minutes means a sorted set per minute, and a union of 1,440 for the day. Counting in memory turns 350,000 writes a second into a few a minute.

Design top-k heavy hitters

The answer in one minute

Scope it

The numbers that decide it

Build the design

Step 1. One counting server

Stage 1: One counting server

Why this stage

What we add

What it costs

Step 2. Split the counting by video

Stage 2: Partition by item

Why this stage

What we add

What it costs

Step 3. Count the windows with a sketch

Stage 3: Sliding windows on sketches

Why this stage

What we add

What it costs

Step 4. An exact batch for the official chart

Stage 4: Exact daily lists

Why this stage

What we add

What it costs

Trace a request: Read the hourly top 100

Stress it

Exact counts or a sketch

Windows, and views that arrive late

One viral video

Stage 5: Pre-aggregate at ingest

Why this stage

What we add

What it costs

Regions, and merging split lists

Stage 6: One pipeline per region

Why this stage

What we add

What it costs

The batch beside the stream

When things break

Break it: what fails, who notices, what users see

What breaks

How we notice

What users see

What we do

Wrap up

The whole design

The full design, stage by stage

Why this stage

What we add

What it costs

The decisions, and what each one costs

How to spend 45 minutes

Follow-up questions

Further reading

Keep reading

The designbeyondcode digest, every Saturday