topk

Start with the simplest case: one machine counts all song plays and maintains the top K. This is fundamentally a data structures problem.

The Data Structure: Min-Heap + HashMap

If you've solved LeetCode's "Top K Frequent Elements", the approach is familiar: a min-heap of size K keeps the K highest-count songs. The song at the root has the lowest count among the top K — any new song that exceeds this threshold displaces it.

Alongside the heap, a hash map stores two things per song:

• The total play count
• The song's index in the heap (if it's in the top K)

The heap index is critical for performance. When a song already in the top K gets another play, we need to update its count and re-heapify. Without the index, finding the song in the heap is O(K). With the index, it's O(1) lookup + O(log K) re-heapify.

Complexity

Where N = number of distinct songs (100M) and K = top items (100–1,000).

✅ Works for: Low-throughput streams where one machine can handle all events. Correct, simple, and easy to implement.

❌ Missing: Can't handle 115K events/sec on a single machine. If the instance crashes, the in-memory heap and counts are lost. Only supports all-time (global window) counts.

A single instance can't handle 115K events/sec. We need to partition the event stream across multiple counting nodes.

Partitioning Strategy

The key design decision: how do we route events to counting nodes?

We choose hash partitioning by song_id. Each counting node maintains a complete, correct count for its partition of songs. The coordinator doesn't need to merge raw counts — it only merges per-partition top-K results.

Coordinator Aggregation

Each partition maintains its own local top-K heap. A Coordinator Service periodically (or on query) collects the local top-K from each partition and merges them into a global top-K:

1. Each of the N partitions returns its local top-K (K items)
2. The Coordinator merges N × K items using a heap
3. The final global top-K is the top K from the merged result

This works correctly because hash partitioning ensures a song's entire count lives on one partition. If song X is globally top-K, it must be in its partition's local top-K (assuming K is the same everywhere).

✅ NFRs addressed: Throughput (partitioned across N nodes — linear scaling), accuracy (hash partition ensures complete per-song counts)

❌ Still missing: If a counting node crashes, its in-memory state is lost. Recovery requires replaying the entire event stream from the beginning — too slow for a 10B events/day system. Still only supports global (all-time) window.

If a counting node crashes, its heap and hash map are gone. Without persistence, recovery means replaying every event from the beginning of the stream — days or weeks of data. At 115K events/sec, even replaying one day takes ~24 hours at full speed. We need a smarter recovery strategy.

Periodic Snapshots

Each counting node periodically serializes its state to a durable store (S3, Redis, DynamoDB):

1. Heap state — The current min-heap contents
2. Count map — The full hash map of song_id → count
3. Stream offset — The Kafka/Kinesis offset of the last event processed

On crash recovery:

1. Load the latest snapshot → restores heap + counts to a known point
2. Replay events only from the snapshot's stream offset to the current head
3. The node is caught up after replaying a small window of events

Snapshot Interval Trade-off

• Frequent snapshots (every 1 min) → fast recovery, but more I/O and storage
• Infrequent snapshots (every 1 hour) → less overhead, but 1 hour of replay on recovery

A good default: snapshot every 5 minutes. At 115K events/sec, that's ~35M events to replay in the worst case — takes about 5 minutes at 2× ingestion speed.

Exactly-Once Semantics

The snapshot + offset approach gives us effectively exactly-once processing:

• Events before the snapshot offset are already counted (in the snapshot)
• Events after the snapshot offset are replayed on recovery
• The boundary is clean — no events are double-counted or missed

This is the same pattern used by Flink's checkpointing mechanism internally.

✅ NFRs addressed: Fault tolerance (crash recovery in minutes, not hours), data durability (state survives node failures)

❌ Still missing: Only supports global (all-time) window. Users asking "top songs in the last hour" can't be served — we'd need to expire old events and maintain time-bounded counts.

A sliding window requires two operations: adding new events as they arrive, and removing old events as they slide out of the window. Adding is easy — it's the same process_event logic. Removing is the hard part.

The Two-Pointer / Dual-Offset Approach

If you've solved sliding window problems on LeetCode, you know the two-pointer technique. We apply the same idea to stream offsets:

• End offset — Points to the latest event processed. Advances as new events arrive. Events at this offset are added to counts.
• Beginning offset — Points to the oldest event still within the window. Advances as time passes. Events at this offset are subtracted from counts.

The top-K result at any moment reflects only the events between the beginning and end offsets.

Processing Logic

For each window update cycle:

1. Advance the end offset — Process new events using the standard heap logic (add to counts, update heap)
2. Advance the beginning offset — For each expired event:
   • Decrement the song's count in the hash map
   • If the song is in the top-K heap and its count drops below the heap root, replace it
   • If the count reaches zero, remove from the hash map

The Subtraction Problem

Removing events from a min-heap is harder than adding. When a song's count decreases, it might no longer deserve to be in the top-K. But which song replaces it? We'd need to scan all non-heap songs to find the new K-th largest — expensive.

Practical solution: Instead of maintaining a perfect top-K heap during subtraction, run a periodic recomputation. Every few seconds, rebuild the heap from the hash map (O(N) to find top K via selection algorithm). Between recomputations, the heap may be slightly stale — but for a music leaderboard, a few seconds of staleness is invisible to users.

Alternative: Bucketed Tumbling Windows

A simpler approximation: divide time into small tumbling buckets (e.g., 1-minute buckets). Each bucket stores its own count map. To answer "top-K in the last hour", aggregate the last 60 buckets. Old buckets are discarded entirely. This avoids per-event subtraction at the cost of bucket-boundary imprecision.

✅ NFRs addressed: Freshness (results reflect the current time window), query flexibility (any time window can be served)

❌ Still missing: At 115K events/sec with 100M distinct songs, exact counting requires ~3.2 GB of memory per window per partition. For multiple windows or limited memory, this is expensive.

The final pieces: approximate counting to reduce memory, and a result cache to serve the top-K API at low latency.

Count-Min Sketch for Approximate Counting

A Count-Min Sketch (CMS) is a probabilistic data structure that estimates frequencies using fixed memory. Instead of a hash map with one entry per distinct song, CMS uses a 2D array of counters with multiple hash functions:

• Structure: depth × width grid of counters (e.g., 5 × 2,718 = ~109 KB)
• Insert: Hash the song_id with each of the depth hash functions, increment the corresponding counter in each row
• Query: Hash the song_id, return the minimum across all rows (this minimizes overcounting from hash collisions)

CMS never undercounts — it can only overcount due to hash collisions. The error is bounded: with 99% probability, the overcount is at most 0.1% of total events.

When to use CMS vs exact counting:

Result Cache

The top-K API doesn't query the counting nodes directly. Instead:

1. The Coordinator periodically (every few seconds) collects local top-K from each partition
2. Merges them into a global top-K
3. Writes the result to a Redis cache or similar fast store
4. The API Gateway reads from the cache — O(1) latency, no fan-out on read path

This separates the write path (event counting, high throughput) from the read path (top-K queries, low latency).

NFR Scorecard

If exact accuracy isn't required — and for most leaderboards it isn't — Count-Min Sketch (CMS) reduces memory by orders of magnitude while maintaining bounded error guarantees.

How CMS Works

CMS is a 2D array of counters with dimensions depth × width:

• depth = number of independent hash functions (controls confidence)
• width = number of counters per hash function (controls accuracy)

Insert (song_id played):

for each hash function h_i (i = 1..depth):
    index = h_i(song_id) % width
    table[i][index] += 1

Query (how many plays for song_id?):

return min(table[i][h_i(song_id) % width] for i in 1..depth)

The minimum is taken because collisions can only increase a counter, never decrease it. The true count is always ≤ the CMS estimate.

Error Bounds

For a CMS with width $w$ and depth $d$, and total events $N$:

$\text{Overcount} \leq \frac{N}{w} \text{ with probability } \geq 1 - \left(\frac{1}{e}\right)^d$

With $w = 2{,}718$ and $d = 5$: the overcount is at most 0.04% of total events with 99.3% confidence.

CMS + Heap for Top-K

CMS alone only answers frequency queries — it doesn't track which items are most frequent. We combine CMS with a min-heap:

1. On each event, update the CMS count for that song_id
2. Query the CMS for the song's estimated count
3. If the count exceeds the heap root, insert into the K-sized heap

This gives us approximate top-K in O(depth) time per event and O(width × depth + K) memory total.

CMS vs HashMap: When to Choose

Hash partitioning routes all plays for a song to one node. But when a song goes viral (a new Taylor Swift release), that partition receives disproportionate traffic — potentially 10× the average — creating a hotspot.

Detecting Hot Keys

Before solving the problem, we need to detect it. Monitor per-partition event rates and flag partitions exceeding a threshold (e.g., 2× average). Alternatively, sample the event stream and run a small top-K on the sample to identify hot songs.

Strategies for Hot Keys

Pre-Aggregation (Recommended)

The most practical approach: add a combiner layer between the stream and the counting nodes. Multiple combiner instances accumulate counts for a short window (e.g., 1 second), then flush aggregated (song_id, delta_count) tuples to the counting nodes.

If a song gets 10,000 plays in one second across 10 combiners, each combiner sends one aggregated update (count: 1,000) instead of 1,000 individual events. The counting node sees 10 messages instead of 10,000 — a 1,000× reduction.

This is essentially the MapReduce combiner pattern applied to stream processing.

An interviewer might challenge your window choice: "Do you really need a sliding window?" The answer depends on the product requirement.

Tumbling Window (Simpler)

• Fixed, non-overlapping time buckets (e.g., hourly)
• At each bucket boundary, reset all counts and start fresh
• No event subtraction needed — just discard old buckets
• Results change at bucket boundaries (potentially jarring for users)

Sliding Window (Smoother)

• Continuously moving window (e.g., "last 24 hours" always means the most recent 24h, not "since midnight")
• Requires event expiration (subtraction) as events leave the window
• Results update smoothly — no sudden resets

Hybrid: Bucketed Sliding Window

In practice, most production systems use a bucketed approximation of sliding windows:

1. Divide time into small tumbling buckets (e.g., 1-minute buckets)
2. Each bucket maintains its own count map
3. To answer "top-K in the last hour": aggregate the last 60 buckets
4. When a bucket expires (older than 1 hour), discard it entirely

This avoids per-event subtraction while giving results that are at most 1 bucket stale (1 minute in this example) — acceptable for virtually all leaderboard use cases.

Production Reality

• Spotify updates Top Songs daily (24-hour tumbling window)
• YouTube's Trending is updated hourly (1-hour tumbling)
• Twitter's Trending Topics use ~15-minute sliding windows
• Amazon's Best Sellers use daily tumbling windows

For most Top-K interviews, start with tumbling (show you understand the simple case), then propose sliding if the interviewer asks for more freshness.

In practice, you wouldn't build the counting and windowing infrastructure from scratch. Stream processing frameworks handle partitioning, windowing, fault tolerance, and exactly-once semantics out of the box.

Apache Flink Approach

Flink provides built-in windowing, state management, and checkpointing. A sliding window top-K in Flink is roughly:

Flink handles:

• Partitioning: .keyBy() does hash partitioning
• Windowing: .window() manages window boundaries and event assignment
• Fault tolerance: Periodic checkpoints (snapshots) to durable storage
• Exactly-once: Via Chandy-Lamport distributed snapshot algorithm

Redis Sorted Set for Global Window

For the simpler global (all-time) top-K, a Redis sorted set (ZSET) is even simpler:

Redis ZSET uses a skip list internally. ZINCRBY is O(log N) and ZREVRANGE is O(log N + K). For a global window with moderate cardinality, this is the simplest production-ready solution.

When to build from scratch vs use a framework:

<discuss-with-ai-button title="Production Stream Processing for Top-K" context="Flink provides built-in partitioning, windowing, and checkpointing. Redis ZSET is the simplest approach for global window top-K. In interviews, design from scratch to show understanding, then mention you'd use Flink in production." points='["How does Flink's checkpointing (Chandy-Lamport) compare to our manual snapshot approach?","What are the latency implications of using Flink vs a custom implementation?","How would you handle late-arriving events in Flink?","Can Redis ZSET handle 115K writes/sec — what are its limits?","How do you test a stream processing pipeline — can you simulate replay?"]'>

The following topics contain open-ended architectural questions for staff+ conversations.