Design a top-K popular items system

Report the K most viewed or played items over sliding windows (last minute, hour, day) from a firehose of billions of events per day.

We are building the top-K service for a food delivery marketplace: from a firehose of menu-item views and orders (billions of events a day), it reports the K most viewed or ordered items per restaurant, per city, and globally over sliding windows of the last minute, hour, and day, and it feeds the "Popular items" shelf on every restaurant's menu page.

Functional requirements

• topK(scope, metric, window, k): scope is a restaurant, a city, or global; metric is views or orders; window is 1m, 1h, or 1d; k up to 100.
• The restaurant menu page shows a shelf of the top 5–10 items, enriched with name, photo, price, and availability.
• Items that are deleted, sold out, or hidden are never shown, even if they are still in the counts.
• Late events (mobile clients offline, retries) and corrections (fraud, refunds, test orders) are applied, and the ranking changes to match.
• Backfill: when scoring rules change (for example, orders weigh 5× views), recompute history.

Non-functional requirements

• Shelf read p99 under 20 ms; it is on the menu page critical path.
• Freshness: 1m and 1h lists lag at most 10–30 s; 1d lists and the shelf may lag up to 15 min.
• Ingest at 5× average peak with no data loss; at-least-once intake with dedup by event ID.
• Approximate is acceptable for global and city lists (rank errors near the cutoff only); per-restaurant shelves must be exact over the daily batch.
• A failed pipeline must never empty a shelf: serve the last good snapshot with its as_of time.

Back-of-the-envelope estimation

• 5B events/day (90% views, 10% orders) → ~58K events/s average, ~250K/s peak at dinner.
• Event ≈ 150 bytes → ~750 GB/day raw, ~22 TB/month in the lake (compressed Parquet ~5×: ~4.5 TB/month).
• 1M restaurants × ~120 items = 120M items. In any minute, maybe 3M distinct items get an event.
• Per-minute bucket for all active items: 3M × (8 B id + 4 B count) ≈ 36 MB. Keeping 60 minute buckets ≈ 2 GB; 24 hourly buckets for the day window ≈ 24 × 20M active items × 12 B ≈ 6 GB. Fits in memory across a few stream workers.
• Shelf reads: 400M menu views/day → ~5K/s average, ~25K/s peak. Snapshot per restaurant ≈ 10 items × 40 B = 400 B → 1M × 400 B = 400 MB total. One KV cluster, or even a CDN-cached file per city.

Public read (menu page, called by the menu service, not clients directly):

GET /v1/restaurants/{id}/popular-items?k=8&metric=orders&window=1d
→ 200 {
  "restaurant_id": "r_812",
  "as_of": "2026-10-12T18:15:00Z",
  "window": "1d", "metric": "orders",
  "items": [
    {"item_id": "i_33", "rank": 1, "score": 412, "name": "Spicy Tuna Roll",
     "price_cents": 1299, "photo_url": "...", "available": true}
  ],
  "source": "batch" | "stream" | "fallback"
}

Analytics read (internal dashboards, trending modules):

GET /v1/topk?scope=city:sf&metric=views&window=1h&k=50
→ 200 {"as_of": "...", "approximate": true, "items": [{"item_id", "score", "error_bound"}]}

Ingest (from the event gateway, not a public API): ItemEvent {event_id, item_id, restaurant_id, city_id, type: view|order, ts_event, ts_received, user_id_hash} on a Kafka topic.

Internal control plane:

• POST /internal/topk/rebuild {scope: "restaurant:r_812" | "all", from, to, scoring_version} → 202 {job_id}. It recomputes snapshots from the lake and swaps them in atomically.
• POST /internal/topk/corrections {event_ids: [...], reason} voids events (fraud, refunds). They drop out at the next incremental run.

Trust and abuse: reads need a service token (mTLS between services); the public menu page never sends a raw k larger than 20. Views are deduplicated per (user_id_hash, item_id, 10 min) and counted only from signed-in or device-attested sessions, so a script reloading a page cannot push an item onto the shelf. source and as_of tell callers how much to trust the list.

Snapshots store item IDs and scores only. Name, price, and availability come from the menu service at read time, so a renamed or sold-out item is right on the next page view.

flowchart LR
  APP[Apps and web] --> GW[Event gateway]
  GW --> K[(Kafka item events)]
  K --> FL[Flink stream topK]
  K --> SINK[Lake sink]
  SINK --> LAKE[(S3 Parquet lake)]
  LAKE --> BATCH[Spark rollup and rank]
  VOID[(Voided events)] --> BATCH
  BATCH --> SNAP[(Snapshot KV)]
  FL --> HOT[(Hot list KV)]
  FL -. "fresh shelf patches" .-> SNAP
  MENU[Menu page service] --> TK[TopK read API]
  TK --> SNAP
  TK --> HOT
  TK --> MS[Menu items service]

1. Ingest. Clients send view and order events to the gateway, which validates them, assigns event_id if missing, and writes to Kafka keyed by restaurant_id.
2. Authoritative batch (the shelf). A lake sink writes hourly Parquet. Every 15 min, a Spark job folds new hours into item_counts_hourly, anti-joins voided_events, sums the last 24 hours per restaurant, ranks, and writes a new snapshot version per restaurant. It flips a pointer only after all writes for the run succeed.
3. Streaming (minute and hour windows). Flink keeps per-item minute buckets, emits per-scope top-K every 10 s to the hot list KV, and optionally patches shelves for restaurants with heavy traffic.
4. Read. The menu service calls the top-K API, which reads one snapshot key, batch-fetches the item details from the menu service (cached 60 s), drops unavailable items, backfills from ranks 9–20 to fill the shelf, and returns it.
5. Corrections and backfills. Voids and scoring changes trigger an incremental rerun of affected hours, or a full rebuild that writes a new snapshot version and swaps it in.
6. Menu item lifecycle. The top-K pipeline never owns names or prices. Deleted items are filtered at read time and pruned at the next run; a re-created item with a new ID starts at zero, so merges of duplicate items are a menu-service concern, mapped through an item_aliases table in the rollup.
7. Cold start and sparse history. New restaurants or those with fewer than 20 orders in 24 h lack a stable signal; widen to 7 or 28 days, then fall back to the merchant's own "featured" flag. Smooth scores (orders + α·views, Bayesian prior from the cuisine) so one order does not crown an item.
8. Time zones. "Last day" is a rolling 24 h in UTC for ranking; daily reports use the restaurant's local day, computed from the same hourly rollups. Start single-region, with the lake and batch in one region and snapshot KV replicated read-only to others; the shelf tolerates minutes of lag.

Windowed rollups and late corrections. Count in tumbling buckets and build sliding windows by summing buckets: 1m is the current minute bucket plus a weighted share of the previous one; 1h is 60 minute buckets; 1d is 24 hour buckets. This makes a slide an O(1) subtract-and-add, not a rescan. Late data: Flink uses event time with a 2-minute watermark; events later than that go to a side output and land in the lake anyway, so the next batch run counts them. The batch keys hourly rollups by event hour, so a late event changes one old hour, and only that hour's partition is recomputed. Voids are applied as an anti-join, never by decrementing in place, so a rerun gives the same answer.

INSERT OVERWRITE item_counts_hourly PARTITION (dt, hour)
SELECT restaurant_id, item_id,
       SUM(CASE WHEN type='order' THEN 1 ELSE 0 END) AS orders,
       COUNT(DISTINCT CASE WHEN type='view' THEN user_id_hash END) AS viewers,
       dt, hour
FROM events e LEFT ANTI JOIN voided_events v ON e.event_id = v.event_id
WHERE (dt, hour) IN (:dirty_hours)
GROUP BY restaurant_id, item_id, dt, hour;
-- then: SUM over last 24 hours, ROW_NUMBER() OVER (PARTITION BY restaurant_id ORDER BY score DESC) <= 20

Snapshot serving contract and freshness. Each run writes shelf:{r}:...:v{run_id}, then sets shelf:{r}:current = v{run_id}. Readers see a whole run or the previous one, never a mix. Rules: if a run fails, the pointer stays put and as_of shows the age; an alert fires at 45 min stale; past 24 h stale, the API returns source: fallback (top items by all-time orders, a small static list per restaurant). Store 20 ranked items, not K, so the read path can drop unavailable items and still fill 8 slots without a recompute. Item details are joined at read time with a 60 s cache; ranking and enrichment fail independently, and either failure yields a degraded shelf, not an empty one.

Hot-restaurant skew and incremental recompute. Partitioning by restaurant_id puts a viral restaurant (100× normal traffic during a promo) on one Kafka partition and one Flink task. Fixes: pre-aggregate in the gateway or a first Flink stage keyed by (restaurant_id, item_id, salt 0–15) over 1 s, then merge per restaurant. That cuts the downstream rate by ~100× for hot keys. For batch, recompute only restaurants with events in dirty hours: on a typical 15-min run that is ~300K of 1M restaurants, and a run takes ~3 min instead of ~20 for all of them. For global and city lists, exact counting of 20M active items is affordable per hour, but the 1m global list uses a Count-Min Sketch (width 2^16, depth 4 ≈ 1 MB, error ≈ 0.004% of the stream) plus a min-heap of size 1,000 per worker, merged into the top 100.

When batch is not enough. A 15-min batch is fine for "popular this week" shelves, but not for "trending right now" or a dish that sells out at 7 pm. Add streaming where staleness costs money: the 1m/1h lists, and shelf patches for the ~1% of restaurants that get over 1K events/min, which Flink overwrites between batch runs, tagged source: stream. Streaming results are provisional: the next batch snapshot replaces them, so stream bugs heal within 15 min and the stream path can stay simple (no exactly-once sinks, idempotent overwrites only).

Backfills after scoring changes or disputes. Scoring is versioned (scoring_version in the snapshot). A change, such as weighting orders 5× views or excluding group orders, runs the hourly rollup over the last 30 days of lake data (~22 TB raw, ~1 h on a 200-core cluster), writes snapshots under a new version, compares rank overlap against the current version on a sample, and then swaps the pointer. A restaurant disputing a shelf (fake views by a competitor) gets events voided, followed by rebuild for that one restaurant, which takes seconds.

• Batch as the source of truth gives exact, correctable, replayable shelves, but they lag by 15 min; streaming covers freshness only where it pays, and as a result there are two code paths to keep consistent.
• Count-Min Sketch keeps global windows in fixed memory but overcounts rare items; exact counting is used wherever the key space is small enough to afford it.
• Storing 20 IDs and enriching at read time keeps availability correct, but adds a dependency on the menu service in the read path (mitigated by a cache and by degrading to IDs with cached names).
• Event-time windows with a 2-min watermark trade a short delay for correct minute buckets; anything later is fixed only by batch.
• Snapshot versioning with pointer swaps doubles storage briefly in exchange for atomic, rollback-able publishes.
• Salting hot keys adds a merge stage and a second of latency, but stops one viral restaurant from stalling a partition.

Mid-level (L4)

• Clarifies windows, K, scope, and exact vs approximate before designing.
• Proposes counting per item into time buckets and a heap or sort to get the top K; shows a Kafka-to-counts-to-cache pipeline.
• Gets the throughput estimate right and puts a cache in front of reads.

Senior (L5)

• Separates the cheap read path (precomputed snapshots) from the compute path, and explains sliding windows as sums of buckets.
• Handles late events with watermarks plus batch reconciliation, applies voids idempotently, and filters unavailable items at read time.
• Knows Count-Min Sketch or Space-Saving and when exact counting is affordable; addresses hot-partition skew.

Staff (L6+)

• Argues that batch is the authoritative baseline and adds streaming only where freshness pays, with stream results healing on the next batch.
• Defines the serving contract: atomic snapshot swaps, as_of, staleness alerts, fallbacks, and never an empty shelf.
• Plans backfills under versioned scoring, abuse resistance on views, cold start, and the boundary with the menu service, and estimates the cost of each recompute.