Design a Top-K System (Heavy Hitters): Count-Min Sketch, Heaps and Windows

One view, from a click to the top-10 list

Follow a single view of video v42 through both paths. This is a good order to explain it in.

1. 1

   Recorded

   The player sends a view event: video v42, time 10:07:31, region eu. The ingest service writes it to the Kafka topic views, with the key v42.

2. 2

   Routed

   Kafka hashes the key, and the hash picks one partition. Every view of v42 lands in that partition and is read by one worker.

3. 3

   Counted

   The worker adds 1 to one counter in each row of the sketch for the current 5-minute slice, then reads the hour's estimate for v42: its estimate in each of the last 12 slice sketches, added together.

4. 4

   Ranked

   If v42 is already in the worker's heap, its count is updated. If not, and its estimate beats the smallest count in the heap, v42 replaces that video.

5. 5

   Merged

   Every few seconds, each worker sends its top 1,000 with counts. The merge step keeps the largest 1,000 overall. This is exact for the estimates, because no video's views are split across workers.

6. 6

   Served

   The merged list goes into Redis under a key such as top:1h. GET /top?window=1h&k=10 reads the first 10 entries.

7. 7

   Archived

   This event is also written to object storage, in a folder for its hour.

8. 8

   Recounted

   After the hour closes and late events have had time to arrive, a batch job counts every view of that hour exactly and writes the exact top 1,000.

9. 9

   Corrected

   The exact hourly counts replace the estimates. Daily and all-time totals are built by adding exact hourly counts per video, never by adding hourly top-k lists.

These are the conversations the interviewer is steering you toward. Practice each one until you can talk through it without notes.

The two paths: fast estimates and a slow exact recount

Two needs pull against each other. Viewers want the trending list to move within a minute. The business wants the numbers to be right, because people may be paid by views or rank. So build two paths. The fast path counts in memory as events arrive and accepts a small, stated error. The slow path stores every event and later counts each closed hour exactly, then overwrites the estimate. This is the lambda architecture. Nathan Marz, who named it, put it this way: the realtime layer only covers the last few hours of data, and everything it computes is later replaced by the batch layer. A bug in the fast path therefore fixes itself within the hour. The cost is two code paths that must agree on what counts as a view. If that worries the interviewer, mention the kappa architecture: one stream path, and to recount you replay the stored log through a fixed version of the stream code. For a top-k system the lambda shape is common because the fast path is approximate on purpose and the slow path is exact on purpose. Late events need a rule on both paths. A view that arrives 3 minutes late must count toward the hour in which it happened. The batch job waits a short grace period before it closes an hour. Watermarks, which tell a stream job how far event time has progressed, are covered in depth in the ad click aggregator walkthrough on this site.

Exact counting first, and where it gets expensive

Always start with the simple answer, then show where it hurts. The simple answer is a hash map from video id to count, plus a sort. Sorting every video on every request is too slow, so keep a min-heap of the top k next to the map instead. With the views partitioned by video id across 14 workers, the exact map for one day is about 32 GB in total, roughly 2.3 GB per worker. That is fine for one day. The trouble comes from multiplication. Each window needs its own counts: last hour, last day, all time. A sliding hour needs one set of counts per slice. A breakdown such as 'top videos per country' multiplies by the number of countries. And most of the memory goes to the long tail: hundreds of millions of videos with a few views each, which will never be in any top list. A count-min sketch fixes exactly this. Its memory is fixed by the error you choose, not by the number of different videos. Say it plainly in the interview: if one exact map per window fits, use it, because exact is simpler to explain to users. Reach for the sketch when windows, slices and breakdowns multiply the memory.

The count-min sketch and its error bound

The count-min sketch comes from a 2005 paper by Graham Cormode and S. Muthukrishnan. It is a table of counters with d rows and w columns, all starting at zero. Each row has its own hash function, which turns a video id into a column number. To add a view of video i, go to each row, find the column that row's hash gives for i, and add 1 there. To read the count of video i, look at those d counters and take the smallest. Other videos can land in that counter too (a collision), and their views only ever add to it, so every counter is at least the true count. The smallest one is the best guess, and it is never too low. You choose two numbers. ε (epsilon) is how large an error you accept, as a fraction of all views. δ (delta) is the chance you accept of going over that error. The paper sets the width w = e / ε, rounded up, where e is about 2.718, and the depth d = ln(1 / δ), rounded up, where ln is the natural logarithm. Its Theorem 1 then says: the estimate is at least the true count, and with probability at least 1 - δ it is at most the true count plus ε times the total number of views. Two cautions. First, the error is relative to the total of all views, not to the video's own count. With ε = 0.001 on a million views, any video may be up to 1,000 too high. That is nothing for a video with 80,000 views and everything for one with 50. So a sketch is good at the top of the list and useless for the tail. A top-k system only needs the top. Redis's own documentation says the same: results under error times total count should be treated as noise. Second, libraries size the table with their own constants. In Redis, CMS.INITBYPROB with error 0.001 and probability 0.002 creates a sketch 2,000 wide and 9 deep, while the paper's formulas would give 2,719 and 7. Read the library's sizing before you quote a bound for it. Two sketches of equal size that use identical hash functions can be added cell by cell, and the result is the sketch of both streams together (Redis offers this as CMS.MERGE). That is what makes slices and partial counts cheap to combine. The full Python is short:

import math, random
P = 2_147_483_647  # a prime larger than any item id

class CountMin:
    def __init__(self, eps, delta, seed=1):
        self.w = math.ceil(math.e / eps)          # width  = e / epsilon
        self.d = math.ceil(math.log(1 / delta))   # depth  = ln(1 / delta)
        r = random.Random(seed)
        self.ab = [(r.randrange(1, P), r.randrange(P)) for _ in range(self.d)]
        self.rows = [[0] * self.w for _ in range(self.d)]

    def _col(self, j, x):
        a, b = self.ab[j]
        return ((a * x + b) % P) % self.w

    def add(self, x, c=1):                        # one counter per row goes up
        for j in range(self.d):
            self.rows[j][self._col(j, x)] += c

    def query(self, x):                           # smallest of the d counters
        return min(self.rows[j][self._col(j, x)] for j in range(self.d))

Why a sketch alone cannot give you the top k

A count-min sketch answers one kind of question: about how many times did this video appear? It cannot answer the question you actually have: which videos appeared most? It stores counters, not video ids, so there is nothing to list. The paper itself solves this with a heap. As each view arrives, add it to the sketch and read back that video's estimate. Then offer the video to a min-heap that holds at most k videos. A min-heap is a small tree-shaped structure that always keeps its smallest item at the top, so checking the smallest costs one step and replacing it costs about log k steps. If the video is already in the heap, update its count. If the heap has room, add it. Otherwise, if the new estimate is larger than the smallest count in the heap, the new video replaces it. With k = 1,000 the heap is about 16 KB. One subtle point: the heap only ever sees videos at the moment they get a view. A video's true count only changes when one of its own views arrives, and that is exactly when it is offered to the heap again, so a growing video is always checked at the moment it grows. The paper proves the matching result for frequent items: every item above a threshold φ (phi, a fraction you choose) times the total is reported, and with probability 1 - δ no item below (φ - ε) times the total is. Redis ships this pairing as one command family. Its Top-K structure uses the HeavyKeeper algorithm, with a table of decaying counters plus a min-heap of the k leaders, and TOPK.ADD tells you which item was pushed out of the list.

import heapq

def offer(heap, k, video, est):
    """heap is a min-heap of (estimate, video) with at most k entries."""
    for i, (_, v) in enumerate(heap):
        if v == video:                  # already in the top k: update its count
            heap[i] = (est, video)
            heapq.heapify(heap)
            return
    if len(heap) < k:
        heapq.heappush(heap, (est, video))
    elif est > heap[0][0]:              # beats the smallest of the k
        heapq.heapreplace(heap, (est, video))

def top_k(events, k, sketch):
    heap = []
    for v in events:
        sketch.add(v)
        offer(heap, k, v, sketch.query(v))
    return sorted(heap, reverse=True)

Windows: tumbling, sliding, and 5-minute slices

A window is the stretch of time a count covers. The two kinds you must know have different costs. A tumbling window has a fixed size and never overlaps the next one: 9:00 to 10:00, then 10:00 to 11:00. Apache Flink's documentation defines it that way, and each event belongs to exactly one window. It is cheap, but the 'last hour' list jumps once an hour and is out of date near the end. A sliding window also has a fixed size but starts a new window every slide interval, for example one hour long every 15 minutes, so the windows overlap and one view belongs to several of them. Kafka Streams calls this kind a hopping window and keeps the name sliding for windows defined by the time difference between two records. Name both so the interviewer knows you know. The cost is real. Flink's documentation warns that a window of one day that slides every second is not a good idea, because each event is copied into every overlapping window. The fix is slices. Count each 5-minute slice once, in its own small sketch. The last hour is then the sum of the latest 12 slice sketches, which works because sketches add cell by cell. Every 5 minutes, add the newest slice and drop the oldest. The heap needs care here: when a slice is dropped, counts go down, and a heap that only ever saw increases is now out of date. So at every slide, rebuild the hour's top list from candidates: every video named in any slice's top 1,000, plus the current leaders, each scored with the summed sketch. That candidate set can miss a video that was never in any slice's list, which is the merging problem in the next section, and its check applies here too. In Flink, prefer an incremental aggregate, which keeps one running value per window, over a function that buffers every event of the window: the documentation says the buffering kind must keep all elements.

Merging partial top-k lists across partitions

This is the trap most candidates walk into. There are two cases, and they behave differently. Case one: the views are partitioned by video id, so each video's views all land on one worker. Then every video in the global top k is also in its own worker's local top-k list, and merging those lists is exact, as far as the estimates are exact. That is why the design keys Kafka by video id. Case two: a video's views are split across the lists you merge. This happens when partitions are by region or by server, when a hot video is salted across workers (next section), and, most often, when you build a day out of 24 hourly top lists. Now merging lists can lose the true winner. The figure shows it with real numbers: three regions each send their top 2, and video D, third in every region with 45 views, never appears, yet its total of 135 beats everything that was sent. A video that is 11th every hour can be first for the day and appear in no hourly top 10. There are three fixes. The best is to merge counts, not lists: add exact hourly counts per video, or add sketches cell by cell, and only then pick the top k. If you must merge lists, ask for longer lists (k' larger than k, here 1,000 for a top 10), then do a second round: ask every partition for the exact count of every video named in any list. Finally check the bound. A video missing from a partition's list has, in that partition, at most the smallest count on that list. Add those smallest counts across partitions; that sum is the most any unlisted video could have. If your k-th answer beats it, the answer is proven. If not, ask for longer lists and repeat. One practical note for Flink: a non-keyed window runs as a single task, so do the per-video counting in keyed windows across many tasks and keep only the small final merge of 1,000-entry lists in the single task.

Hot partitions: one video, a fifth of all views

Keying by video id spreads load well when no single video is huge. A viral video breaks that. Suppose one video takes 20% of 350,000 views per second, which is 70,000 per second, on top of the roughly 35,000 that each of 8 workers already handles. The worker that owns it gets 105,000 per second against a limit of 50,000. Its queue of unread events grows, its counts fall behind, and its part of the top list goes stale exactly when people care most. Adding workers does not help, because one key always hashes to one place. Two fixes, best used together. First, pre-aggregate before the network: the ingest service counts views per video for one second and sends (video, count) pairs. A video with 70,000 views per second then costs one message per ingest machine per second. Second, salt the hot keys: add a random number from 0 to 7 to the key of a video that is hot right now, so its events spread over 8 partitions. Each partial count is then one piece of that video's total, and a second small step adds the pieces back together before ranking. That is case two from the merging section, so the merge must add counts, not lists. Only salt the few videos that need it. You can find them cheaply with the sketch itself: any video whose estimated rate crosses a threshold gets salted for the next few minutes.

Other heavy-hitter algorithms: Misra-Gries and Space-Saving

The count-min sketch plus a heap is not the only answer, and naming the others shows depth. Misra-Gries (1982) keeps k - 1 counters. A tracked item gets its counter raised. A new item takes a free counter if one exists. If none is free, every counter goes down by 1 and counters that reach zero are freed. Every item that appears more than n/k times in a stream of n items is guaranteed to still hold a counter at the end. Space-Saving, by Metwally, Agrawal and El Abbadi (2005), changes one step: when the counters are full, the new item takes over the counter with the smallest count and adds 1 to it, remembering the old value as its possible over-count. With m counters, the smallest counter is at most n/m, so no count is too high by more than n/m, and any item that appears more than that smallest count is guaranteed to be tracked. Both keep the item ids themselves, so they give the top list directly with no separate heap. Both can also be merged: Apache DataSketches ships a Frequent Items sketch based on Misra-Gries that merges across machines. Its documentation states the gap between the upper and lower bound as at most the total count times 3.5 / M, where M is the map size. In its own test, 10 million items with M = 2,048 gave a worst case of 17,090 and an actual largest error of about 3,000. In an interview, a fair summary is: count-min plus heap if you also need a count for any video on request, Space-Saving if you only need the top list.

How a Top-K System actually does it

Every method on this page is published and in use. The count-min sketch is defined in Cormode and Muthukrishnan's paper (Journal of Algorithms, 2005): width e/ε, depth ln(1/δ), and an estimate that is never below the truth and, with probability at least 1 - δ, at most ε times the total too high. That paper also finds heavy hitters by pairing the sketch with a heap. Redis offers both pieces. Its count-min sketch commands size a table from an error and a probability; with 0.001 and 0.002 it builds a table 2,000 wide and 9 deep. Its Top-K structure is based on the HeavyKeeper algorithm (USENIX ATC 2018), which decays the counters of small items, and it keeps the leaders in a min-heap. Misra and Gries published their counter-based method in 1982, and Metwally, Agrawal and El Abbadi published Space-Saving in 2005, where no count is more than n/m too high with m counters. Apache DataSketches ships a mergeable Frequent Items sketch built on Misra-Gries, with a documented bound of 3.5/M times the total count. For windows, Apache Flink defines tumbling and sliding windows, runs non-keyed windows on a single task, drops late events by default, and warns against tiny slides on long windows. Kafka Streams calls overlapping fixed windows hopping windows. The fast-plus-batch layout is Nathan Marz's lambda architecture (2011), in which the batch layer later replaces whatever the realtime layer computed. Across all of them, one lesson repeats: approximate counting is safe at the top of a list where a few items get most of the events and unreliable in the tail, so build the system so that the error lands where nobody looks.

Frequently asked questions