Engineering Atlas › System Design Interviews

Design a top-K trending service

20 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

Requirements

Functional

Non-functional

Out of scope: personalised ranking, spam and bot filtering (assume an upstream filter), and arbitrary ad hoc time ranges.

Back-of-envelope

API

GET /v1/top?window=1h&scope=country:BR&k=100
  -> 200 {
       "window": "1h",
       "scope": "country:BR",
       "as_of": "2026-10-15T12:34:00Z",
       "exact": false,
       "error_bound": 15000,
       "items": [{"item_id": "v_8842", "count": 1294021}, ...]
     }

POST /v1/events   (internal, batched)
  {"events": [{"item_id": "v_8842", "ts": 1760531640123, "country": "BR", "category": 7}, ...]}
  -> 202

exact and error_bound are part of the contract, so a product team knows whether a number can be shown on a public chart.

Data model

Kafka topic views, keyed by item_id. Kafka's default partitioner hashes the key, so every event for an item goes to the same partition. This is the decision that makes exact merging possible.

Counter state per shard (in the stream processor, checkpointed):

Snapshot store (Redis or a replicated key-value store): key topk:{window}:{scope}:{bucket_start}, value a compact sorted list of (item_id, count) plus metadata. The key includes the bucket, so a reader can ask for the previous minute and writers never overwrite a list a reader is paging through.

Batch tables (warehouse): daily_item_counts(dt, scope, item_id, views), partitioned by dt and clustered by scope. The all-time table is item_totals(item_id, views) updated by adding each day's partition.

High-level design

flowchart LR
  C[Clients] --> I["Ingest API (1s combiner)"]
  I -->|"key = item_id"| K[("Kafka views")]
  K --> W["Counter shards: minute buckets, sketch + heap"]
  W -->|"per-shard top K"| P[("Kafka topk.partials")]
  P --> M["Merger"]
  M --> S[("Snapshot store")]
  K --> A["Archiver"] --> L[("Object store")]
  L --> B["Daily exact batch"]
  B --> S
  Q["Query API"] --> S
  R[Reader] --> Q

Write path. Clients send view events to the ingest tier. Each ingest node sums events per (item, scope) for one second and emits one record with a count. Records go to Kafka keyed by item ID. Counter shards consume their partitions, add counts into the current minute bucket, update sketches for the per-scope windows, and keep a heap of candidates. When a minute closes (by event time, with a few seconds of grace), each shard emits its local top K for each window and scope to topk.partials.

Merge path. The merger collects one partial list per shard per window. Because items are partitioned, no item appears in two lists, so the global top K is the top K of the concatenation. It writes the snapshot and moves an alias such as topk:1h:global:latest to it.

Read path. The query API reads the latest snapshot for the window and scope, slices the first k entries and returns them. A local in-process cache with a 1-second TTL absorbs the hottest keys.

Batch path. Nightly, a Spark job counts the archived day exactly per scope and writes the daily and all-time tables and snapshots, which replace the streaming daily list.

Deep dives

1. Exact counting versus sketches

Exact counting with a hash map is the simplest correct answer, and the arithmetic says it fits for the global scope (about 90 MB per shard per hour). It stops fitting when scopes multiply: 6,000 country and category combinations each with their own long tail.

For those, each shard keeps a count-min sketch and a min-heap of size K. On each event the shard adds to the sketch, reads the item's estimate (the minimum over d rows), and offers it to the heap: if the item is already a candidate, update it; if the heap has room, insert; otherwise replace the heap minimum only if the new estimate is larger. The sketch never underestimates, so a true heavy hitter cannot be missed because of the sketch. It can only be overestimated by at most ε·N with high probability.

I also considered Space-Saving (Metwally et al.), which keeps exactly m counters and, when full, evicts the smallest and lets the newcomer inherit its count. It has a tighter guarantee per byte for top K and no hashing. A common belief is that it cannot be merged; in fact Space-Saving and Misra-Gries summaries are equivalent and both merge with no loss in their error bound (Agarwal et al., PODS 2012), which is why Apache DataSketches ships a mergeable frequent-items sketch (DataSketches). What these counter summaries cannot do is subtract: you cannot remove the minute that just slid out of the hour. A plain count-min sketch is linear, so the expiring minute's counter array can be subtracted element-wise, which suits a sliding hour built from minute buckets. Redis's built-in Top-K is a third option: it uses HeavyKeeper, which decays the counts of small flows so heavy ones keep their slots, plus a min-heap of K items (Redis docs). Since Redis 8 it ships in core Redis rather than as a separate Redis Stack module (Redis). I chose exact maps for global scopes and sketch plus heap for per-scope windows, with the batch job as the exact backstop.

2. Windows and the merge

A sliding hour cannot be computed by merging 60 per-minute top K lists. An item that ranks 1,001st in every minute could be in the hourly top 1,000, and it would appear in none of the minute lists. So each shard keeps the full minute counts for the last 60 minutes and maintains hour_running incrementally: add the new minute's map, subtract the minute that falls out. Its heap is rebuilt from hour_running in O(n log K) once per minute, which for 1.5M items per shard is tens of milliseconds.

Across shards, the merge is exact only because of item partitioning. If events were spread by a random key, an item's count would be split across shards, and the top K of per-shard top Ks is wrong. The fix there is either to re-key by item before counting, or to ask every shard for its top K' (K' several times K) and accept an approximation. The day window is not slid in the stream at all. It is computed by the batch job and, for freshness, approximated in the stream by summing hourly sketches.

3. Viral items, skew and serving

Because partitioning is by item, a viral video sends all its events to one partition. At 350k events/s, an item with 5% share is 17.5k events/s, which one consumer handles. At 30% share during a live event it is 100k/s, and that partition lags while the rest are idle. The one-second combiner in the ingest tier fixes this: with 200 ingest nodes, the hottest item produces at most 200 records per second into Kafka, regardless of its view rate.

Serving is easy once results are precomputed: snapshots are kilobytes, held in the query API's memory and refreshed when the alias moves.

Variant: A scalable like counter

A like counter looks like top-K with K = 1, but the requirement flips from "approximately right, ranked" to "exactly right, per item, forever". A view event may be counted twice with little harm, because the batch job later corrects it. A like may not, because the number sits beside a heart the viewer toggles, and users notice an off-by-one. That one change rewrites the design.

The fact is an edge, not an event. Store one row per (user, post) with a state and a version, partitioned by user. A like is a conditional write that reports an affected row only when the state flips and the version is newer. Retries, double taps and an unlike arriving before its like all produce no delta. Only a real flip emits a +1 or -1, and the delta is published from the database change stream (CDC or an outbox row), so a crash between commit and publish cannot lose it. Top-K counts requests; this design counts state changes.

Numbers shift too. Assume 500M daily users at 20 likes each: 10B likes a day, about 116k/s on average and 350k/s at peak. Reads dominate at roughly 10 to 1, because every post view shows a count and the viewer's own state, batched 20 posts at a time. The expensive storage is the edges (about 60 B a row, 600 GB a day), not the counters.

Aggregation is per key, with exact checkpoints. The log is keyed by post, so one aggregator owns each counter and folds a second of deltas into one write; a post taking 50,000 likes a second costs one write a second. The aggregator commits its log offset in the same transaction as the counter updates, so a restart neither skips nor repeats deltas. Top-K gets the same property from Flink-style checkpoints but tolerates sketch error, which this design cannot. If one post outgrows a single partition, salt its key into sub-counters that readers sum, as in Firestore's sharded counters (Firestore docs), which trade extra read cost for write throughput.

Serving adds read-your-writes. The aggregator sets the cache to the committed total instead of incrementing it, which is idempotent. The count may lag a few seconds, but the heart comes from the edge row, and the read path adds the viewer's own unapplied delta so the number never jumps back. A recount from the edges, sampled nightly, catches drift, because a derived counter eventually disagrees with its source.

Down the stack

Layer 1The contract

  • A minute window's top K is published within 10 seconds of the minute's end plus grace, and an hour window's within 15 seconds.
  • For sketch-based lists, every reported count c satisfies true ≤ c ≤ true + ε·N with probability at least 1 − δ, and the response states the bound. An item whose true count beats the K-th true count by more than ε·N will, with that same probability, be in the list once its next event arrives, because its estimate can only grow.
  • Daily and all-time lists are exact with respect to the archived events after the batch job completes, and are labelled exact: true.
  • Counts reflect each Kafka record once: counter state and consumed offsets are checkpointed together, using the asynchronous barrier snapshots Flink is built on (Carbone et al.).

Layer 2The mechanism

  • Partitioning. For a keyed record the Kafka producer picks the partition from a hash of the key modulo the partition count (Kafka producer configs). Adding partitions changes that mapping for new records, and Kafka does not move existing data (Kafka operations docs), so an item's counts would briefly live on two shards. The count is fixed up front (say 128) and over-provisioned.
  • Count-min update. For row i, the column is h_i(item) mod w, computed from two base hashes as h1 + i·h2 (the Kirsch-Mitzenmacher trick), often the two halves of one 128-bit hash, so one hash computation per event serves all d rows. Update increments d counters; a query reads d counters and returns the minimum. A conservative-update variant increments only counters equal to the current minimum, which reduces overestimation (Count-min sketch). The cost is that the update depends on the other counters, so the sketch is no longer a simple sum of per-event contributions and subtracting an expired bucket element-wise is unsafe. Use it only for windows that are rebuilt rather than slid.
  • Heap. A binary min-heap of K entries plus a hash map from item to heap index, so updating an existing candidate is O(log K) instead of a scan.
  • Window state. Minute buckets are keyed state with timers at each minute boundary. When the timer fires, the shard folds the bucket into the running hour, subtracts the expiring bucket and emits the partial list.

Layer 3The machine

  • Sketch memory access. A 5.4 MB sketch does not fit in a typical L2 cache but may fit in a shared L3. Each update touches d = 5 random counters; if they miss to DRAM, that is about 5 × 80 ns = 400 ns per event, so one core does roughly 2 to 3M updates/s. Laying the d counters of an item in one cache line (a "blocked" sketch) cuts that to one miss, at a small cost in accuracy.
  • Heap. K = 1,000 entries of 16 B is 16 KB, which sits in L1 and L2. Heap operations cost tens of nanoseconds.
  • Hash map for exact counts. A Go or Java map with 1.5M entries uses roughly 50 to 100 B per entry with overhead, about 100 MB, and rebuilding it each hour causes a GC spike. Open-addressing tables with primitive keys and values (for example 16 B per slot at 50% load) use a third of that.
  • Network. Partial lists are 128 shards × 16 KB per window and scope each minute, negligible next to the event stream.

Why it's built this way

  1. Counting needs all of an item's events in one place, so the stream is partitioned by item. That makes merging exact and cheap.
  2. Partitioning by item concentrates viral items, so counts are combined before the partition step.
  3. Scopes multiply state faster than traffic grows, so per-scope windows use fixed-size sketches whose error is bounded and stated.
  4. Sketch error and streaming lateness are unacceptable for payouts, so a batch job over the durable log produces exact daily numbers.
  5. Readers vastly outnumber window changes, so results are computed once per minute and served as static snapshots.

Failure modes

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
Exact hash map per shardExact, simpleMemory grows with distinct items and scopesFew scopes, tens of millions of items
Count-min sketch + heapFixed memory, mergeableOverestimates by up to ε·NMany scopes, approximate lists acceptable
Space-SavingTight bound, no hashing, mergeableCannot subtract an expired bucketTumbling windows and rollups
Partition by itemExact merge by concatenationHot items on one partitionDefault
Partition randomlyEven loadMerge needs full counts or is approximateOnly with a re-key step after
Batch for long windowsExact, cheap per eventHours of delayDaily and all-time charts

Leadership lens

Check yourself

  1. Why can't the hourly top K be computed by merging the 60 per-minute top K lists?

    Show answer

    An item that ranks 1,001st in every minute could be in the hourly top 1,000 yet appear in none of the minute lists. So each shard keeps full minute counts for the last 60 minutes and slides hour_running by adding the newest minute and subtracting the expiring one.

  2. Why does the design partition the event stream by item ID, and what does that cost when a video goes viral?

    Show answer

    With all of an item's events on one shard, no item appears in two partial lists, so the global top K is just the top K of the concatenation. The cost is that a viral item concentrates on one partition, which the one-second ingest combiner fixes by capping it at about one record per ingest node per second.

  3. Exact counting for the global scope fits in memory. Why do per-scope windows switch to a count-min sketch, and how big is one sketch?

    Show answer

    Exact global hourly state is about 2.8 GB, or 90 MB per shard across 32 shards, but 200 countries x 30 categories pushes exact per-scope counts to tens of GB per hour. With ε = 1e-5 and δ = 0.01 a sketch is about 272k x 5 counters, roughly 5.4 MB, with an error bound of 15k views per hour.

  4. A counter shard crashes mid-minute. Why are counts neither lost nor doubled?

    Show answer

    Counter state and consumed Kafka offsets are checkpointed together, so the shard restores from the last checkpoint and replays from the checkpointed offsets. The only visible effect is that the minute it was processing may publish a few seconds late.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →