Design Top K Trending: Stream Counting, Count-Min Sketch and Sliding Windows

Listen

Listen first. We ingest view events, each with an item, a user, a time and a region. Clients ask for the top K items, with K up to a thousand, over the last minute, hour or day, globally or for one region. Personalised ranking and recommendations are out of scope. This is about counting, at scale.

Now the qualities, and ask about accuracy early, because it changes everything. The list can be a minute behind. The top items must be correct, but small errors near rank one thousand are fine. However, if daily counts feed creator payouts, those numbers must be exact, even if they arrive an hour later. That split will give us two paths.

Ask scale

Scale. Ten billion view events a day is a hundred thousand a second, three hundred thousand at peak, and a terabyte of raw data daily. About a hundred million distinct items get views each day. An exact count map is around five gigabytes per window, per region. Reads are only a few thousand a second, and everyone asks the same question, so they cache perfectly.

10B view events/day
  10B / 10⁵ s = 100,000 events/s
  peak ×3 ≈ 300,000 events/s
event ≈ 100 B → 1 TB/day raw

distinct items per day ≈ 100M
exact map: 100M × ~50 B ≈ 5 GB
  per window, per region

reads: 5,000 req/s, same answers

Define API

The API is small. Ingest is internal and batched, a few hundred events per call, so we are not paying for one request per view. The read endpoint takes a window, K and a region, and returns ranked items with counts and an as of timestamp, so clients can show how fresh the list is.

internal ingest (batched):
POST /v1/events
  [{ item_id, user_id, ts, region }]

GET /v1/trending
  ?window=1h&k=100&region=IN
→ { as_of, items: [
    { item_id, count, rank } ] }

as_of tells clients how fresh

Data model

Data model. Raw events go to object storage as Parquet files partitioned by hour: cheap, and the source of truth for replays. The stream keeps per-minute counts as working state. Results are tiny: for each window, region and time, a ranked list of a thousand items in a key-value store. Exact daily counts land in a warehouse table.

raw events → object storage
  Parquet, partitioned by hour

per-minute counts (stream state)
  (minute, item_id) → count

top-K store (Redis / KV)
  key topk:1h:IN:2026-10-24T14:05
  value [(item, count) × 1,000]

exact daily counts → warehouse

High-level design

The naive design: one server with a map from item to count and a min heap of size K. It breaks fast. Three hundred thousand updates a second is too much for one machine, the maps are gigabytes per window, a crash loses all state, and a sliding hour means re-summing everything every minute. So we distribute it.

The real design. Ingest writes events to Kafka keyed by item id, so every view of one item lands on the same partition. A stream job, Flink for example, counts per partition per minute and keeps a local top thousand. A merger combines all the local lists into a global top thousand per window and writes it to the store. The API just reads it.

This detail wins points. The merge is exact because of the partition key. Each item is counted in exactly one place, so its local count is its full count, and the global top K must appear in some local top K. If events were spread randomly, an item could be number one thousand and one everywhere yet first overall, and the merge would miss it.

Deep dive: hot items

Deep dive one: hot items. Keying by item makes a viral video a hot partition. If one item gets eighty thousand views a second, partition twelve does nearly all the work while sixty-three others idle, and its consumer lag climbs. The very thing we want to measure, a spike, is what breaks the pipeline.

The fix is two-stage aggregation. Each ingest server sums its own events for about a second and sends item plus count instead of raw events. Eighty thousand views a second across twenty ingest servers becomes twenty small messages a second. The keyed counters then add partial sums. Raw events still go to the lake for the exact path.

Deep dive: windows

Deep dive two: sliding windows. Count in tumbling one-minute buckets. The last hour is the sum of the last sixty. Each minute, add the newest bucket and subtract the one that just expired, so the rolling total updates in one pass. The day uses twenty-four hour buckets. Crucially, keep per-item counts for each bucket, not just each bucket's top list.

Why keep full counts? Because rankings do not add up. An item that sits just outside the top thousand every single minute can be in the top hundred for the hour, yet it never appears in any minute's list. So windows add counts first, and only then rank. Ranking is the last step, never the thing you sum.

tumbling 1-min buckets per item

last hour at 14:05 =
  buckets 13:06 … 14:05 (60)

every minute:
  hour[item] += new bucket
  hour[item] −= bucket from 61 ago

day: 24 one-hour buckets
keep counts, not just top K

Deep dive: memory

Deep dive three: memory. If a hundred million items per bucket is too much, use a count-min sketch. Four rows of about a million counters, sixteen megabytes, no matter how many items. Each event increments one counter per row. The estimate is the minimum of the four. It can only overcount, and by a bounded amount. Keep a small heap of the current top candidates beside it.

Sketches combine well. Two count-min sketches of the same shape add cell by cell, so the minute-bucket window trick still works. The weaknesses: small items look a bit bigger than they are, and a sketch cannot list its items, so keep a heap of candidates. Another option is the Space-Saving algorithm, which tracks a fixed set of counters and replaces the smallest.

d = 4 rows × w = 2^20 counters
4 bytes each → 16 MB, any # items

add(x): row i: C[i][hash_i(x)]++
est(x): min over the 4 rows

only overcounts, by at most
  ≈ 2.7 × N / w (high prob.)
N = 10B/day → at most +26,000

pair with a heap of K candidates

Deep dive: accuracy

Now the accuracy split from the requirements. The stream path is fast and approximate, good enough for a trending page. A batch job runs every hour over the raw events in the lake, counts exactly with deduplication, and overwrites the stored lists for closed hours. Payouts read only the exact numbers. This fast-plus-exact pairing is often called the lambda architecture.

One more question to ask: does trending mean most viewed, or rising fastest? Raw counts show the same giant items every day. Trending usually compares the last hour to that item's normal rate over the past week, with smoothing so a jump from one view to four is ignored. And count each user once per item per window, so bots cannot buy a spot.

Trade-offs

Failure handling. Stream counters checkpoint their state, so a dead node restores and replays Kafka from the saved offset. If the merger is late, the API keeps serving the previous list and the as of field says so. Monitor consumer lag, partition skew, end-to-end freshness, and the gap between stream and batch counts, which tells you how wrong the fast path is.

At ten times the traffic, add Kafka partitions, pre-combine for a few seconds instead of one, and switch minute buckets to sketches. For many regions, compute each region in parallel and build the global list from regional counts, not regional rankings, for the same reason as before. Reads stay trivial: the answer can sit in a CDN for thirty seconds.

Say it like this

In the interview: partition counts by item, so merging local top K lists is exact. Pre-combine at ingest so viral items do not create hot partitions. Build windows from one-minute buckets, add counts and only then rank. Use a count-min sketch and a candidate heap when memory is tight. Run a fast stream path for the page and an exact batch path for anything that touches money.