Top-K Heavy Hitters

What is a top-K heavy hitters system?

A top-K heavy hitters system identifies the K items seen most frequently in a stream of events, such as the top 10 trending hashtags on Twitter or the top 100 searched products on Amazon. The core loop (count everything, rank it, return the top K) sounds simple. The hard part is two things: counting accurately at millions of events per second without blowing out memory, and merging local top-K lists across dozens of distributed stream processors into a single globally correct ranking without re-processing history.

The key design question is how to combine bounded-memory counts from many processors into one useful global ranking without shipping every distinct item or reprocessing the full event history.

TL;DR

Accept item events into a durable log, aggregate them with stateful stream processors using bounded-memory Count-Min Sketches plus a top-K candidate structure, merge compatible sketches across partitions, and publish precomputed Redis sorted-set snapshots every 30 seconds. Reads become a single rank lookup even though the write path is much heavier.

Use a ring of time buckets to support several windows without replaying history, make consumers replayable and idempotent, and expose snapshot age and approximate counts in the API. Exact audit counting, personalization, and metadata enrichment are separate concerns.

Scope and assumptions

These are illustrative planning assumptions for the design; they are not capacity guarantees:

• The system ingests about 1 billion events per day, roughly 11,500 events/second sustained and 35,000 events/second at peak.
• It supports last_1m, last_1h, and last_24h windows, with an optional longer window such as seven days, and K up to 100. Rankings may be up to 30 seconds stale.
• Item IDs are normalized before counting. At-least-once delivery is acceptable when the event contract includes a stable event ID or the product accepts small duplicate-count error.
• Approximate counts are acceptable for a user-facing ranking, provided the sketch parameters and error behavior are documented. A separate raw-event or audit sink is needed for exact reconciliation.
• The service owns event acceptance, counting, ranking, and snapshot reads. Authentication, catalog metadata, personalized ranking, and exact per-session deduplication are outside the primary design.

Functional Requirements

Core Requirements

1. Users can submit item events (for example, a search query string, a product view, a hashtag click) to the system.
2. Users can query the top K items for a given time window (for example, top 10 in the last hour, top 100 in the last 24 hours).
3. Rankings update within 30 seconds of new events arriving.
4. Multiple time windows are supported simultaneously (last 1 minute, 1 hour, 24 hours) without re-processing history.

Below the Line (out of scope)

• User authentication and per-user personalization
• Item metadata enrichment (images, display names)
• Exact deduplication of duplicate events from the same user session

The hardest part in scope: Merging distributed top-K results correctly. Naively unioning the top-K list from each stream processor is wrong. An item ranked 11th on every individual node can be globally first if its traffic is spread evenly across partitions, and you miss it entirely. The solution (merging Count-Min Sketches) is the deepest technical challenge in this system, and merging Count-Min Sketches is the subject of Deep Dive 3.

User authentication is below the line because it does not change the counting or ranking path. To add it, attach a user_id to each event but route the event through the same Kafka topic and count pipeline unchanged.

Item metadata enrichment is below the line because it belongs in a separate lookup service. The ranking layer stores only item IDs and approximate counts. Display names and images are joined at the API response layer using a separate catalog service.

Exact deduplication is below the line because it requires a per-user Bloom filter or a Redis SET of (user_session, item_id) pairs, which adds significant storage complexity. Approximate deduplication is acceptable for trending signals: if a user clicks the same product 20 times in a session, the trend still reflects genuine engagement.

Non-Functional Requirements

Core Requirements

• Throughput: Handle 1 billion events per day (roughly 11,500 events per second at steady state, with 3x peak burst = 35,000 events per second).
• Staleness: Rankings are stale by at most 30 seconds. Sub-second freshness is not required; 30-second staleness is imperceptible for trending content.
• Memory: Memory per time window is bounded regardless of item cardinality. At 1 million distinct items per day, exact counting needs 8 MB per window. Count-Min Sketch reduces this to under 200 KB per window.
• Availability: 99.9% uptime. A brief gap in ranking freshness is acceptable; returning stale results from a pre-computed snapshot is always preferable to a total outage.
• Query latency: Top-K reads return in under 10 ms p99. Rankings are pre-computed, so reads resolve to a single Redis ZREVRANGE command.

Below the Line

• Exact (non-approximate) counts for all items (audit-grade counting requires storing every event individually, which does not bound memory)
• Sub-second ranking latency (requires a hot path with no batching window, approaching the Redis throughput ceiling at 35K events/second)

Read/write ratio: 1 billion events per day vs roughly 100,000 top-K queries per day gives a 10,000:1 write-to-read ratio. This is an extremely write-heavy system. The write path is the entire design challenge: ingestion throughput, stream aggregation, and memory-bounded counting all exist to handle this asymmetry. The read path is trivial by comparison, because pre-computed snapshots reduce every query to a Redis ZREVRANGE.

The 10,000:1 ratio tells you immediately that the system must decouple writes from reads. You cannot compute rankings synchronously on every write event, and you do not need to: a 30-second staleness window gives you time to batch-aggregate and snapshot asynchronously.

Write this ratio down early: it rules out computing rankings synchronously on every incoming event and points toward cheap write absorption plus snapshot-based reads.

30-second answer / outline

1. Validate and publish each item event to Kafka, partitioned for scalable consumption, and acknowledge it once the event is durably accepted.
2. Have stateful stream processors maintain a ring of 30-second Count-Min Sketch slots and a bounded top-K candidate set for each supported window.
3. Every 30 seconds, merge compatible sketches across processors, estimate candidate frequencies, and write the resulting rankings to Redis sorted-set snapshots.
4. Serve GET /top-k with a single ZREVRANGE, returning the snapshot age and approximation metadata so callers understand freshness and accuracy.
5. Persist processor state and Kafka offsets so a failed processor can replay or restore its windows; serve the last good snapshot while a new one is being built.

5-minute explanation

Start with the write/read asymmetry: billions of events arrive continuously, while ranking reads are comparatively sparse. A synchronous database aggregation or a Redis write for every distinct item would couple user traffic to the hottest part of the system. Kafka provides a durable write-ahead buffer, and stream processors absorb the work asynchronously.

Each processor keeps bounded state rather than an unbounded hash map. A Count-Min Sketch estimates frequencies with a known overcount bound, while a Space-Saving-style candidate structure gives the system items to evaluate because a sketch cannot enumerate every item by itself. All processors use the same hash configuration, so their matrices can be added element by element to form a global estimate. The snapshot builder then extracts the top K from those candidates.

Time windows use a ring of short tumbling slots. A one-hour ranking merges the trailing 120 thirty-second slots; a 24-hour ranking merges 2,880. Slots are overwritten as they expire, so memory stays bounded and a new window does not require a historical replay. The merge and snapshot work runs off the query path every 30 seconds.

The query service reads only the current Redis snapshot. It should return snapshot_age_seconds, window, K, and whether counts are approximate. If a processor, merge job, or Redis replica is unavailable, the service can return the last valid snapshot or a clearly marked stale result while Kafka and state stores recover. The main correctness boundary is the merge: unioning local top-K lists is not sufficient because a globally important item can rank below K on every individual partition.

45-minute interview approach

This is a time-boxed agenda for answering the design question, with the distributed counting and merge problem receiving the most attention.

• 0–5 minutes — Clarify the contract: Confirm what an item and event mean, supported windows, K limits, freshness, exact versus approximate ranking, duplicate semantics, metadata needs, and whether the result is global or tenant-scoped.
• 5–10 minutes — Establish scale: Calculate event rate, peak burst, distinct-item cardinality, query rate, memory per processor, snapshot frequency, and the write/read asymmetry.
• 10–15 minutes — Define entities and APIs: Walk through Event, TimeWindow, TopKSnapshot, stable event IDs, POST /events, GET /top-k, validation, and freshness fields.
• 15–22 minutes — Draw ingestion: Show the stateless API, Kafka partitions, producer acknowledgement, processor ownership, backpressure, and replay boundary. Briefly show why direct database writes fail.
• 22–31 minutes — Deep dive on counting and merge: Prioritize CMS plus candidate tracking, common hash parameters, element-wise sketch merge, error bounds, hot items, and why local-list union is wrong.
• 31–36 minutes — Add windows and reads: Draw the ring buffer, snapshot builder, Redis sorted sets, query cache/read replicas if needed, and the 30-second freshness contract.
• 36–41 minutes — Reliability, security, and operations: Cover state restoration, duplicate events, late timestamps, stale snapshots, access control, rate limits, PII, lag, and replay/backfill.
• 41–45 minutes — Trade-offs and close: Compare exact maps, Redis ZINCRBY, streaming frameworks, batch processing, and true sliding windows; recap the bounded-memory merge invariant and invite follow-ups.

Core Entities

• Event: A single occurrence of an item being counted (a search query, a product view, a trending signal). Carries item_id, event_type, and timestamp.
• Counter: The running approximate frequency for an item_id within a specific time window. Stored in-memory inside stream processors as a Count-Min Sketch structure, not as individual key-value rows.
• TopKSnapshot: A pre-computed, point-in-time list of the top K items and their approximate counts for a given time window. Stored in Redis as a sorted set and refreshed every 30 seconds.
• TimeWindow: A named time boundary (for example, last_1m, last_1h, last_24h) that scopes a TopKSnapshot. The system maintains one active TopKSnapshot per TimeWindow.

Schema details for the snapshot store (sorted set key naming, TTL policy, serialization format) are deferred to the deep dives. The four entities above are sufficient to drive the API design and High-Level Design.

API Design

The system exposes two endpoints: one to ingest events and one to query rankings. Events are ingested asynchronously because the write volume (35K events/sec at peak) makes synchronous counting impractical at the app server layer.

FR 1 and FR 3 - Ingest an event:

POST /events
Body: {
  item_id: "product:abc123",
  event_type: "view",
  timestamp: 1711670400
}
Response: HTTP 202 Accepted
Body: { message: "Event queued for processing" }

202 instead of 200: the event is not counted the moment the POST completes. It enters a Kafka queue and is counted asynchronously by a stream processor. Returning 200 would imply the event was counted immediately, which is false and misleading to callers. 202 accurately signals acceptance for later processing.

FR 2 - Query top K for a time window:

GET /top-k?window=last_1h&k=10
Response: {
  window: "last_1h",
  k: 10,
  snapshot_age_seconds: 14,
  items: [
    { rank: 1, item_id: "product:abc123", approximate_count: 94821 },
    { rank: 2, item_id: "hashtag:worldcup", approximate_count: 87443 }
  ]
}

The response includes snapshot_age_seconds so callers know exactly how stale the pre-computed result is. This transparency lets downstream services decide whether to display the result immediately or wait for a fresher snapshot before refreshing a high-visibility leaderboard.

High-Level Design

The critical flows are separated deliberately: the write path durably accepts events and updates bounded state, the snapshot path merges and ranks off the query path, and the read path serves an already-computed result.

1. Event ingestion at scale

Direct database writes saturate the primary at 50K events per second, making a write queue the first non-negotiable addition to the system.

The simplest design routes every event from the app server directly to a relational database. At 35,000 events per second at peak, a single Postgres instance cannot sustain the write rate without growing queue depth unboundedly (typical ceiling: 10,000-20,000 simple writes per second before latency degrades). The fix is a Kafka topic that absorbs burst writes and lets stream processors consume at their own pace.

Components:

• Client: Web or mobile client that generates item events (product views, search queries, clicks).
• App Server: Accepts POST /events, validates the payload, and publishes the event to Kafka. Does not perform any counting.
• Kafka: Durable event queue that absorbs write bursts. Decouples the ingestion rate from the stream processing rate. Events are retained for 24 hours to support replay after processor failure.
• Stream Processor Fleet: A horizontally scalable fleet (Kafka Streams, Flink, or Spark Streaming) that consumes from Kafka and maintains in-memory frequency counts. Each processor owns one or more Kafka partitions.
• Redis Counter Store: Receives aggregated item counts flushed from stream processors every 30 seconds. Stores counts as Redis hashes keyed by time window and item ID.

Request walkthrough:

1. Client sends POST /events with item_id and event_type.
2. App Server validates the payload and publishes to the Kafka topic item-events, using item_id as the partition key (so all events for the same item always go to the same processor).
3. Kafka durably persists the event.
4. Each stream processor reads from its assigned partitions and increments an in-memory count for the item_id.
5. Every 30 seconds, each processor flushes aggregated counts to Redis Counter Store using HINCRBY batches.

This diagram covers the write path only. The query path and ranking pre-computation come in the next section.