Design Top-K / Trending

“Show me the 10 most-watched videos in the last hour.” “What are the top trending hashtags right now?” “Which products are selling fastest this minute?” These are all the same problem: top-K by frequency over a window, computed from an event stream so large that you cannot simply keep a counter per item.

YouTube sees billions of views a day; Twitter/X processes hundreds of thousands of events per second. There may be hundreds of millions of distinct items (videos, hashtags, products). Keeping an exact integer counter for every one of them, updated on every event, and sorted continuously, does not fit in memory or keep up with the write rate. This question is a showcase for probabilistic data structures and the streaming-plus-batch pattern.

Understanding the Problem

Functional Requirements

• Top-K items by frequency (count of events) — typically K is small (10, 100, 1000)
• Over a time window: last 1 minute / 1 hour / 1 day, often sliding
• Near-real-time: trending should reflect the last few seconds to minutes
• Scoped queries optionally (top-K per country, per category)

Non-Functional Requirements

• Firehose ingestion: hundreds of thousands to millions of events/sec
• Bounded memory: must not grow with the number of distinct items
• Low query latency: reading the current top-K is a cheap lookup
• Good-enough accuracy: approximate ranks are fine for “trending”; exact counts only where it matters (billing, official charts)

Scale Estimation

Take 1,000,000 events/sec and 200,000,000 distinct items over a day. An exact Map<itemId, count> would need ~200M entries — gigabytes — and every event is a random-access update to a huge hash map, plus continuous re-sorting to answer top-K. That does not scale. The insight: we do not need every count accurately; we need the few big ones. That relaxation is what makes the problem tractable.

Core Entities and APIs

Data Model

Conceptually we maintain, per window bucket, a compact summary rather than raw counts:

interface Event {
  itemId: string;     // video / hashtag / product
  ts: number;         // event time
  scope?: string;     // optional: country, category
}

// Per shard, per time bucket — fixed size, independent of #items:
interface Summary {
  sketch: CountMinSketch;      // approximate frequency of any item
  heap: MinHeap<{itemId, estCount}>;  // the current top-K candidates, size K
}

API Design

GET /trending?window=1h&k=10&scope=IN   — current top-K for a window/scope
# Ingestion is not a public API — events arrive via a log (Kafka).

High-Level Design

Events land in a durable log (Kafka), partitioned by item id so all events for one item go to the same shard. Each stream processor (Flink/Kafka Streams) consumes a partition and maintains, per time bucket, two small structures:

1. A Count-Min sketch — estimates the frequency of any item in fixed memory.
2. A size-K min-heap — holds the current top-K candidates for that shard.

On each event: update the sketch, read the item’s estimated count, and update the heap (if the estimate beats the heap’s minimum, it goes in and the smallest falls out). To get the global top-K, merge the per-shard heaps — a small, cheap operation since each holds only K entries.

Deep Dive: Approximate Counting

Exact counting is the thing that does not scale, so we approximate frequency with a Count-Min sketch (covered in the Data Structures for Big Data lesson). It is a 2-D array of counters with several hash functions; to count an item you increment one cell per hash row, and to estimate its count you take the minimum across those rows (minimum because hash collisions only ever add, never subtract).

class CountMinSketch:
    def __init__(self, width=1<<20, depth=5):
        self.w, self.d = width, depth
        self.table = [[0]*width for _ in range(depth)]
        self.seeds = list(range(depth))
    def add(self, item, c=1):
        for i, s in enumerate(self.seeds):
            self.table[i][hash((s, item)) % self.w] += c
    def estimate(self, item):
        return min(self.table[i][hash((s, item)) % self.w]
                   for i, s in enumerate(self.seeds))

It uses a fixed amount of memory — a few megabytes — regardless of whether there are a thousand distinct items or a billion. The trade-off is that estimates can over-count (never under-count) due to collisions; for trending, slightly inflated counts on a few long-tail items do not change who the clear leaders are. Pair the sketch with a size-K min-heap so you are not scanning all items to find the top: the heap always holds the current K biggest estimates, and a new event only matters if its item’s estimate exceeds the heap’s smallest.

Why not keep exact counts for just the top items? You do — the heap is the small set of “hot” items you track precisely-ish. The sketch answers “is this newcomer bigger than my current smallest leader?” for the hundreds of millions of items you are not tracking, in O(1) and constant space.

Deep Dive: Windowing

“Trending” always means “recently”, so counts must age out. Keep per-bucket sketches — for example one sketch per minute — rather than one ever-growing sketch.

• Tumbling windows: a fresh sketch per minute; a “last hour” query merges (sums) the 60 most recent minute-sketches. Count-Min sketches add element-wise, which makes merging trivial.
• Sliding windows: as time advances, drop the oldest bucket and add a new one — a ring buffer of sketches. This gives a smoothly moving “last hour” without recomputing from scratch.

The same element-wise addition lets you merge across shards: each shard owns a partition of items, and summing their per-bucket sketches (plus merging their K-heaps) yields the global, windowed top-K. Memory stays bounded by (#buckets) × (sketch size), independent of event volume.

Deep Dive: Accuracy — the Batch Reconciliation

The streaming path is fast but approximate: sketch over-counts, heaps can churn, and a late or duplicated event can nudge a rank. For trending, that is completely fine. For anything that must be exact — official “most-viewed” charts, creator payouts, ad billing — run a slower exact path alongside and reconcile, the lambda architecture from the Stream Processing lesson.

• Streaming (approximate): sketch + heap, updated continuously, serves the live “trending now” widget within seconds.
• Batch (exact): periodically (hourly/daily) run an exact aggregation over the raw event log — a straightforward GROUP BY itemId / MapReduce — to get true counts and the authoritative top-K. This is slow but perfectly accurate.

Serve the real-time view from the streaming path and correct it with the batch results when they land. Users get instant-but-fuzzy trending that is periodically snapped to exact truth — the best of both worlds. (A modern Kappa variant does the same by replaying the log through the stream processor at higher fidelity, avoiding a separate batch codebase.)

Key Takeaways

1. Top-K is a heavy-hitters problem. You do not need every count — only the few big ones — and that relaxation is what makes firehose-scale feasible.
2. Count-Min sketch gives approximate frequency in fixed memory. A few MB answers “how often has this item appeared?” for any of billions of items, over-counting slightly but never under-counting.
3. A size-K min-heap tracks the current leaders so you never sort all items; an event only matters if its estimate beats the heap’s smallest.
4. Partition by item id and merge. Each shard keeps its own sketch + heap; element-wise sketch addition and K-heap merges produce the global top-K cheaply.
5. Window with per-bucket sketches. One sketch per minute, summed or slid over a ring buffer, ages counts out and supports “last hour / last day” without recomputation.
6. Reconcile with an exact batch path when accuracy matters (billing, official charts): fast approximate streaming for the live view, slow exact batch to correct it — the lambda pattern. See the Data Structures for Big Data and Stream Processing lessons for the underpinnings.