Problem RestatementProblem
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.
RequirementsRequirements
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.
High-Level ArchitectureArchitecture
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
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
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 --> RCore 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.
Why partitioning makes this correct: Kafka sends all events for an item to the same counter. So each item's full count lives on one machine, and the global top-K is always inside the union of the local top-Ks. The aggregator only merges a few thousand candidates.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.
Late events: an event from 12:00:58 may arrive at 12:01:10. Stream processors use a watermark, which means "wait up to X seconds for late data before closing a bucket". Anything later is either dropped or fixed by the batch job.Deep Dive A — Finding the top K without storing every itemDeep dive
The stream carries 100 million distinct items per window. How much memory does the answer need?
A counter for every item
Keep a hash map from item to count, sort it when someone asks. Correct by construction, and fine for a demo with a thousand items.
At real scale the map is the problem: 100M keys at roughly 60 bytes each is about 6 GB per window, per aggregator, and sorting 100M entries to answer "top 10" throws away 99.99999% of the work. Memory grows with the number of distinct items, which is exactly the number we do not control.
Count-Min Sketch plus a small heap
Replace the map with a fixed 2D array of counters and a few hash functions. To add an item, increment one cell per row; to read its count, take the minimum across rows. Keep a heap of the current top K alongside it.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
E["event - item X"] --> H1["hash 1"]
E --> H2["hash 2"]
E --> H3["hash 3"]
H1 --> R1["row 1 - cell +1"]
H2 --> R2["row 2 - cell +1"]
H3 --> R3["row 3 - cell +1"]
R1 --> M["count = min of the three cells"]
R2 --> M
R3 --> M
M --> HEAP["top-K heap"]Memory is now a few MB and fixed, no matter how many distinct items arrive. The sketch never under-counts; collisions only push a count up. The catch: a rare item that collides with a viral one can be over-counted into the heap, and you cannot ask the sketch "which items are heavy?" — only "how heavy is this item?", so the heap has to be maintained separately.
Space-Saving for the live answer, exact batch for the numbers that matter
Keep exactly M counters, say 10,000. When a new item arrives and the counters are full, evict the smallest and let the newcomer inherit its count. The structure answers the question directly — it is the candidate list — and it guarantees that any item genuinely above the 1/M frequency threshold is still in the list.
Then split the problem by what the number is for:
- Trending, "what's hot right now" — the approximate answer is the product. Nobody audits a trending list.
- Anything tied to money or an SLA — recount exactly from the raw events in the nightly batch job, and let those numbers overwrite the streamed ones.
Naming that split is the part interviewers listen for: approximation is a choice about which errors are acceptable, not a shortcut.
Deep Dive B — Hot items and many dimensionsDeep dive
- 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 intoitem#1..item#8sub-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.
Trade-offs & AlternativesTrade-offs
| Decision | Choice | Why | Alternative |
|---|---|---|---|
| Counting | Exact counts per partition | Correct top-K | Count-Min Sketch: much less memory, slightly off |
| Windows | 1-min buckets summed | Simple sliding windows | Per-event timestamps: exact but very heavy |
| Freshness vs accuracy | Real-time path + hourly batch fix | Fast and eventually exact | Batch only: accurate but hours late |
| Serving | Precomputed results in Redis | Very fast reads | Query an OLAP DB on demand: flexible, slower |
Common Follow-up QuestionsFollow-ups
- "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.
Wrap-UpWrap-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.