Design Top-K / Trending Topics (Heavy Hitters)

"Design trending hashtags for Twitter/X", "find the top 10 most viewed products on Flipkart in the last hour", "show the top searched queries right now". These are all the same problem: heavy hitters in a stream. It is popular in interviews because it lets the interviewer move smoothly from a LeetCode-style warm-up (top K frequent elements) to probabilistic data structures to distributed stream processing, and see how far you can go.

This guide follows that same ladder: start exact and simple, see where it breaks, and add exactly as much machinery as the numbers require.

Popular vs Trending: Pin Down the Question

The first two minutes decide whether your design answers the right question.

Also ask about scope (global, per country, per city), freshness (updated every few seconds or every few minutes), and accuracy (is an approximate rank acceptable? For trends, almost always yes. For billing, as in ad click aggregation, no).

Requirements

Functional

• Ingest hashtag events from posts (and optionally searches).
• Return top-K (K ≤ 50) trending hashtags for a region and window: last 5 minutes, 1 hour, 24 hours.
• Results refresh at least every minute.
• Exclude spam, banned and manipulated tags.

Non-functional

• Read latency under 50 ms (it is on the home/explore screen); reads vastly outnumber computations.
• Handle extreme skew: one tag can be 10%+ of all traffic during a final over.
• Approximate counts acceptable for real-time; daily reports should be exact.
• Fault-tolerant: a crashed worker must not lose or double count large amounts of data.

Back-of-the-Envelope Estimates

• 500M posts/day, average 2 hashtags → 1B hashtag events/day ≈ 11.6k/s average; a 10x spike during a big match → ~120k/s.
• Distinct hashtags per day: assume 50M (a long tail of one-off tags).
• Exact counts for one day, one region: 50M entries × ~80 bytes (string, counter, hash map overhead) ≈ 4 GB. Fits on one large machine, but multiply by 1,440 minute buckets × 200 regions and it stops fitting anywhere.
• Event size ~100 bytes (tag, region, user id, timestamp) → ~100 GB/day into Kafka. Small.
• Reads: 100M DAU opening explore a few times → ~5k reads/s, trivially served from a cache because the answer is the same for everyone in a region.

Conclusion to say aloud: the throughput is modest; the challenges are skew, window bookkeeping and memory per window. That points toward partitioned streaming plus sketches.

Step 1: Exact Counting with a Hash Map and a Min-Heap

On one machine: count with a hash map, then keep a min-heap of size K. Each item is pushed; when the heap exceeds K, pop the smallest. Time O(N + D log K) for N events and D distinct items, memory O(D). This is LeetCode 347 territory; the heap technique is covered in heap and priority queue problems.

import heapq
from collections import Counter

def top_k(events, k):
    counts = Counter(events)                      # O(N)
    heap = []                                     # min-heap of (count, tag)
    for tag, c in counts.items():                 # O(D log K)
        heapq.heappush(heap, (c, tag))
        if len(heap) > k:
            heapq.heappop(heap)
    return sorted(heap, reverse=True)

print(top_k(["#ipl", "#ipl", "#rcb", "#ipl", "#csk", "#rcb"], 2))
# [(3, '#ipl'), (2, '#rcb')]

Scaling exact counting: partition by key

The key insight: if every event for a given hashtag goes to the same worker (hash partition by hashtag), then each worker holds the exact count of its tags, and the global top-K is exactly the top-K of the union of each worker's local top-K. No tag can be split across workers, so nothing is missed. With 64 partitions, each worker holds ~1/64 of the distinct tags.

Exact partitioned counting is a perfectly good answer for one window. Memory becomes the problem when you need many overlapping windows, many regions, and fine-grained buckets. That is where sketches come in.

Step 2: The Count-Min Sketch

A count-min sketch (Cormode and Muthukrishnan, 2005) is a d × w table of counters and d independent hash functions. To add an item, increment one counter per row (the column chosen by that row's hash). To estimate an item's count, take the minimum of its d counters.

• Collisions only ever add to a counter, so estimates never undercount.
• Taking the minimum across rows picks the row with the least collision noise.
• Guarantee: with w = ⌈e/ε⌉ and d = ⌈ln(1/δ)⌉, the estimate exceeds the true count by at most εN with probability at least 1 − δ, where N is the total count in the sketch.

Sizing example

Want error ≤ 0.01% of the stream (ε = 0.0001) with 99.9% confidence (δ = 0.001): w = ⌈2.718 / 0.0001⌉ = 27,183, d = ⌈ln 1000⌉ = 7. That is 190,281 counters × 4 bytes ≈ 760 KB, regardless of whether there are 1 thousand or 50 million distinct hashtags. With 1B events/day, the error bound is 100,000 counts, which is small compared with what a trending tag gets, but large compared with a tag that has 500 mentions. Sketches are great at finding heavy hitters and useless for ranking the long tail.

Dry run (w = 4, d = 2)

Estimate(#ipl) = min(row0[1]=3, row1[3]=3) = 3 (true: 2, both rows collided). Estimate(#rcb) = min(3, 1) = 1, exact, because row 1 had no collision. Estimate(#csk) = min(1, 3) = 1, exact. More rows mean more chances that at least one row is collision-free.

Making it a top-K: sketch plus candidate set

The sketch counts but does not remember which items exist. Pair it with a small map of the current top candidates: on each event, update the sketch, read the new estimate, and if it beats the weakest candidate, swap.

import hashlib
import heapq
import math

class CountMinSketch:
    def __init__(self, epsilon=0.001, delta=0.001):
        self.w = math.ceil(math.e / epsilon)          # columns
        self.d = math.ceil(math.log(1 / delta))       # rows (hash functions)
        self.table = [[0] * self.w for _ in range(self.d)]
        self.total = 0

    def _cols(self, item):
        data = item.encode()
        # one independently salted hash per row
        return [int.from_bytes(hashlib.blake2b(data, digest_size=8,
                                               salt=row.to_bytes(16, "little")).digest(),
                               "little") % self.w
                for row in range(self.d)]

    def add(self, item, count=1):
        self.total += count
        for row, col in enumerate(self._cols(item)):
            self.table[row][col] += count

    def estimate(self, item):
        return min(self.table[row][col] for row, col in enumerate(self._cols(item)))

    def merge(self, other):                         # sketches are linear: add cellwise
        for r in range(self.d):
            for c in range(self.w):
                self.table[r][c] += other.table[r][c]
        self.total += other.total

class TopK:
    """Count-min sketch for counts + a small candidate map for the current top K."""
    def __init__(self, k, **cms_args):
        self.k, self.cms, self.top = k, CountMinSketch(**cms_args), {}

    def add(self, item):
        self.cms.add(item)
        est = self.cms.estimate(item)
        if item in self.top or len(self.top) < self.k:
            self.top[item] = est
        else:
            weakest = min(self.top, key=self.top.get)
            if est > self.top[weakest]:
                del self.top[weakest]
                self.top[item] = est

    def result(self):
        return heapq.nlargest(self.k, self.top.items(), key=lambda kv: kv[1])

stream = ["#ipl"] * 50 + ["#diwali"] * 30 + ["#budget"] * 20 + [f"#rare{i}" for i in range(500)]
t = TopK(3, epsilon=0.01, delta=0.01)
for tag in stream:
    t.add(tag)
print(t.result())   # [('#ipl', 50), ('#diwali', 30), ('#budget', 20)]

The min(self.top, …) scan is O(K); for K ≤ 50 that is fine. For larger K, use a min-heap with lazy updates or an indexed heap.

#include <algorithm>
#include <cmath>
#include <cstdint>
#include <functional>
#include <iostream>
#include <limits>
#include <string>
#include <vector>

class CountMinSketch {
public:
    CountMinSketch(double epsilon, double delta)
        : w_(static_cast<size_t>(std::ceil(std::exp(1.0) / epsilon))),
          d_(static_cast<size_t>(std::ceil(std::log(1.0 / delta)))),
          table_(d_, std::vector<uint32_t>(w_, 0)) {}

    void add(const std::string& item, uint32_t count = 1) {
        const uint64_t base = std::hash<std::string>{}(item);
        for (size_t i = 0; i < d_; ++i) table_[i][col(base, i)] += count;
    }

    uint32_t estimate(const std::string& item) const {
        const uint64_t base = std::hash<std::string>{}(item);
        uint32_t best = std::numeric_limits<uint32_t>::max();
        for (size_t i = 0; i < d_; ++i) best = std::min(best, table_[i][col(base, i)]);
        return best;
    }

    size_t width() const { return w_; }
    size_t depth() const { return d_; }

private:
    static uint64_t mix(uint64_t x) {                 // splitmix64 finaliser
        x += 0x9e3779b97f4a7c15ULL;
        x = (x ^ (x >> 30)) * 0xbf58476d1ce4e5b9ULL;
        x = (x ^ (x >> 27)) * 0x94d049bb133111ebULL;
        return x ^ (x >> 31);
    }
    size_t col(uint64_t base, size_t row) const {   // independent hash per row
        return mix(base + 0x632be59bd9b4e019ULL * (row + 1)) % w_;
    }

    size_t w_, d_;
    std::vector<std::vector<uint32_t>> table_;
};

int main() {
    CountMinSketch cms(0.0001, 0.001);                // error <= 0.01% of N w.p. 99.9%
    std::cout << cms.width() << " x " << cms.depth() << "\n";   // 27183 x 7
    for (int i = 0; i < 1000; ++i) cms.add("#ipl");
    for (int i = 0; i < 300; ++i) cms.add("#diwali");
    std::cout << cms.estimate("#ipl") << " " << cms.estimate("#diwali") << " "
              << cms.estimate("#never-seen") << "\n";             // 1000 300 0
}

Complexity: O(d) per update and query, O(d·w) memory, independent of the number of distinct items. Two sketches with the same dimensions and hash functions merge by cell-wise addition, which is what makes them so useful in distributed and windowed settings. (A related trick, "conservative update", increments only the counters equal to the current minimum and reduces overestimation, at the cost of losing mergeability guarantees for deletions.)

Alternative: Space-Saving and Misra-Gries

Counter-based algorithms keep only m (item, count) pairs and directly track the candidates:

def space_saving(stream, m):
    """Keep at most m counters. Any item with true frequency > N/m is guaranteed
    to be present; each count overestimates by at most its recorded error."""
    counts, errors = {}, {}
    for x in stream:
        if x in counts:
            counts[x] += 1
        elif len(counts) < m:
            counts[x], errors[x] = 1, 0
        else:
            victim = min(counts, key=counts.get)      # use a stream-summary structure for O(1)
            floor = counts.pop(victim)
            errors.pop(victim)
            counts[x], errors[x] = floor + 1, floor   # inherit the evicted count
    return sorted(counts.items(), key=lambda kv: -kv[1])

print(space_saving(list("aaaaabbbcdaaefbbg"), 3))
# [('a', 7), ('b', 5), ('g', 5)]  -> g's error is 4, so its true count is at least 1

Both are good interview answers. Mentioning the trade-off between them, rather than just one, is what earns the "strong hire" note.

Windows: Tumbling, Sliding and Hopping

"Top hashtags in the last hour" needs a window definition:

• Tumbling: fixed, non-overlapping (10:00–11:00, 11:00–12:00). Cheap but stale for up to an hour, and jumpy at boundaries.
• Sliding: always the last 60 minutes, evaluated continuously.
• Hopping: a 60-minute window that advances every 1 minute: the practical approximation of sliding.

Implementation with minute buckets

Keep counts (exact maps per partition, or sketches) per 1-minute bucket in a ring of 60. Maintain a running total:

1. At each minute boundary, add the just-closed bucket to the running window.
2. Subtract the bucket that fell out of the window (61 minutes old) and free it.
3. Recompute top-K from the running window's candidates.

With count-min sketches this is literally adding and subtracting two 760 KB arrays, so maintaining 5-minute, 1-hour and 24-hour windows for 200 regions is cheap. For the 24-hour window, use coarser buckets (hourly) to bound memory: 24 sketches instead of 1,440.

Event time vs processing time

Mobile clients upload posts late (offline, flaky networks). Assign events to buckets by event time, allow a lateness bound (say 2 minutes) using watermarks, and drop or side-output events that arrive later. Frameworks like Apache Flink handle watermarks and windowing natively.

Time Decay and the Trending Score

Exponential decay instead of hard windows

A hard 1-hour window treats a mention from 59 minutes ago the same as one from 1 second ago, then forgets it completely. Exponential decay weights each mention by e−λ·age, with half-life ln 2 / λ. The score updates smoothly and needs only one number per item.

Efficient trick (forward decay): instead of decaying every score every second, add eλ·(t − t0) for an event at time t, where t0 is a fixed reference. Scores grow over time, but their relative order is exactly the order of the decayed scores. Periodically move t0 forward and rescale all scores to avoid floating-point overflow.

import math

HALF_LIFE_S = 15 * 60
LAM = math.log(2) / HALF_LIFE_S

class DecayedCounter:
    def __init__(self, t0):
        self.t0, self.score = t0, {}

    def add(self, tag, t):
        self.score[tag] = self.score.get(tag, 0.0) + math.exp(LAM * (t - self.t0))

    def decayed(self, tag, now):          # value as of `now`, comparable to a raw count
        return self.score.get(tag, 0.0) * math.exp(-LAM * (now - self.t0))

    def rebase(self, new_t0):             # call every few hours to avoid overflow
        f = math.exp(-LAM * (new_t0 - self.t0))
        self.score = {k: v * f for k, v in self.score.items()}
        self.t0 = new_t0

Trending = velocity relative to baseline

Ranking by decayed count still surfaces evergreen giants. To find what is unusually active, compare the current rate with the item's own history (same hour last week, or a trailing 7-day average), for example a z-score-like ratio:

def trending_score(now_count, baseline_mean, baseline_std, min_count=50):
    if now_count < min_count:                 # ignore tiny counts (noise, spam)
        return 0.0
    return (now_count - baseline_mean) / (baseline_std + 1.0)

trending_score(12000, 9000, 1500)   # 2.0   -> #ipl: big but expected
trending_score(900, 40, 15)         # 53.75 -> a breaking-news tag: trending

The baseline itself comes from the batch layer (historical hourly counts per tag), stored in a key-value store and fetched for the few thousand current candidates. Production systems also weigh signals such as the number of distinct users (not just mentions) and diversity of locations.

Streaming Architecture

Why each stage exists

• Stage 1 pre-aggregation defuses hot keys. Without it, every #ipl event during a final lands on one Stage 2 partition. With 5-second local combining across 100 Stage 1 workers, a tag with a million mentions per minute becomes about 1,200 partial updates per minute.
• Stage 2 keyed by (region, tag) gives exact per-key counts within its partition (or per-partition sketches if memory demands it), and its local top-K lists are correct inputs for a global merge.
• Stage 3 merges small lists (K × partitions per region), applies trending scores and editorial/policy filters, and publishes.
• Precomputed results in Redis: every user in a region sees the same list, so reads are a single key fetch. This is why read latency is trivial.

Fault tolerance

Flink checkpoints keyed state and Kafka offsets together, giving effectively exactly-once state updates on recovery. Even without that, trend lists are tolerant of small errors; you would never accept that for billing, but here a few duplicate counts after a crash do not change the ranking. Kafka retention lets you replay the last few hours to rebuild state. For more on the log underneath, see design a distributed message queue like Kafka.

MapReduce vs Streaming (and Using Both)

The MapReduce version of top-K is a nice whiteboard exercise: map emits (tag, 1); a combiner sums locally; reduce by tag gives exact counts; a second job keeps a top-K heap per reducer and a final single reducer merges them. The streaming version is the same dataflow running continuously.

The Lambda architecture runs both: streaming for speed, batch for correctness, merging at query time. The Kappa architecture runs only streaming and reprocesses by replaying the log. For trends, a pragmatic split is streaming for the live list and batch for baselines and the "top hashtags of 2026" report.

Spam, Manipulation and Quality

The hardest part of real trending systems is not counting; it is not getting gamed. Coordinated campaigns and bot networks can push a tag with thousands of accounts.

• Count distinct users, not mentions: a HyperLogLog per (tag, window) estimates unique authors in ~12 KB with ~1% error. 10,000 mentions from 50 accounts is not a trend.
• Per-user caps: count at most one mention per user per tag per window. A Bloom filter per window can cheaply skip repeats.
• Account quality weighting: new accounts, accounts flagged by spam models, and accounts with no history count for less.
• Velocity anomaly detection: organic trends ramp up across geographies and client types; bot-driven spikes often have unnatural shapes.
• Policy filters and human review for sensitive topics; a denylist checked in Stage 3.
• Normalisation: case folding, Unicode normalisation, merging obvious variants (#IPL2026, #ipl2026).

Interview Walkthrough, Follow-ups and Mistakes

1. Clarify: top-K or trending? Windows? Regions? Freshness? Approximate OK?
2. Estimate: events/s, distinct keys, memory for exact counts per window. Show where exact breaks.
3. Single machine: hash map + min-heap, complexity.
4. Scale: partition by key, local top-K, global merge; explain why merge is exact only when partitioned by key.
5. Memory and windows: minute buckets, count-min sketch (with sizing math), mergeability.
6. Trending: decay and baseline comparison.
7. Architecture: Kafka, pre-aggregation for hot keys, Flink, Redis, cached reads.
8. Quality: distinct users, spam weighting.

Follow-up questions to prepare

• "Why not just use a Redis sorted set with ZINCRBY?" Works well at moderate scale for one window (and is great for leaderboards), but you would need one sorted set per window/bucket/region, a hot key on viral tags, and sliding windows require subtracting expired counts. Fine for an MVP, say so.
• "Can the count-min sketch undercount?" No, only overcount (with non-negative updates). Subtracting expired buckets keeps that property because you subtract exactly what you added.
• "How would you do this per city for 5,000 cities?" Hierarchical: count at city level, roll up to state and country by merging (sketches merge exactly). Store only top lists per city; most cities have tiny volumes.
• "Same problem for search queries?" Same pipeline; then feed the results into autocomplete suggestions.

Common mistakes

• Jumping to count-min sketch without first showing exact counting and why it does not fit.
• Forgetting that a sketch cannot list items: no candidate heap, no answer.
• Merging local top-Ks from randomly partitioned data.
• Ignoring hot keys; the system that cannot handle #ipl on final day is the system that fails.
• Computing top-K on every read instead of precomputing per region and window.
• Treating trending as "most mentions" without a baseline.

Frequently Asked Questions

Pranjal Jain

Ex-Microsoft SDE · IIT Kanpur · Founder of Prepflix. Helps engineers crack startup and product company interviews across India.