Design Trending Topics and Top-K Requests

This is lesson 5 of System Design Case Studies, Part 2. The key-value store in Lesson 1 gave us partitioning by key hash. Here we count events in a stream and combine counts from many partitions. A later case, Lesson 9 (sales ranking by category), asks a similar "rank things" question over a smaller, more exact data set, and Lesson 10 covers the pipeline that feeds streams like this one.

Words used in this lesson

• Stream is data that keeps arriving (one event per search, per view, per like). You cannot stop it and read it twice.
• Top-K (also heavy hitters) are the K items with the largest counts.
• Window is the time range you count in, such as "the last hour". A tumbling window is a fixed block (10:00 to 11:00, then 11:00 to 12:00). A sliding window moves with time ("the last 60 minutes until now").
• Exact counting keeps one counter per distinct item. Approximate counting uses less memory and accepts a small, known error.
• Count-min sketch is a table of counters with d rows and w columns. Each row has its own hash function. To add an item, add 1 to one cell in every row. To estimate an item, read its cell in every row and take the minimum.
• Collision is when two different items hash to the same cell. It only ever makes a count too high, never too low. That is why taking the minimum works.
• Min-heap is a tree where the smallest value is at the top. A heap of size K whose top is the K-th largest count lets you ask "does this item beat the current K-th?" in one step.
• Shard (partition) is one machine's share of the data. Here we split by a hash of the key.
• Hot key is one key that gets so many events that its shard is overloaded.
• Decay means older events count less. Half-life is the time after which an event's weight is halved.
• Zipf distribution is a popularity pattern where item number k gets a share proportional to 1/k^s. Search queries and hashtags often look roughly like this (an assumption; check on real data).

1. Requirements

Functional

• Ingest a stream of events (a search, a hashtag use, a video view).
• Return the top K items (say K = 10 to 100) for a window (last 5 minutes, last hour, last day) and for a scope (global, country, language).
• Show "trending" items: those that rise fast, not only those that are always big.
• Optional: filter out spam and banned terms, and group variants (#Cup, #cup).

Non-functional

• Fresh: a new burst should show in the list within about a minute (assume).
• Fast reads: the top-K list is read much more often than it changes. Serve from a cache in single-digit milliseconds.
• Scalable writes: peak is hundreds of thousands of events per second.
• Bounded memory even if the number of distinct items is in the hundreds of millions.
• Approximate is acceptable for the long tail, but the top of the list should be right, and the error should be known.
• Fault tolerant: losing one machine must not lose the ranking.

Out of scope here: natural-language grouping of similar topics, and moderation policy.

2. Back-of-envelope numbers

Every number is an assumption.

Events per day            5 billion              (hashtag uses + queries)Average rate              5e9 / 86,400 = ~58,000 events/sPeak rate                 3 x average = ~175,000 events/sEvent size on the wire    100 bytes  -> 175,000 x 100 B = 17.5 MB/s at peak  (tiny)Distinct items per hour   50 million (assume most appear once or twice)Exact counting, 1 hour:   50M items x (~10 B key + 8 B counter) = 0.9 GB   (real hash tables add overhead,                                                                           so plan for 2 to 3x that)Many scopes: 20 regions x 5 windows x several languages, each needing its own table, so exact tables multiplyCount-min sketch, one bucket:  w = 2,000 columns x d = 5 rows x 4 B = 40 KB60 one-minute sketches (1-hour sliding window) = 60 x 40 KB = 2.4 MB per scope24 h of 1-minute buckets = 1,440 x 40 KB = 56 MB per scope                 (fits anywhere)

What this arithmetic shows:

• The network and CPU load is small: 175,000 events/s is easy to split over a few dozen machines.
• The hard part is memory over many scopes and windows. One exact table for one hour is fine (under 3 GB). But 20 regions x 5 windows x many languages, and long histories, grow fast. A sketch makes each scope cost kilobytes.
• Also say plainly: if the whole stream fits on one box with exact counters, use exact counters. They are simpler, and there is no error to explain. The sketch is for when it does not fit.

3. API

# Ingest (internal; usually a message queue, not HTTP)POST /v1/events[ { "item": "#worldcup", "ts": 1760000000, "region": "IN", "lang": "hi" }, ... ]    # a batch→ 202 Accepted# ReadGET /v1/trending?region=IN&window=1h&k=10→ 200 OK{ "window": "1h", "asOf": "2026-10-11T10:04:00Z",  "items": [ { "item": "#worldcup", "count": 182340, "approx": true, "error": "+/- 120",               "trend": 3.4 }, ... ] }# AdminPOST /v1/blocklist   { "item": "#spamtag" }

Notes:

• Writes should be batched and go through a queue. The ingest endpoint only answers "accepted".
• The response says approx: true and gives an error bound. An honest API tells the reader that counts are estimates.
• trend is a ratio (current rate divided by the usual rate; see section 6.6).

4. Data model

Queue (Kafka-style):   topic "events", partitioned by hash(item)      -> one item always goes to one partitionWorker state (memory): per partition, per minute:   CountMin(w, d) + small TopK heap of candidate itemsSnapshots (object store, every minute):  sketch tables + candidate lists, so a restarted worker reloads themServing store (Redis sorted set):  key "trend:{region}:{window}"  member=item  score=count                                   refreshed every 10-60 s by a merger job; clients read ZREVRANGE 0..K-1

A Redis sorted set (ZSET) keeps members ordered by a numeric score. Reading the top 10 is one call. The read path never touches the sketches.

5. High-level architecture

 Clients ---> API/ingest ---> Queue: topic "events", key = hash(item), P partitions                                         |                    +--------------------+--------------------+                    v                    v                    v              Worker 1             Worker 2    ...       Worker P         (each owns some partitions)        per-minute CountMin    per-minute CountMin    per-minute CountMin        + candidate heap       + candidate heap       + candidate heap                    |                    |                    |                    +----- every 10-60 s: local top-K candidates (and counts for the window) ----+                                         v                                   Merger job              (sum bucket counts over the window, apply decay, take the global top-K)                                         v                              Redis ZSET  trend:{region}:{window}                                         v                      Trending API (cache, CDN for 10-30 s)  <--- readers

Because the queue is partitioned by the item, all events for one item reach one worker. So each worker's counts for its items are complete, and the global top-K is simply the top-K of the workers' top-K lists (proof in 6.4). The sketch is only needed because each worker still sees millions of distinct items.

6. Deep dives

6.1 Exact versus approximate

The honest decision rule: use an exact map until it stops fitting. Then use a sketch for counting and a small heap for listing.

6.2 The count-min sketch and the min-heap

Here are the two data structures I ran (sketch.ts, shortened where marked):

export class CountMin {  w: number; d: number; table: Uint32Array; total = 0;  constructor(w: number, d: number) { this.w = w; this.d = d; this.table = new Uint32Array(w * d); }  add(key: string, n = 1) {                       // +n in one cell of every row    for (let r = 0; r < this.d; r++) this.table[r * this.w + (hash32(key, r * 0x9e3779b1 + 1) % this.w)] += n;    this.total += n;  }  estimate(key: string): number {                 // the minimum over the rows    let m = 0xffffffff;    for (let r = 0; r < this.d; r++) { const v = this.table[r * this.w + (hash32(key, r * 0x9e3779b1 + 1) % this.w)]; if (v < m) m = v; }    return m;  }  merge(o: CountMin) {                            // same w, d and hashes: add cell by cell    for (let i = 0; i < this.table.length; i++) this.table[i] += o.table[i];    this.total += o.total;  }}// TopK (not shown in full): a min-heap of {key, count} with a Map<key, position>,// offer(key, count): if the key is in the heap, update it; else if the heap has room, add it;//                    else if count > heap top (the smallest of the K), replace the top.

For every event the worker does: cm.add(key); heap.offer(key, cm.estimate(key)). Cost per event is d hash computations plus O(log K) for the heap. With d = 5 that is tiny.

Error. Let N be the total number of events in the sketch. The rule I remember: if w = e/ε columns and d = ln(1/δ) rows, then with probability at least 1 - δ the estimate is at most true + ε x N. I measured it on 1 million events (88,109 distinct items), d = 5:

   width | memory  | bound e/w*N | max over-count (all keys) | mean over-count, top-10 | mean over-count, 1000 rare keys | top-10 correct     500 |   10 KB |        5437 |                      2718 |                   626.5 |                           595.4 | 10/10    2000 |   39 KB |        1359 |                       728 |                   118.2 |                           102.3 | 10/10    8000 |  156 KB |         340 |                       153 |                    18.1 |                            15.7 | 10/10   32000 |  625 KB |          85 |                        22 |                     2.5 |                             1.5 | 10/10

Read it carefully:

• The worst error stayed under the bound in all four runs, and went down as the table grew (quadruple the width, about 4 times smaller error).
• The error is about the same for every key (118 for the top-10, 102 for rare keys at width 2,000). It depends on how crowded the table is (N / w), not on how big the key is.
• So the relative error is tiny for heavy keys (118 on 130,823 is 0.09%) and huge for rare ones (a rare key with a true count of 1 or 2 reads as about 100). A sketch is good at finding the big ones and bad at counting the small ones. That is exactly what top-K needs, and it is why you must not use it to show a count for a rare item.

The top-10 from the 39 KB sketch with the heap, against the exact counts:

   rank key       estimate exact  error    1   t0         131017 130823   194    2   t1          61054  60959    95    3   t2          39056  38961    95    4   t3          28701  28615    86    5   t4          22353  22240   113    ...   10   t9          10451  10340   111   exact map needs at least 88109 x (about 10 B key + 8 B count) = 1.6 MB; the sketch is 39 KB

The order is right, and the counts are about 0.1% high. In this test an exact map would be 1.6 MB, which is small. A sketch only wins when the distinct count is much larger (50 million items in section 2 means about 0.9 GB or more) or when you keep thousands of scopes and buckets.

A common improvement (not tested here). Conservative update only raises the cells that are currently the smallest for the item, which lowers over-counting. Space-Saving is a different algorithm that keeps exactly K counters and needs no hash table at all; I did not re-read its paper.

6.3 Sliding windows: ring of buckets

A single sketch cannot forget. So split time into buckets (one per minute), keep one sketch per bucket, and the window "last 60 minutes" is the sum of the last 60 sketches. When a minute ends, the oldest bucket is reused (cleared). I ran a 180-minute scenario: topic Y is huge in minutes 0 to 59 and then stops, X is busy in minutes 100 to 129, Z starts at minute 165, and 1,000 background events per minute come from 5,000 items.

at minute 179all-time counts       : Y=180000  X=45000  bg0=19870last 60 min (sketches): X=15000  Z=12000  bg0=6678last 60 min (exact)    : X=15000  Z=12000  bg0=6678

The all-time list is dominated by a topic that died two hours ago. The windowed list is right, and the 60 sketches (one per minute, 1,000 x 4 x 4 B = 15.6 KB each) match the exact answer. Memory is 938 KB for this scope.

How a query works: a sketch cannot list its keys, so each bucket also keeps a candidate list (a heap of its top 20). The query takes the union of the candidates, and for each candidate adds its estimate over the 60 buckets. A key that is not a candidate in any bucket but is moderately big in all of them can be missed (the same problem as in 6.4).

Trade-offs of bucket size: smaller buckets (10 s) give a smoother window edge and fresher results but more buckets to sum. A one-hour window with one-minute buckets has an edge error of up to one minute. The reads are cheaper if you pre-add: keep a running sum, add the new bucket, subtract the expiring one (this works because sketches are linear).

6.4 Partitioning and merging the top-K

How should the stream be split over machines? Two choices, and the choice changes the merging rule.

(a) Partition by the item (hash of the key, as in Lesson 1). All events for an item go to one worker, so the worker's count for it is complete. Then the global top-K is contained in the union of the local top-K lists: if an item is in the global top K, it cannot be outranked by K items on its own shard (they would also outrank it globally), so it is in its shard's local top K. Merge = take the top K of the union. It is exact (given exact local counts). I tested this with 4 shards:

   (a) by key hash, local top-10 merged:   10/10 of the true top-10, counts exact

(b) Spread events round-robin (or by user, or by time). Each worker sees only part of each item's count. The local top-K lists can now disagree. Two facts:

• Sketches merge exactly. Four sketches built on four slices of the stream, added cell by cell, are identical to one sketch built on the whole stream. I checked every cell: same? true. This is because the table is just a sum.
• Local top-K lists do not merge exactly. An item that is 11th on every shard can be 1st globally. I tested this with flat popularity (Zipf exponent 0.15, 8 shards, round-robin):

3b. Flat popularity (zipf s=0.15, 5,000 keys, 200,000 events), 8 shards, events spread round-robin   each shard sends its local top-10  -> 6/10 of true top-10 found, reported counts too low by 55 on average   each shard sends its local top-20  -> 8/10 of true top-10 found, reported counts too low by 48 on average   each shard sends its local top-50  -> 8/10 of true top-10 found, reported counts too low by 30 on average   each shard sends its local top-200 -> 10/10 of true top-10 found, reported counts too low by 9 on average

With the skewed Zipf stream in the earlier test (exponent 1.1) the round-robin merge found 10 of 10 even with local top-10 lists, because the big items are far ahead of the rest. With flat popularity it found only 6. The lesson: local top-K merging is safe only if the data is skewed or the partition is by key. If you must partition another way, either send bigger local lists (K x 10 or more), or send the sketches and build the candidate list from the union of all local candidates, then query every candidate in the merged sketch (a second pass).

6.5 Hot keys

Partitioning by item has one weak point: a single item that is a large share of all traffic lands on one worker. I simulated a key that gets 30% of the events, over 8 shards:

   hot key split in 1: busiest shard 3.69x the mean, quietest 0.33x   hot key split in 8: busiest shard 1.43x the mean, quietest 0.57x

The fix is to split the hot key into sub-keys (HOT#0 to HOT#7) by a random suffix on the way in. Each sub-key goes to a different worker. The merger adds the counts of all sub-keys back together (strip the suffix). The busiest shard went from 3.69x to 1.43x the average load.

How do you know which key is hot? Two ways: a first-stage local pre-aggregation at the ingest layer (each ingest server counts for 1 second, then sends (item, count) pairs, which also reduces queue traffic a lot), or a small detector (a heap on a sketch) that flags items over some share of traffic and switches on key-splitting for them. Pre-aggregation is the first thing I would add.

6.6 Trending is not the same as top

Raw counts say what is big (the word "the", a news site that is always popular). Trending means what is rising. Two common scores (my wording of common practice, not a specific product's formula):

• Ratio: score = (count in the last 10 minutes) / (expected count for 10 minutes), where expected comes from the same hour in past weeks or a long-window average. Add a small constant to the denominator so a brand-new term does not divide by zero. Also require a minimum count so a term that went from 1 to 5 is not "5x trending".
• Decay: keep one score per item, score = score x 0.5^(elapsed / half-life) + 1 for each event. A burst scores high and then fades by itself, and you need only one number and one timestamp per item, not buckets.

I ran the decay on the same 180-minute scenario with a 15-minute half-life:

decayed score (half-life 15 min): Z=8858  X=2472  bg0=2445

The windowed count put X (15,000 in the last hour, but all of it more than 50 minutes ago) first and Z (12,000, rising now) second. With decay, Z is first by a wide margin (8,858 against 2,472), because its events are recent. Which is "right" depends on the product. Decay is smooth and cheap. A window is easy to explain.

Decay inside a sketch: one way is to multiply the whole table by a factor every minute (cost: touch every cell). A cheaper trick (from memory, "forward decay"): give each event a weight that grows with time (e^(lambda x t)), add that to the sketch, and divide by the current factor at read time. Weights are stored in floating point, so you must re-scale now and then. I did not run this variant.

6.7 Spam, bots and one noisy user

A top-K system is a target: a botnet can push a hashtag into the list. Basic defences: count unique users per item (each user counts once per window) rather than raw events, which needs a distinct-count sketch (HyperLogLog is the usual tool; not tested here); rate-limit per user and per IP; a blocklist applied before counting; and a human review queue for items that enter the list from nothing very fast.

6.8 Serving reads

The merger writes the final list into a Redis sorted set (or just a JSON blob) per scope and window, every 10 to 60 seconds. API servers cache it for a few seconds, and a CDN (Lesson 4) can cache it for 10 to 30 seconds. Readers see data that is at most a minute old, and read cost is independent of event volume. Keep the previous list: if the merger fails, serve the old one with a stale marker rather than an empty page.

7. Failure modes and trade-offs

Trade-offs to say aloud

• Exact vs approximate: memory and merge cost against explaining an error. Start exact.
• Window vs decay: easy to explain against smooth and cheap.
• Freshness vs cost: a 10-second merge is fresher and costs 6 times more merging than a 60-second one.
• Partition by item vs spread: by item gives exact merging but hot keys; spreading fixes hot keys and breaks local top-K merging.
• Bucket size: small buckets are fresh and costly; big buckets are cheap and blurry.

8. How to say it in an interview (60 seconds)

"Events go to a partitioned queue keyed by the item, so one worker sees all events for an item. Each worker keeps one count-min sketch per minute plus a small heap of candidate items. A sketch is a d x w table: add one cell per row, estimate with the minimum over rows, so it never undercounts. With 40 KB and one million events, my test showed errors of about 0.1% on the top items and the right top 10, though rare items read about 100 too high, so a sketch finds big items and does not count small ones. The window is the sum of the last 60 minute-buckets; I tested it against exact counts and it matched. Because we partition by item, the global top-K is the top-K of the local top-K lists, exact. If we partition differently, sketches still add cell by cell, but local lists can miss items on flat data (6 of 10 in my test), so send bigger lists. For a hot key I split it into sub-keys: busiest shard from 3.7x to 1.4x the mean. For trending I rank by rise over the usual rate or by an exponential decay score, and a merger writes the result into a Redis sorted set that the API caches. If the data fits in exact counters, I would use those."

Run it yourself

Files: sketch.ts (classes), topk.ts (about 5 seconds) and window.ts. Run node topk.ts and node window.ts in one folder. They import with ./sketch.ts and use only Node's standard library. Constructor parameter properties are avoided because Node's type stripping does not support them.

Interview questions

Key takeaways

• Do the arithmetic first. If exact counters fit, use them. A sketch is for when distinct items or the number of scopes make exact counting too big.
• A count-min sketch is a d x w table: add to one cell per row, read the minimum. It never undercounts. Measured: 39 KB gave about 0.1% error on the top items and the right top 10, while rare items read about 100 too high.
• A sketch cannot list keys. Pair it with a min-heap of size K.
• Sliding window: one sketch per time bucket, sum the last N. Sketches add cell by cell, exactly.
• Partition by item and the global top-K is the top-K of local top-K lists. Spread events, and you need bigger lists (6 of 10 found with top-10 lists on flat data, 10 of 10 with top-200).
• Hot key: pre-aggregate and split into sub-keys. Measured: 3.69x to 1.43x the mean load.
• Trending means rising, not big: use ratio to baseline or exponential decay.

What's next

Lesson 6: Design a Unique ID Generator. The queue, the shards and the buckets in this lesson all need names that never repeat. That lesson shows how to make 64-bit IDs that are unique across machines and sort by time.