Walkthrough: Designing a Top K System

The problem

You’re asked to design a system that returns the top-K most frequent items over a time window at high throughput — trending hashtags on Twitter, top viewed videos on YouTube, live leaderboards in a game. The interviewer says something like: “Design a system that returns the top 100 trending hashtags over the last hour, with freshness in the seconds.”

Sounds like sort-and-take-top-K. It isn’t. This is the canonical “approximation at scale” problem, and the trap is in the first thirty seconds — candidates who reach for a database + ORDER BY LIMIT will be out of memory and IOPS before the second follow-up question. Candidates who evolve their design step by step — naive DB → cache → stream → probabilistic structures — land the signal the interviewer is looking for: recognizing that exact counting at event-stream scale is infeasible, and the right answer is a bounded-error approximation with explicit accuracy guarantees.

Below is how I’d walk through this, roughly in the order I’d speak the words, with the architecture evolving step by step.

1. Clarify before you design

First 3–5 minutes. Resist the urge to start drawing.

Questions I’d ask:

• What are we counting? Hashtag mentions, video views, product clicks, search queries? The cardinality of the key space matters — hashtags are tens of millions; search queries are billions.
• K value. Top 10, top 100, top 1000? K determines how much we need to track precisely versus approximate.
• Time window. Last hour sliding? Last 24 hours? All-time? Window semantics drive the entire state model.
• Freshness SLA. How stale can results be — one second, one minute, five minutes? Sub-second is a different system than every-minute.
• Exact or approximate. Is “approximately correct” acceptable? For trending surfaces (social feeds, dashboards), yes. For billing or compliance, no.
• Global or per-segment. One global top-K or sliced by region, language, user segment? Segmentation multiplies the state.
• Event volume. Throughput of the incoming stream? The trending question at Twitter scale is ~1M events/sec.

Exact vs approximate is the single biggest question. If the interviewer says “must be exact,” the design shifts toward durable storage and batch reconciliation, and the problem becomes much less interesting.

Say the interviewer confirms: counting hashtag mentions, K=100, sliding 1-hour window, freshness within 10 seconds, approximate acceptable, global top-K, event volume ~1M/sec peak.

Requirements

Functional Requirements

• Ingest item events carrying: item ID, event type, timestamp
• Query the top-K most frequent items within a configurable sliding time window (configurable K, configurable window size)
• Approximate results are acceptable — exact counts are not required

Non-Functional Requirements

• High ingest throughput: 1,000,000 events/second sustained
• Query latency < 100 ms end-to-end
• Bounded memory per processing node — approximation is the mechanism that enforces this, not a nice-to-have
• Window freshness < 1 minute (result reflects events from no more than 60 seconds ago)
• K up to 1,000

2. Capacity estimate

Brief. The numbers decide the architecture.

• 1M events/sec × 3600 seconds = ~3.6B events per hour in the window.
• Unique keys per hour: assume ~10M unique hashtags active per hour, ~100M all-time.
• Exact per-key count in a hash map: 10M keys × (~50 bytes key + 8 bytes count) ≈ 580 MB. Fits in memory on one machine for one hour, but write contention and hot keys will melt it.
• Storing every raw event for an hour: 3.6B × ~100 bytes ≈ 360 GB per hour. Writing that to a database at 1M/sec is already not going to work.

I’d say out loud: “Storing raw events and aggregating on read is a non-starter at this rate. Maintaining an exact count per key in a single hash map is technically feasible in memory, but it collapses under the write rate and the hot-key distribution. And the product doesn’t need exactness — nobody cares if a hashtag is ranked 47th or 49th. So the design will evolve from naive toward an approximation with bounded error.”

3. API design

Two very different surfaces. Ingestion is the firehose. Query is dashboard-style, low-QPS, latency-sensitive.

POST /events
  body:    { key, timestamp, metadata{region?, segment?, ...} }
  returns: 202 Accepted

GET /top-k
  params:  window=1h, k=100, segment?
  returns: [{ key, approx_count, rank }, ...]

Two decisions worth calling out:

• Query is cached aggressively. The top-K result for the current window changes slowly (second-to-second). A 1–5 second TTL at the edge is fine and eliminates 99%+ of read traffic.
• Counts are labeled approx_count. This is a correctness contract with the consumer: counts are bounded-error estimates, not ground truth.

4. Data schema

Most state is in-memory in the stream processor. Persistent stores are thin.

sketches_live       (per-shard, per-minute, in-memory)
  shard_id, minute_bucket → CMS + Space-Saving

sketches_persistent (periodic flush to durable storage)
  shard_id, minute_bucket, type → serialized sketch bytes

topk_cache          (query-serving)
  window_spec → top-K result (5-second TTL)

raw_events_archive  (S3, for replay / recomputation)
  hourly partitions of raw events

Storage choices:

• sketches_live — in-memory in the Flink task (Apache Flink — a distributed stream processing framework). Checkpointed to RocksDB every 30 seconds.
• sketches_persistent — Redis or object storage. One serialized sketch per shard per minute. Small, cheap.
• topk_cache — Redis. Query endpoint reads here first; recomputes from sketches only on cache miss or staleness.
• raw_events_archive — S3. Not on the query path. Used for replay, debugging, and reprocessing if aggregation logic changes.

I’d say explicitly: “The expensive part of this system is counting, which lives in-memory. The persistent stores are archives and result caches. Raw events go to S3 but aren’t used for the online top-K — the design choice to keep raw data separately is about future flexibility, not about computing top-K itself.”

5. High-level architecture

flowchart LR
  Client[Client] --> Gateway[Ingestion gateway]
  Gateway --> Kafka[(Kafka · partitioned by key)]
  Kafka --> Flink1[Flink shard 1]
  Kafka --> Flink2[Flink shard 2]
  Kafka --> FlinkN[Flink shard N]
  Flink1 --> B1[60 buckets · CMS + Space-Saving]
  Flink2 --> B2[60 buckets · CMS + Space-Saving]
  FlinkN --> BN[60 buckets · CMS + Space-Saving]
  B1 --> Merger[Query-time merger]
  B2 --> Merger
  BN --> Merger
  Merger --> Cache[(Top-K cache · 5s TTL)]
  Dashboard[Dashboard] --> Cache
  Gateway -.raw events.-> S3[(S3 · raw event archive)]

6. Detailed workflows

Ingest path, step by step

A single item event from a client to the counting structures in every processing shard.

1. Client → Ingestion gateway. The client fires POST /events with {key: "#superbowl", timestamp: 1718000000, metadata: {region: "US"}}. The gateway server-stamps the event (overriding or supplementing client timestamp for bucketing — this prevents clock skew on event sources from corrupting window assignments) and validates the request shape.

2. Ingestion gateway → Kafka. The gateway produces the event to the events Kafka topic. Partition key is the item key (#superbowl), not a random assignment. This ensures all events for a given key land on the same partition, so a single Flink shard owns the count for that key. The gateway returns 202 Accepted once Kafka acknowledges the write.

3. Kafka → Flink consumer shard. Each Flink task consumes one or more Kafka partitions. Because partitioning is by key, each shard receives a deterministic subset of the key space — no coordination between shards is needed for counting. Within a shard, events for a given key arrive serially.

4. Flink shard → CMS update. For each incoming event with key k, the shard hashes k through all d hash functions and increments the corresponding counters in the current minute’s Count-Min Sketch. This is O(d) — constant time regardless of key cardinality. The sketch does not store the key itself, only the counter increments.

5. Flink shard → Space-Saving update — in parallel with step 4, on every event, with no threshold gate. The shard also runs the Space-Saving admission rule on key k against the current minute’s candidate map. CMS and Space-Saving are not pipeline stages: Space-Saving does not wait for the CMS count to cross some threshold before admitting a key. Every event touches both structures independently. If k is in the map, its count increments. If not and the map is full, the minimum-count entry is evicted and replaced with k inheriting min_count + 1.

6. Periodic snapshot → sketches_persistent. Every 30 seconds, the Flink task checkpoints its RocksDB state (which includes all 60 per-minute sketches). Completed minute-buckets are also serialized and written to sketches_persistent (Redis or object storage), making them available to the query merger independently of the live Flink process.

Query path, step by step

A top-K query from the dashboard to a returned ranked list.

1. Client → Query API. The dashboard calls GET /top-k?window=1h&k=100. The Query API first checks the topk_cache in Redis. If a result exists with TTL > 0, it is returned immediately — query path ends here. At 5-second TTL and typical dashboard refresh rates, the vast majority of queries hit the cache.

2. Query API → fetch sketches from all shards. On cache miss, the Query API fetches the 60 per-minute bucket sketches from all N shards. This is a fan-out read: for each of the last 60 minute-buckets, retrieve the CMS and Space-Saving structures from each of the N Flink shards (via sketches_persistent). Total reads: 60 × N sketch pairs.

3. Merge CMS structures. For each of the 60 time buckets, merge the N per-shard CMS structures into a single CMS by element-wise addition of counters (shards use identical d, w, and hash functions, which is a deployment invariant). Then merge the 60 per-minute CMSes into a single 1-hour CMS by the same element-wise addition. The merged sketch is mathematically equivalent to a single sketch that observed all events across all shards over the last hour.

4. Merge Space-Saving structures. Apply the same merge to the Space-Saving structures: sum counts for keys present in multiple structures, keep the top-m by count across the union. Merged error is bounded by the sum of input errors, which remains small given the shard count. The merged Space-Saving gives the top-K candidate set for the 1-hour window.

5. Extract top-K from merged Space-Saving. From the merged candidate set, take the top K entries by stored count. For each candidate key, the final count estimate can be cross-checked against the merged CMS — estimate(key) returns the minimum-row value, which is a tighter upper bound than the Space-Saving stored count alone. In practice, for keys genuinely in the top-K the two estimates are very close.

6. Return sorted result → cache. The Query API returns [{key, approx_count, rank}] sorted descending by count. The result is written to topk_cache with a 5-second TTL. Subsequent queries within that window are served from cache at sub-millisecond latency.

7. Evolving the architecture step by step

This is where SDE II and SDE III answers diverge. I’d walk through four stages, each fixing the previous stage’s bottleneck.

Stage 1: Naive — write every event to a database

flowchart LR
  Client[Client] --> API[API server]
  API --> DB[(Database)]
  Dashboard[Dashboard] --> API
  API -->|ORDER BY count DESC LIMIT 100| DB

Schema: one row per event, or a (key, count) row updated via UPDATE counts SET count = count + 1 WHERE key = ?.

Why this fails:

• 1M writes/sec to a single database is impossible. Even sharded, the write amplification of updating an index on count for every event kills it.
• Hot keys (a viral hashtag) cause lock contention on a single row.
• ORDER BY count LIMIT 100 over millions of rows every few seconds at dashboard load is expensive.
• No natural way to express a sliding window — you’d have to store per-event rows and aggregate on read, which explodes storage.

I’d call this out quickly: “This is the first thing that comes to mind and it breaks at roughly 10K events/sec. Let’s fix the read side first, then the write side.”

Stage 2: Add Redis for counts and top-K

flowchart LR
  Client[Client] --> API[API server]
  API --> Redis[(Redis · ZSet)]
  API -.write raw.-> DB[(Database · audit)]
  Dashboard[Dashboard] --> Cache[(Result cache)]
  Cache --> Redis

  classDef new fill:#eef2f1,stroke:#2c5f5d,stroke-width:2px,color:#1f2937;
  class Redis,Cache new

Use a Redis sorted set (ZSet) — scores are counts, members are hashtag keys. ZINCRBY on write, ZREVRANGE 0 99 for top 100.

What this fixes:

• ZSet is an in-memory skiplist; ZINCRBY is O(log N) and far faster than a DB row update.
• ZREVRANGE gives top-K directly — no ORDER BY scan.
• Raw events still write to a database asynchronously for audit and replay (more on this below).

What still breaks at 1M/sec:

• A single Redis instance tops out around 100–200K ops/sec. We’d need to shard.
• Hot keys still cause contention on a single Redis shard.
• Sliding window is still unsolved. A single ZSet is all-time counts. You could use TTLs, but INCR + EXPIRE on the same key either never expires (if you refresh TTL on every write) or drops all data at once (if you don’t) — neither is a sliding window. What you actually want is: “count only events that arrived in the last hour.” A single counter can’t express that because a counter has no memory of when its increments arrived. You need time-bucketed counts that you can age out independently.
• Memory: one ZSet entry is ~80 bytes. 100M unique keys all-time ≈ 8 GB on a single Redis, which is feasible but uncomfortable. Per-minute buckets for sliding windows multiply this.

Stage 3: Kafka + stream processor, bucketed sliding windows

Now we decouple ingestion from counting and handle the sliding window properly.

flowchart LR
  Client[Client] --> Gateway[Ingestion gateway]
  Gateway --> Kafka[(Kafka · partitioned by key)]
  Kafka --> Flink[Flink · per-shard buckets]
  Flink --> Buckets[(60 per-minute ZSets per shard)]
  Flink --> Merger[Merge 60 buckets · every 10s]
  Merger --> Cache[(Top-K result cache)]
  Dashboard[Dashboard] --> Cache
  Gateway -.raw events.-> S3[(S3 · raw event archive)]

  classDef new fill:#eef2f1,stroke:#2c5f5d,stroke-width:2px,color:#1f2937;
  class Kafka,Flink,Buckets,Merger,S3 new

Key changes:

• Kafka absorbs the 1M/sec write rate. Partition by the counted key so each downstream shard owns a deterministic subset. Within a shard, increments are strictly serial. (Apache Kafka — durable, partitioned event log.)
• Bucketed sliding window. Each shard maintains 60 separate counting structures, one per minute. “Last hour” is the merge of the last 60 buckets. Old buckets age out; new events go to a new bucket each minute. This is the standard sliding-window pattern and solves the TTL problem cleanly.
• Raw events go to S3 for replay, audit, and recomputation. You almost always want this in production — if the aggregation logic changes (e.g., “count only verified users”), you can reprocess history. For the online top-K path, raw events are not read.
• Result cache. Query path reads from a pre-merged top-K cache (updated every 10 seconds by the merger), not from the live buckets directly.

What’s still unsolved:

• Per-shard bucket state is still a hash map of {key → count}. Under hot-key bursts or very high cardinality, memory per shard grows unboundedly. A trending event can push unique-key counts into the hundreds of millions.
• We can bound memory explicitly by using a probabilistic counter.

Stage 4: Count-Min Sketch + Heavy Hitters per bucket

Replace the {key → count} hash map in each bucket with a Count-Min Sketch (for count estimates) plus a Space-Saving structure (for the top-K candidate set).

flowchart LR
  Client[Client] --> Gateway[Ingestion gateway]
  Gateway --> Kafka[(Kafka · partitioned by key)]
  Kafka --> Flink1[Flink shard 1]
  Kafka --> Flink2[Flink shard 2]
  Kafka --> FlinkN[Flink shard N]
  Flink1 --> B1[60 buckets · CMS + Space-Saving]
  Flink2 --> B2[60 buckets · CMS + Space-Saving]
  FlinkN --> BN[60 buckets · CMS + Space-Saving]
  B1 --> Merger[Query-time merger]
  B2 --> Merger
  BN --> Merger
  Merger --> Cache[(Top-K cache · 5s TTL)]
  Dashboard[Dashboard] --> Cache
  Gateway -.raw events.-> S3[(S3 · raw event archive)]

  classDef new fill:#eef2f1,stroke:#2c5f5d,stroke-width:2px,color:#1f2937;
  class B1,B2,BN,Merger new

This is the final design. Memory per bucket drops from “unbounded” to tens of KB, independent of key cardinality. Now I need to explain exactly why CMS and Space-Saving together — neither alone is sufficient.

8. Deep dives

Why CMS alone is not enough

A common misconception: “just use Count-Min Sketch, it counts frequencies.”

CMS answers one question: given a key, estimate its count.

It does not answer “what are the top-K keys?” The sketch is a 2D array of counters with no key list stored — you physically cannot enumerate the keys that went into it. To find top-K with CMS alone, you’d need a separate list of every key you’ve ever seen, and then query the sketch for each. That defeats the purpose.

So you need one of two things:

• A heap of candidate keys updated as events arrive, using CMS for the count. Simple, but has a latency problem: a genuinely new trending key has to accumulate enough estimated count to beat the current Kth element before it enters the heap. If the 100th-place key has count 150,000, a viral new hashtag must reach that count before it shows up — potentially minutes behind reality.
• A Heavy Hitters algorithm like Space-Saving, which directly maintains a bounded-size top candidate list as events arrive, with an admission rule that lets new keys enter quickly.

For a trending system where freshness is the point, Space-Saving is the right choice. CMS is kept separately for the “estimate the count of a specific key” query (e.g., “how many times was #superbowl mentioned?”), which Space-Saving can’t answer for keys outside its top list.

How Count-Min Sketch works

Intuition first. Imagine you want to count frequencies but can’t afford a slot per key. Instead, use a grid of counters where every key hashes to one cell per row. When a key arrives, you tick its cell in every row. To estimate a key’s count, you read all its cells and take the minimum — collisions can only inflate counts, never deflate them, so the minimum is the tightest upper bound.

The formal structure: a 2D array of counters, d rows by w columns, plus d independent hash functions h_1, h_2, ..., h_d, each mapping keys to column indices.

To estimate count("#superbowl"): h₁ maps to col 2 → 5, h₂ maps to col 1 → 4, h₃ maps to col 4 → 4. Estimate = min(5, 4, 4) = 4. (The bolded cells are the ones read for this key.)

On increment(key): for each row i in 1..d, compute h_i(key) and increment counters[i][h_i(key)] by one.

On estimate(key): for each row i, read counters[i][h_i(key)]. Return the minimum of the d values.

The minimum is the tightest upper bound, because every row counts at least as much as the true count (incrementing only ever adds; hash collisions only inflate), so taking the min across rows reduces collision error.

Error bounds: with w = ⌈𝑒/ε⌉ columns (where 𝑒 ≈ 2.718) and d = ⌈ln(1/δ)⌉ rows, the estimate is within εN of the true count (where N is total events) with probability 1-δ.

Worked example. Set ε=0.001 (0.1% error), δ=0.01 (99% confidence). Then w ≈ 2718, d ≈ 5. Memory: 5 × 2718 × 4 bytes = ~54 KB per sketch. One hour at 1M events/sec is 3.6B events, so per-key error is at most ~3.6M — meaning you can’t resolve keys with true count below ~3.6M from each other, but you can resolve anything above it. For trending hashtags where the top-K all have counts in the tens of millions, that’s plenty of resolution.

The state we keep is d × w integers. 54 KB, not 580 MB. And critically, the size doesn’t grow with key cardinality — it’s determined only by the desired error.

How Space-Saving (Heavy Hitters) works

Maintain a fixed-size map of m candidate keys with counts, where m > K (typically m = 10K for top-100).

On increment(key):

• If key is in the map, increment its count.
• Else if the map has fewer than m entries, add key with count 1.
• Else find the minimum-count entry in the map. Replace its key with the new key. Set the new key’s count to min_count + 1. The displaced key’s count is inherited by the new entry — this is the “saving” in Space-Saving.

On query: return the top-K entries by count.

Concrete trace (m=4, stream: A A B A C D D D E):

After the stream: A and D are correctly identified as top-2. E is tracked with a count of 2 (overestimate — its true count is 1). B is gone. The eviction rule ensures new genuinely-trending keys surface fast.

Why inheriting the min count is correct, not cheating

First reaction: “new key only appeared once — why does it get min_count + 1?” That feels like gross overestimation.

The guarantee: the stored count is an upper bound on the true count, bounded above by true_count + min_count. It never underestimates. Two cases:

• New key is genuinely trending. It will keep receiving increments. Its stored count grows from min_count + 1 upward; its true count catches up quickly. Rank is correct.
• New key is a one-hit. It doesn’t get more increments. The next new key will evict it (its count is now the new minimum), and it exits cleanly. It occupied a slot for a short time but didn’t corrupt the top-K.

This admission rule is exactly what lets new trending keys surface quickly — the property CMS + heap doesn’t have.

Mergeability. Two Space-Saving structures covering disjoint streams can be merged: sum counts for shared keys, keep top m by count. The merged error bound is the sum of the input bounds. This is the property that makes sharding and sliding-window bucket merges correct.

Worked example. m = 10K for top-100. At ~50 bytes per entry, that’s ~500 KB per heavy-hitters structure. Per shard, per bucket. Trivial memory.

How CMS and Space-Saving work together

This is the part the article hasn’t said clearly enough: Space-Saving receives every single event, without any threshold or pre-filter from CMS. They are not a pipeline. They are two independent data structures updated in parallel on every event at the same time.

Here is the exact division of labor:

Neither structure alone answers both questions. That’s why you need both.

Why no threshold between them?

The threshold idea feels natural: “only promote a key to Space-Saving once its CMS count crosses some floor.” The problem is that the threshold is relative to a single bucket — one minute of data — and a key that is a genuine global top-K over the full hour may have a low count in any individual minute, especially early in a trending surge.

Concrete example. Suppose #oscars starts trending at minute 45 of the hour. In that one minute it receives 80,000 events — well below the current minimum count in a hot Space-Saving map (say 500,000). Under a threshold rule, it never enters the per-minute candidate set. At merge time, its counts are absent from the Space-Saving structures and it falls out of the top-K result, even though it was clearly trending in the last 15 minutes of the window.

The no-threshold design — admitting every key via the Space-Saving eviction rule — prevents this. A key that starts surging in minute 45 will evict a low-count incumbent in that minute’s Space-Saving map, and will be present at merge time to accumulate across the buckets where it did appear.

How a slow-but-steady key survives across buckets

A different concern: a key that is steady but never dominant in any single bucket — say it ranks 400th per minute but 5th over the hour.

Space-Saving per bucket has m slots (say 10,000 for top-100). Any key that receives at least one event in a minute has a chance of staying in the map as long as its count exceeds the current minimum. A key with steady, moderate volume will hold a slot in most minute-buckets because steady volume keeps its count above the lowest-count entries. At merge time, its counts from all 60 buckets are summed, and it rises to its correct global rank.

The failure case is a key so rare per-minute that it is evicted from every single bucket before merge. With m = 10K for a top-100 query, you would have to be outside the top 10,000 per minute to be evicted from every bucket — which means you cannot possibly be top-100 over the hour either (because even steady presence across 60 buckets can’t sum to top-100 total if you were below 10,000th every minute). So the eviction threshold is self-consistent with the final query goal.

The role of CMS at query time

Once Space-Saving gives the candidate list from the merged top-K, CMS can refine the count estimates. The Space-Saving stored count is an upper bound (slightly inflated by inherited eviction counts); the merged CMS can give a tighter point estimate for each candidate key via min_row(key). In practice the two are very close for keys genuinely in the top-K, and the difference is invisible at the presentation layer.

Sliding windows: bucket mechanics

The window semantics are harder than the counting. Three options:

(a) Tumbling windows. Reset at every hour boundary. Simple, but “last hour” jumps — at 10:59 reflects 59 minutes; at 11:00 reflects 0 minutes. UX is jarring.

(b) Bucketed sliding window. One sketch per minute for the last 60 minutes. “Last hour” = merge of the last 60 per-minute sketches. Each bucket is finalized once and never updated; old buckets age out. Standard answer.

(c) Exponential decay. Each event contributes with a decay factor. No explicit window — recent events count more. Elegant, but harder to reason about time boundaries.

Go with (b):

• 60 buckets per shard, one per minute.
• Each bucket is its own CMS + Space-Saving pair.
• Query-time: merge the 60 buckets across all shards. CMS merge is column-wise addition; Space-Saving merge is the property above.
• Age-out: at the start of each minute, the oldest bucket is dropped, a new one is started.

Merge cost: 60 buckets × N shards. At 64 shards and 60 buckets, that’s 3,840 sketch merges per query — milliseconds at most, because each merge is a small fixed-size addition. Cached for 5 seconds, so query cost amortizes to near-zero.

Distributed counting and merge correctness

For 1M events/sec peak, one machine can’t maintain the live sketches — partitioning by key spreads load across N Flink tasks.

The merge correctness property is where the design earns its keep:

• CMS merge. Given two sketches with identical d, w, and hash functions, merge is element-wise addition. The merged sketch is mathematically equivalent to a single sketch that observed both streams. Error bounds compose additively.
• Space-Saving merge. Sum counts for shared keys, keep top m by count. Merged error is bounded by the sum of input errors. See the Mergeable Summaries paper for the formal treatment.

Because both structures merge correctly, sharding is free at the algorithm level. Only cost is query-time merge, which is bounded and cached.

Hot-key mitigation

A single hashtag going viral during the Super Bowl sends 10% of events to one Kafka partition. That Flink task becomes the bottleneck. Mitigation: salt hot keys. Detect keys in the top 0.1% of recent volume (a small CMS at the gateway is enough) and append a random suffix 0..N to their partition key. A second merge stage keyed on the unsalted key combines the N salted buckets at query time.

9. Failure modes

• Flink task crash. Restart from last checkpoint (every 30s). RocksDB state backend persists sketches; replay of ~30s of Kafka reconstructs in-memory state. Small count drift during replay; well within error bars.
• Kafka partition unavailable. With acks=all and min.insync.replicas=2, single broker loss is invisible. Two broker losses on same partition stall writes; gateway rejects; SDK retries with backoff.
• Sketch memory overflow. Can’t happen by construction — sketches are fixed-size. This is precisely why we chose them. The approximation is not “nice-to-have”; it’s the reason memory is bounded under adversarial input.
• Query merger overloaded. Pre-merge in background every 10s, cache the result. User queries hit cache, not merger.
• Clock skew on event timestamps. Server-stamp at the ingestion gateway; use server time for bucketing, not client timestamp.
• Hot-key melt without salting. Detection: per-task backlog metric. Mitigation: enable salting above a volume threshold via config flip.
• Heavy-hitters eviction of genuine top-K. If m is too close to K, a genuine top-K key can get evicted during a burst. Mitigation: set m generously (10–100× K); monitor displaced-key distribution.

Pattern to notice: name what fails, what degrades gracefully, what doesn’t. Because the product tolerates small error, almost everything here degrades to “slightly less accurate top-K” rather than “no top-K.” That’s a property of the approximation choice, not an accident.

10. What I’d skip, and say I’m skipping

Explicit defers:

• Spam / bot filtering. Separate upstream system.
• Per-tenant isolation for multi-tenant APIs.
• Historical archival / backfill from raw event archive — possible via the S3 events, but a separate pipeline.
• Alerting on trending changes (“this hashtag just jumped 10×”) — a consumer of the top-K stream, not part of this system.
• Tradeoffs considered and rejected:
  • Exact counting with sharded Redis ZSets. Works up to ~100K events/sec before hot-key contention destroys it.
  • Batch-only (nightly Spark job). Fails freshness SLA by two orders of magnitude.
  • CMS with a heap instead of Space-Saving. Works, but new trending keys have to accumulate past the Kth-place count before entering the heap — Space-Saving’s admission rule fixes this.

11. Wrap-up

One crisp sentence:

The design evolves from a DB to a cache to a stream to a probabilistic stream, with each stage fixing the previous stage’s bottleneck. The final trade-off is exactness for tractability: top-K is approximate with bounded error, but it holds the bound under 1M events/sec, a sliding hour window, and arbitrary sharding.

12. Follow-up variations

Interviewers often pivot to one of these after the main design. Each has a short answer grounded in what you’ve already built.

Small scale: what changes when the numbers are smaller?

The full CMS + Kafka + Flink design is justified by 1M events/sec and the memory problem that comes with it. If the interviewer’s numbers are smaller, the right answer is a simpler one — and saying so is the signal.

The pattern: approximation is the last tool you reach for, not the first. Use it when exact counting is provably infeasible at the given scale. If someone asks “why not just use a Redis ZSet?” at 1M/sec, the answer is: at that rate a single ZSet saturates in seconds and a sharded exact map runs out of memory at high-cardinality bursts — that’s the tipping point.

Segments: top-K per category, region, or tag

A segment is any dimension you want to slice top-K by: language, region, topic category, user cohort. The follow-up is usually: “How would you support top-100 trending hashtags per country?”

The answer: independent sketch sets per segment, using a compound partition key.

Change the Kafka partition key from item to (segment, item). Each Flink shard now owns a deterministic subset of (segment, item) pairs. Each shard maintains one CMS + Space-Saving pair per segment per time bucket instead of one global pair per bucket. Query GET /top-k?segment=US&window=1h merges only the US-segment sketches across all shards.

Memory cost: num_segments × num_buckets × sketch_size. With 200 countries × 60 buckets × ~600 KB (CMS + Space-Saving): ~7 GB total across all shards. Spread over 64 shards, that’s ~110 MB per shard — manageable.

What does NOT work: a single global sketch with post-hoc filtering. A CMS has no notion of a segment label per counter — you cannot ask “what is the count of key X within segment US?” on a sketch that merged all regions. You must maintain separate sketches if you want per-segment answers.

Dynamic segments (e.g., user-defined tags) add a complication: you cannot pre-partition for tags you don’t know yet. The practical answer is a soft cap on active segments (e.g., only maintain sketches for the top 1000 segments by event volume, routed by a lightweight CMS at the gateway that tracks segment activity).

Global replication: serving top-K worldwide

Two different questions hide under “global”:

(a) One global top-K, served everywhere fast. The result set is tiny — 100 key-count pairs. Compute it in one region (or a designated aggregator), then replicate to a Redis node in each region with a 5-second TTL. Edge CDN caches serve the result from the nearest PoP. Latency to the query is sub-millisecond; freshness lag is the replication round-trip (typically < 1s within a region, 1–3s cross-continent).

(b) Per-region top-K. Treat region as a segment (see above). Each region runs its own pipeline independently. No cross-region merge needed. A user in Japan gets top-k?segment=JP, served from the Japanese pipeline. Simpler operationally; the only coordination is the query routing layer.

(c) Global top-K computed from per-region streams. More complex: each region produces per-minute sketches, a global aggregator merges them, and the merged result is replicated everywhere. The mergeability property you already use for sharding is exactly the property that makes this correct — merging two CMS/Space-Saving structures from different regions produces the same result as a single structure that observed both streams. The cost is merge latency: if regions produce a snapshot every 30s and the global merge takes 5s, freshness is ~35s per cycle. Acceptable for most trending surfaces.

The answer to reach for in an interview: option (a) for a global trending feed (one answer, cheap to distribute), option (b) for per-region feeds (simple, independent, no cross-region coupling).

Other common follow-up questions

Multiple simultaneous time windows (last hour, last day, last week)

The bucketed design already handles this. Per-minute buckets support any window that’s a multiple of minutes: “last hour” = merge last 60, “last 6 hours” = merge last 360. For longer windows (daily, weekly), add a second tier of per-hour buckets — each hour-bucket is itself a merged summary of 60 per-minute buckets. Query time specifies which tier to merge. Memory cost scales linearly with the number of tiers; sketch sizes are fixed.

Weighted events (not all events are equal)

CMS handles weighted increments natively: instead of increment(key) by 1, call increment(key, weight). Space-Saving needs a matching change: evict the entry with the lowest weighted count; inherit that weighted minimum. A verified-user action weighted at 2 and a bot action at 0 (i.e., filtered upstream) work correctly without any structural change to the sketches.

Exact counts for specific keys (compliance, billing)

The approximate online pipeline cannot be made exact after the fact — CMS is lossy by design. If the business later requires exact counts for a subset of keys (e.g., hashtags in election reporting), the answer is a parallel batch pipeline over the raw event archive in S3. The online pipeline stays approximate; the batch pipeline runs Spark/Hive jobs over immutable hourly partitions and produces exact counts on demand. The two pipelines coexist, serving different consumers. This is why archiving raw events is load-bearing, not optional: it is the only path to correctness for compliance cases.

Trending alerts: detecting sudden spikes

The current system answers “what are the top-K now?” but not “what just jumped 10×?”. A trending alert is a consumer of the top-K stream: compare the current top-K result to the snapshot from N minutes ago, compute rank-change or count-growth-rate per key, and fire an alert if the delta exceeds a threshold. This is a read-only layer on top of the existing output — no changes to the counting pipeline itself. The hard part is defining “sudden”: a fixed count-growth threshold will false-positive during expected peak hours. A relative threshold (growth vs the same hour last week) is more robust.

Exact top-K with a small key space

If the interviewer says “there are only 500 possible keys” (e.g., top products in a catalog), the entire approximation story is moot. A single Redis ZSet with exact counts handles it trivially, with a sliding window built from per-minute ZSets. The CMS + Space-Saving design is overkill and you should say so.

What separates SDE II from SDE III on this question

• SDE II usually lands a sharded counter, a sliding window, and reaches for approximation when prompted on memory.
• SDE III walks the evolution explicitly (DB → Redis → Kafka + stream → probabilistic), names approximation in the first few minutes, explains why CMS alone is insufficient, pairs CMS with Space-Saving for direct top-K maintenance, articulates the mergeability property as the reason sharding is free, picks bucketed sliding windows with merge math, and names hot-key salting as a real production concern.
• Staff/Principal surfaces three things SDE III typically doesn’t:
  • Product contract. “What does ‘approximate’ mean to the business? If the true top-1 trending hashtag appears as rank-3 for 30 seconds, is that acceptable?” The answer changes the error budget and may change the architecture.
  • Silent operational risk. Heavy-hitters eviction produces no error signal — if a genuine top-K item gets evicted during a burst, the result is quietly wrong. Requires explicit monitoring (track eviction rate; alert when displaced-key distribution shifts).
  • Long-term irreversibility. If the system later needs exact counts for compliance (e.g., election-related hashtag reporting), the CMS-based online path can’t be retrofitted. The raw-event archive in S3 is the escape hatch — but only if it was built in from day one.

The differentiator isn’t tool knowledge. It’s whether you can walk a design from naive to correct, naming at each step what breaks and what fixes it, and whether you recognize that “top-K at scale” is fundamentally an approximation problem.

Further reading

• Count-Min Sketch paper (Cormode & Muthukrishnan, 2005) — the original paper. Short and readable.
• Space-Saving paper (Metwally, Agrawal, El Abbadi, 2005) — the original Heavy Hitters algorithm. Mergeability is explicit.
• Mergeable Summaries (Agarwal et al., 2012) — formal treatment of why these structures compose correctly under sharding.
• Designing Data-Intensive Applications — Kleppmann. Chapter 11 on streaming is the textbook reference.
• Apache Flink: windowing documentation — mechanics of tumbling vs sliding windows in the engine most production implementations use.

Related on calm.rocks:

• Walkthrough: Designing an Ad Click Aggregation System — same streaming architecture, different aggregation, different correctness story. Hot-key salting discussed in full.
• Reference: Cache Access and Invalidation Patterns — the top-K query cache is a canonical short-TTL read-through cache.