Design a top-k popularity ranking system

Questions I would ask: what is "popular", raw event count or distinct users? Is the window sliding (last 60 minutes, updated each minute) or tumbling (each clock hour)? How many items and how many cities or categories? How exact must the ranking be, and does anyone pay money based on it? Can events arrive late or twice? How fresh must the list be?

Assumed scope: raw event count, sliding one-hour window with a one-minute hop, 10 million distinct items active in an hour, results per city, approximate is acceptable (no billing), freshness of about a minute. Out of scope: personalised ranking, spam and fake-event filtering, and distinct-user counting (a follow-up).

Planning figures for the exercise, not facts about any company.

• Rate: 1 billion events a day ÷ 86,400 s = about 11,600 a second; at 5×, 58,000 a second. At 100 B an event that is 5.8 MB/s, under the ~10 MB/s per-partition planning figure, so volume alone does not force many partitions. Parallelism for the counting does.
• Window volume: 11,574 × 3,600 s = about 42 million events in an hour at the average rate (about 5 times that if a whole hour ran at peak).
• Exact state, worst case: 10 million items × 60 one-minute buckets × 16 B = 9.6 GB, assuming every item appears in every minute. Real data is sparse, so this is an upper bound. Split across 8 workers by item id, that is 9.6 ÷ 8 = 1.2 GB each. It fits in memory.
• Sketch alternative: a count-min sketch with error target ε = 0.0001 has width about e ÷ ε = 27,200 and depth 5. At 4 B per cell that is 27,200 × 5 × 4 = 544 KB per bucket, about 33 MB for 60 buckets. The worst over-count is ε × N = 0.0001 × 42 million = 4,200, with probability about 99.3% (1 − e^-5). It only ever over-counts.
• Recovery: replaying the last minute at peak is 58,000 × 60 = 3.5 million events; at an assumed 200,000 a second replay, about 17 s.
• Output: 50 items × 100 B = 5 KB a list. At 10,000 reads a second that is 50 MB/s from a cache.

The numbers force: exact counting is affordable at 10 million items, so use it, shard by item id, and keep the sketch for when cardinality reaches hundreds of millions or billions.

1. Ingest. Services publish {event_id, item_id, city, event_time} to a log. The first stage is a light consumer, holding only one second of state, that pre-aggregates one second of events into partial counts (city, item_id, count) and re-publishes them keyed by hash(item_id). This is what tames a viral item at 20,000 events a second: each pre-aggregator emits at most one record per item per second.
2. Windowed counters. Eight workers each own a slice of items. A worker keeps two things: a total per (city, item_id) for the last 60 minutes, and a sparse map of minute to per-item counts. An event adds to the total and to the current minute's map. At each minute boundary the worker removes the bucket that has just fallen out of the window: for every entry in that bucket, subtract its count from the total. The cost of a roll is the size of the expired bucket, not the whole state.
3. Local top-K. Once a minute each worker scans its totals with a 50-entry min-heap per city. A scan of about 1.25 million items (10 million ÷ 8) with a heap of 50 is a few million operations, well inside the minute.
4. Merge. A merger takes each worker's top 50 and picks the global top 50 from those 8 × 50 = 400 candidates. This is exact, not heuristic. Each item lives on exactly one worker, so an item in the true top 50 must be in the top 50 of its own worker: if it were not, 50 items on its worker would already beat it. This only holds because the partition key is the item id.
5. Serve. The merger writes top:{city}:1h as a versioned blob (about 5 KB) to a key-value store, and the app API reads that one key. Put a 30 s CDN or API-side cache in front. Nobody computes anything at read time.

Hard problem 1: time. Bucket by event time, not arrival time. A phone that was offline can deliver events minutes late. Use a watermark that allows 2 minutes of lateness; anything older is dropped and counted in a metric. Because nothing is billed, an occasional dropped late event is fine. If a daily top-K were needed for reporting, a nightly batch job over the archived log would give the settled numbers.

Hard problem 2: crash recovery. Each worker snapshots its totals and buckets to object storage every minute together with its log offset. On restart it loads the snapshot and replays from the offset, which at peak is under a minute of events (the 17 s figure above). The list is slightly stale during that time and then catches up.

Duplicates. The log is at-least-once, so some events will count twice. A ranking tolerates a tiny error, so I would skip exact de-duplication and say so. If the ranking drives anything paid, add a short-window event_id check in the first stage.

Dimensions. Keep the dimension in the key, (city, item_id), and a heap per city served. Do not precompute every combination (city × category × time window); precompute the few the product shows.

• Exact counts versus a count-min sketch. Pick exact at 10 million items because it is 1.2 GB per worker and has no error. Flip to a sketch plus a candidate heap when distinct items reach hundreds of millions or billions, such as search queries, and accept an over-count bounded by ε × N.
• Sliding window versus tumbling. Pick a sliding hour with a one-minute hop for a list that moves smoothly. Flip to tumbling clock hours if the product only needs "top of the last hour" once an hour, since it needs no subtract step.
• Raw count versus a decayed score. Pick raw count for "most popular". Flip to an exponentially decayed score when the product means "trending now", because a steady best seller should not outrank a sudden riser.
• Partition by item id versus by event id. Item id keeps all counts for an item on one worker and makes the merge exact, but skews on a viral item, which is why the pre-aggregation stage exists. Event id spreads load evenly but forces a shuffle before counting. Pick item id; flip if one item exceeds what a pre-aggregated stream can carry.

Typical mistakes:

• Running GROUP BY item ORDER BY count DESC LIMIT 50 over the last hour of raw events on every request.
• Keeping one global sorted set and updating it for every event at 58,000 a second.
• Partitioning by anything other than item id, then merging the per-partition top 50 as if it were exact.
• Handling a sliding window by recounting 42 million events each minute instead of adding the new bucket and subtracting the expired one.
• Using arrival time and then wondering why late phone events distort the list.
• Reaching for a sketch without checking that exact state already fits in memory.
• Ignoring the viral item, so one worker is overloaded while the others idle.