Top-K Trending Items (Heavy Hitters)

0) Problem Restatement

Design a system that watches a huge stream of events and always knows the top K items. Interviewers phrase it in many ways:

• The 10 most viewed URLs today, and in the last 5 minutes.
• The 100 users with the most events in the last hour.
• The services with the most error logs in the last 5 minutes.
• Trending hashtags, or the top songs in each country right now.

The stream is too big to store and re-count every time someone asks, so we must count as the events arrive and keep the answer ready.

1) Requirements

1.1 Functional

• Ingest events like { item_id, timestamp, country }.
• Answer "top K items in the last N minutes/hours/day" for a few fixed windows (1 min, 1 hour, 24 hours) and for all time.
• Support grouping (e.g., per country), with K up to about 100.

1.2 Non-Functional

• Fresh: results at most about a minute old.
• Fast reads: under 50 ms.
• Scale: billions of events per day.
• Accuracy: top items must be right; small count errors are okay for trending, but may need to be exact for billing.

1.3 Scale Estimates

• 5 billion events/day ≈ 60,000 events/sec, with peaks of 200K/sec.
• 100 million distinct items (URLs, songs). Keeping an exact counter for each item per minute would mean billions of counters, which is too much memory for a single machine.

1.4 API Design

• GET /v1/topk?window=1h&k=10&country=IN → [{ item_id, count }, ...]
• Events enter through a queue (Kafka topic events), not a public API.

2) High-Level Architecture

2.1 Overview

• Kafka: receives all events, partitioned by item_id, so every event for one item goes to the same partition.
• Stream processors (e.g., Flink): each one counts items for its partitions in small time buckets and keeps a local top-K.
• Aggregator: merges the local top-K lists into a global top-K for each window and writes it to a fast store.
• Top-K store (Redis): holds the final answers, ready to read.
• Batch job (optional): recomputes exact results from raw logs every hour to correct the fast path.

2.2 Architecture Diagram

flowchart LR
    P["Producers - apps, services"] --> K[("Kafka - partitioned by item_id")]
    K --> F1["Counter 1 - local top-K"]
    K --> F2["Counter 2 - local top-K"]
    K --> F3["Counter N - local top-K"]
    F1 --> AGG["Aggregator - merge top-K"]
    F2 --> AGG
    F3 --> AGG
    AGG --> R[("Redis - top-K per window")]
    API["Top-K API"] --> R
    K --> S3[("Raw event log")]
    S3 --> B["Hourly batch - exact counts"]
    B --> R

3) Core Algorithm (Simple Version)

Keep a count per item, then pick the K largest using a min-heap of size K. A min-heap is a structure where the smallest element sits on top. For each item, if its count is bigger than the smallest one in the heap, replace it. This takes O(N log K) time instead of sorting all N items.

4) Handling Time Windows

• Tumbling window: fixed, non-overlapping buckets (12:00–12:01, 12:01–12:02). Easy.
• Sliding window ("last 60 minutes, right now"): keep 1-minute buckets of counts and add up the last 60. Every minute, add the new bucket and drop the oldest one.

For "last 24 hours", use 1-hour buckets. We never store per-second data for long windows.

5) Deep Dive A — When exact counting uses too much memory

With 100M items per window, even one counter per item per bucket gets big. Two well-known tricks help:

• Count-Min Sketch: a small 2D array of counters with several hash functions. To add an item, increase one cell in each row. To read a count, take the minimum across rows. It never under-counts and only slightly over-counts, using fixed memory (a few MB) no matter how many items there are. Pair it with a heap of the current top K.
• Space-Saving algorithm: keep exactly M counters (say 10,000). When a new item arrives and the counters are full, replace the smallest counter and inherit its count. Heavy hitters are guaranteed to stay in the list.

Use approximate counting for "trending". For anything tied to money, rely on the exact batch path.

6) Deep Dive B — Hot items and many dimensions

• Hot keys: one viral song can get 50K events/sec, all landing on one partition. Fix: pre-aggregate at the producer or in a first stage (count locally for 1 second, then send {item, +523}), or split the hot key into item#1..item#8 sub-keys and add them back up.
• Per-country top-K: use the key (country, item). The work grows with the number of countries, so only pre-compute the dimensions the product actually shows.
• Two-stage merge: with 200 counters, merge in a tree (200 → 20 → 1) so no single aggregator becomes a bottleneck.

7) Trade-offs & Alternatives

8) Common Follow-up Questions

• "What if K changes, e.g., top 1,000?" Keep a larger candidate list (say 5K) per partition so any K up to that works.
• "Top K under a strict memory limit on one machine?" Use Space-Saving or Count-Min Sketch with a heap, and explain the error bounds.
• "How do you avoid spam inflating counts?" Deduplicate by user per window (count unique users, e.g., with HyperLogLog) and filter bots before counting.

9) Wrap-Up

Partition events by item so each count lives in one place. Count in small time buckets inside a stream processor, keep a local top-K with a min-heap, merge the local lists into a global answer, and store it in Redis for fast reads. Use sketches when memory is tight, pre-aggregate hot items, and let a batch job correct the fast path when exact numbers matter.