Everyone thinks top-K is a SELECT article_id, COUNT(*) FROM shares GROUP BY article_id ORDER BY count DESC LIMIT 100. Store every share, group, sort, take the top 100, done. Then the interviewer adds the constraints that make it a real problem: it is not one all-time ranking, it is three sliding windows at once - the top 100 in the last 5 minutes, the last hour, and the last 24 hours; there are 100 million articles; there are a billion shares a day, spiking hard when something goes viral; and the answer has to be near-real-time, so a share that just happened should be able to move an article onto the list within a second or two, not on the next hourly batch.

The reason the naive GROUP BY dies is not the sort - sorting 100 things is nothing. It is that a sliding window means the counts are constantly expiring out the back while new shares pour in the front, so there is no single static table to sort; and it is that you cannot afford to hold an exact per-window counter for 100M articles times three windows in memory, nor re-scan a day of shares on every query. The whole design comes down to four decisions: how you count a firehose without storing a counter per article (approximate counting - Count-Min Sketch), how you make counts expire so a “last 5 minutes” number actually forgets old shares (time-bucketed windows), how you keep only the top K without sorting 100M candidates every second (a heap plus the Space-Saving algorithm), and how you merge top-K across shards without the merge being wrong (over-fetch). All while a top-K read stays in single-digit milliseconds.

Let me do it properly.

Functional Requirements (FR)

In scope:

  • Record a share. A user shares an article; the system counts it. This is the firehose - a billion of these a day.
  • Top-K for a window. Return the K most-shared articles (article_id, share count, rank) for a given time window. K is small and fixed, typically 10, 50, or 100.
  • Three fixed sliding windows. Last 5 minutes, last 1 hour, last 24 hours, all live at once. “Sliding” means at any instant the 1-hour window is exactly the trailing 60 minutes, not a fixed clock-aligned hour.
  • Near-real-time freshness. A share should be reflected in the relevant top-K within ~1-2 seconds. This is a “trending now” feature, so lag makes it useless.
  • Per-category / per-region variants (optional). Top-K within “sports” or within “EU”. Same machinery, more keys.

Explicitly out of scope (say it so you own the scope):

  • Exact counts. We do not promise the count is exactly right to the unit. For a trending list, “roughly 1.2M shares, ranked 4th” is the product; being off by a fraction of a percent on the count never changes the top-K in a way anyone cares about. This is the single assumption that unlocks the whole design - we will trade a bounded, tiny count error for orders of magnitude less memory.
  • Arbitrary historical windows. “Top articles between 3 and 5 weeks ago” is an offline analytics query against a data warehouse, not this system. We serve three fixed trailing windows.
  • Per-article exact rank on demand. This is not the leaderboard problem (“what is my article’s rank, even if it is 4-millionth”). We return the top K; ranks below K are not a supported query.
  • Share validation / fraud. Deciding whether a share is a real human or a bot ring is a separate abuse system. We assume shares arriving here are already filtered, and note where that gate sits.
  • Full share audit trail. We keep counts, not a queryable log of who shared what when. That is a separate event store / warehouse.

The one decision that drives everything: this is a streaming heavy-hitters problem over sliding windows, at a scale where you cannot keep an exact counter per key. So the design is dominated by approximate counting structures (which give bounded error in sublinear memory), time-bucketing to make windows expire, and a bounded top-K structure so you never sort the full key space.

Non-Functional Requirements (NFR)

  • Scale: 100M distinct articles. 1B shares/day. Average ingest ~11.6K shares/sec, but virality is spiky - a single article can take a large share of traffic in a burst, so size for a peak of ~50K-100K shares/sec. Read traffic (people loading the trending panel) is high but the answer is identical for everyone, so it is trivially cacheable - effectively a handful of reads/sec after caching, per window.
  • Latency: top-K read p99 under ~10ms (from cache, sub-millisecond). Share-ingest ack p99 under ~10ms - a share is fire-and-forget, we ack fast and count asynchronously.
  • Freshness: ~1-2 seconds from a share to it being reflected in the windows. Not synchronous, but not a batch job.
  • Availability: 99.9%+. Trending is a prominent, user-facing widget. It should degrade to slightly-stale rather than error. Losing a few seconds of counts during a failover is acceptable; the sky does not fall if “trending” is 3 seconds behind.
  • Consistency: eventual and approximate by design. Two servers may briefly disagree on the exact count or on rank 97 vs 98. That is fine. What is not fine is a wildly wrong list (a dead article showing as #1) or the 5-minute window failing to forget old shares.
  • Durability: low for individual shares. This is the surprising one: we do not need every single share to be durably persisted before we count it, because the product is a statistical aggregate. Losing 50ms of shares on a crash shifts a count by a rounding error. We optimize for throughput over per-event durability - the opposite of a payments system. (We can still tee the raw stream to a durable log for rebuild and analytics, but the counting path does not block on it.)

Back-of-the-Envelope Estimation (BoE)

Let me ground the design in numbers.

Traffic:

Distinct articles:               100,000,000   (100M)
Shares per day:                  1,000,000,000 (1B)

Average ingest QPS = 1B / 86,400 sec ≈ 11,574 shares/sec
Virality is spiky (a story explodes, everyone shares in one hour):
Peak ingest QPS (say 5-8x avg)   ≈ 50,000 - 100,000 shares/sec

Reads (top-K panel loads): very high volume, but ONE answer per window
  for the whole planet -> collapses to ~1 cache refresh/sec per window.
  Effective backend read QPS after caching: a few per second.

The daily average of ~11.6K/sec is the honest steady state; the design must survive the ~50K-100K/sec viral spike. The read side is a non-event because every user asking “what is trending” gets the same three lists, so we compute each once per second and serve it from cache to everyone.

Why exact counting does not fit in memory:

Naive: one exact counter per article per window.
  100M articles * 8 bytes (count) = 800 MB per window ... seems fine?

But a SLIDING window needs per-article, per-time-BUCKET counts, not one number:
  5-min window at 10s buckets  = 30 buckets
  1-hour window at 1-min buckets = 60 buckets
  24-hour window at 5-min buckets = 288 buckets
  Total buckets to retain ≈ ~300 (share the fine buckets, see deep dive)

  100M articles * ~300 buckets * 8 bytes = 240 GB  <-- and most are ZERO

Exact per-article, per-bucket counting is ~240 GB of mostly-empty counters, dominated by the 100M cold articles that get one share a week. That is the wall the naive design hits, and it is why we reach for approximate counting (Count-Min Sketch), whose memory depends on the desired error, not on the 100M key count.

Approximate counting memory (Count-Min Sketch):

One CMS with width w = 2^20 (~1M columns), depth d = 4 rows, 4-byte counters:
  w * d * 4 = 1,048,576 * 4 * 4 ≈ 16 MB  per sketch  (INDEPENDENT of 100M!)

We keep one sketch PER TIME BUCKET (they get summed over a window):
  ~300 buckets * 16 MB ≈ 4.8 GB total, fits in one node's RAM.

The magic: a Count-Min Sketch’s size is fixed by the error you tolerate (columns) and confidence (rows), completely independent of whether there are 100 or 100M articles. We go from 240 GB of exact counters to ~5 GB of sketches, and the price is a small, bounded overcount on collisions (deep dive).

Top-K candidate memory:

We never sort 100M articles. We keep a bounded "heavy hitters" structure
of the top ~m candidates (m = a few thousand, say 10x-100x K):
  10,000 candidates * (16-byte id + counts) ≈ a few MB. Trivial.

Storage / bandwidth:

Top-K response: 100 entries * ~50 bytes ≈ 5 KB, served from cache.
Raw share event: ~50-100 bytes on the wire.
Ingest bandwidth at 100K/sec * 100 B ≈ 10 MB/sec. Modest.
Optional durable raw log (for rebuild/analytics): 1B/day * 100 B = 100 GB/day
  -> to cheap object storage / a warehouse, off the hot path.

Bandwidth is a non-issue. The scarce resources are ingest throughput (absorbing 100K approximate-count updates/sec) and memory (which the sketch makes bounded). Size for those.

High-Level Design (HLD)

The system splits into a write path that takes the share firehose, shards it, and folds each share into time-bucketed approximate counters while maintaining a bounded top-K per shard; and a read path that merges the per-shard top-K candidates for a requested window and serves the result from a ~1s cache. The heart of it is the Counting Layer: per-bucket Count-Min Sketches plus a Space-Saving heavy-hitters heap, per shard.

              WRITE PATH (share firehose, ~50K-100K/sec peak)
   ┌──────────────────────────────────────────────────────────────────┐
   │  Clients / App  --share-->  API Gateway / LB  --> Share Ingest    │
   │                                                    (stateless)     │
   │                              202 Accepted (fire-and-forget)        │
   └───────────────────────────────────┬───────────────────────────────┘
                                        │  produce {article_id, ts}
                              ┌─────────▼──────────────────┐
                              │  Kafka  (partitioned by     │
                              │  hash(article_id), N parts) │  <- same article
                              └─────────┬──────────────────┘     always same shard
                                        │
             ┌──────────────────────────┼──────────────────────────┐
             ▼                          ▼                          ▼
   ┌───────────────────┐      ┌───────────────────┐      ┌───────────────────┐
   │  Counter Shard 0  │      │  Counter Shard 1  │ ...  │  Counter Shard M  │
   │  ---------------  │      │                   │      │                   │
   │  ring of per-     │      │  ring of per-     │      │  ring of per-     │
   │  bucket CM-       │      │  bucket CM-       │      │  bucket CM-       │
   │  Sketches         │      │  Sketches         │      │  Sketches         │
   │  + Space-Saving   │      │  + Space-Saving   │      │  + Space-Saving   │
   │    top-m heap     │      │    top-m heap     │      │    top-m heap     │
   │    per window     │      │    per window     │      │    per window     │
   └─────────┬─────────┘      └─────────┬─────────┘      └─────────┬─────────┘
             │  each shard exposes its local top-m for each window │
             └──────────────────────────┬─────────────────────────┘
                                         ▼   READ PATH
                              ┌─────────────────────────┐
                              │   Aggregator / Merger    │  GET /top?window=5m
                              │   - pull each shard's    │  - fan-in shard top-m
                              │     top-m for the window │  - sum, re-rank
                              │   - merge -> global top-K│  - write cache
                              └───────────┬─────────────┘
                                 ┌────────▼─────────┐
                                 │  Top-K Cache     │ (one entry per window,
                                 │  (Redis / CDN)   │  ~1s TTL, served to all)
                                 └──────────────────┘

   (side path)  Kafka --tee--> Durable Raw Log (object store / warehouse)
                                for rebuild + offline analytics, OFF hot path

Request flow - a share (the write hot path):

  1. The client sends POST /shares with article_id (and optional category/region). The Share Ingest service is stateless; it does the bare minimum - stamp a server-side ts, then produce the event to Kafka - and immediately returns 202 Accepted. No durable write blocks the ack; a share is a statistical drop in the bucket.
  2. Kafka partitions by hash(article_id) so every share of a given article always lands on the same partition, and therefore the same counter shard. This is the key routing decision: it keeps each article’s count in one place, so a shard’s local count for its own articles is complete, not a fragment.
  3. The Counter Shard consumer folds the share into the current time bucket’s Count-Min Sketch (increment the article’s cells) and offers the article to that window’s Space-Saving top-m heap. It does this for each window granularity it maintains.
  4. Within ~1-2 seconds the sketches and heaps reflect the share, so the next merge sees it.

Request flow - a top-K read:

  1. A user loads the trending panel; the app calls GET /top?window=5m&k=100. The Read Service returns the cached list for that window - one precomputed answer for the whole planet. Sub-millisecond.
  2. The cache is kept warm by a background Aggregator that, ~once per second per window, asks each Counter Shard for its local top-m candidates for that window, sums each candidate’s count across shards (querying the sketches for exactness on the merge), re-ranks, takes the global top-K, and overwrites the cache entry. The many millions of identical reads collapse into one merge per second per window.

The key architectural insight: counting is approximate and sharded-by-article, top-K is bounded, and windows expire by bucket rotation. We never store a counter per article per window (sketches make memory independent of key count), we never sort 100M candidates (Space-Saving keeps only the heavy hitters), and “last 5 minutes” actually forgets old shares because old buckets rotate out of the ring. The durable raw log exists only for rebuild and analytics; the counting path never waits on it.

Component Deep Dive

Four hard parts: (1) counting a firehose over 100M keys without a counter per key, (2) making a sliding window actually expire old shares, (3) maintaining the top K without sorting everything, and (4) merging top-K across shards without the merge being wrong. I walk each from naive to scalable.

1. Counting the Firehose: Hash Map -> Count-Min Sketch

Approach A: An exact counter per article (the naive instinct)

Keep a hash map article_id -> count, increment on each share.

For a single all-time count this is fine. Where it breaks is the combination of 100M keys and windows. Per window you need a separate count, and for a sliding window you need counts broken down by time bucket (so you can expire the old ones - part 2). That is 100M keys times ~300 buckets of mostly-zero 8-byte counters ≈ 240 GB, dominated by the enormous cold tail of articles that get a share a week and will never be in any top-K. You are spending almost all your memory on keys that can never matter. An exact map does not know in advance which keys are heavy, so it must hold them all.

Approach B: Count-Min Sketch (approximate, memory independent of key count)

The insight: for a trending list we do not need the exact count of every article, we need good counts for the few heavy ones. A Count-Min Sketch (CMS) gives exactly that. It is a 2D array of counters, d rows by w columns, with d independent hash functions. To add a share for an article, hash the id with each of the d functions and increment one counter in each row. To estimate its count, hash again and take the minimum of those d counters.

Count-Min Sketch: d=4 rows, w=8 columns (tiny, for illustration)

           col0 col1 col2 col3 col4 col5 col6 col7
  h1 ->    [ 3][ 0][12][ 0][ 5][ 0][ 0][ 1]   article X hashes to col2 in row0
  h2 ->    [ 0][ 9][ 0][ 0][ 3][12][ 0][ 0]   ... col5 in row1
  h3 ->    [ 1][ 0][ 0][12][ 0][ 0][ 4][ 0]   ... col3 in row2
  h4 ->    [12][ 0][ 0][ 0][ 0][ 2][ 0][ 6]   ... col0 in row3

  add(X):     increment the 4 cells X maps to
  estimate(X): min(12, 12, 12, 12) = 12   (min cancels most collisions)

Why the minimum: two different articles can collide in one row and inflate that counter, but they are very unlikely to collide in all d rows, so the smallest of the d cells is the least-polluted estimate. CMS never undercounts (collisions only ever add), and it overcounts by a bounded amount:

With width w and depth d:
  overestimate error <= epsilon * (total count N), with probability 1 - delta
  where epsilon = e / w   and   delta = e^(-d)

  w = 2^20 (~1M cols): epsilon ≈ 2.7e-6  -> error <= a few millionths of N
  d = 4 rows:          delta   ≈ 0.018    -> ~98% confidence
  Memory = w * d * 4 bytes ≈ 16 MB, whether there are 100 or 100M articles.

That last line is the whole point: CMS memory is fixed by the error you tolerate, not by the number of articles. We swap 240 GB of exact counters for ~16 MB per bucket, and the price is a tiny, one-directional, bounded overcount. For a heavy article (millions of shares), an overcount of a few dozen from collisions is noise. The error is worst relative to tiny counts - which is fine, because tiny-count articles are never in the top-K anyway.

One gap: a CMS can estimate a count if you hand it an id, but it cannot enumerate the heavy hitters - it does not know which ids are hot. That is what part 3’s heap is for.

Approach C: CMS + conservative update

A refinement: conservative update (a.k.a. minimal increment). On add, instead of incrementing all d cells, only increment the cell(s) that equal the current minimum. This cannot break the never-undercount guarantee and measurably reduces overcount from collisions, tightening the estimate for free. Cheap win, standard in production sketches.

2. The Sliding Window: Full Rescan -> Time-Bucketed Ring

A “last 5 minutes” count must forget shares older than 5 minutes. This is the part the naive GROUP BY cannot do without re-scanning raw data.

Approach A: Recompute from raw shares on each query (naive)

Keep every raw share with a timestamp; on a query, scan the last 5 min / 1 hr / 24 hr and aggregate.

Where it breaks: the 24-hour window is ~1B rows to scan per query, and even the 5-minute window is millions of rows. You cannot re-aggregate a billion events every second. And you would be storing every raw share forever. This is a batch-analytics pattern masquerading as a real-time one.

Approach B: Time-bucketed counters, summed over the window (a ring buffer)

Chop time into fixed buckets and keep a separate counting structure (a CMS, from part 1) per bucket. A window’s count for an article is the sum of its estimates across the buckets covering that window. Old buckets rotate out of a ring buffer and their memory is reused - that is how the window forgets.

Fine ring for the 5-min and 1-hour windows: 10-second buckets
  [b0][b1][b2] ... [b359]   (1 hour = 3600s / 10s = 360 buckets)
   ^current           ^oldest, about to be overwritten

  last 5 min  count(X) = sum of estimate_X over the newest 30 buckets
  last 1 hour count(X) = sum of estimate_X over all 360 buckets

Coarse ring for the 24-hour window: 5-minute buckets
  [B0][B1] ... [B287]       (24 hours = 1440min / 5min = 288 buckets)
  last 24h count(X) = sum of estimate_X over all 288 buckets
  • Bucket granularity is the accuracy/memory knob. Fine buckets (10s) make “last 5 minutes” precise to +/-10s at the edge; coarse buckets (5min) keep the 24-hour ring small. We use two tiers rather than one 10s ring covering 24h (that would be 8,640 buckets * 16 MB = way too much). The 5-minute-window slack of up to 10 seconds is invisible to users.
  • Rotation is the expiry mechanism. When wall-clock crosses a bucket boundary, advance the ring head to the next slot and zero it (reuse the memory). The share that was in the bucket 5m1s ago is simply no longer summed. No delete job, no scan - the window slides because the ring head moves.
  • Query cost is bounded and tiny: summing 30, 360, or 288 per-bucket estimates for a candidate article is a few hundred cheap additions, not a scan of raw shares.

The classic objection - “summing per-bucket CMS estimates compounds the overcount across buckets” - is real but small: error is still bounded by epsilon * (total shares in the window), and for heavy hitters that is a rounding error. If you want the 24-hour window tighter, use an exponentially-decaying / smoothing variant or just accept the coarse bucket, since 24h trending does not need second-level precision.

Approach C: Exponential Histogram (for very tight window bounds)

If the interviewer pushes on exact sliding-window sums under strict memory, mention Exponential Histograms (Datar-Gionis-Indyk): they maintain a sliding-window count within a 1+/-epsilon factor using O((1/epsilon) log N) buckets of exponentially growing width, merging old fine buckets into coarse ones as they age. It is the theoretically clean answer for a bounded-error sliding sum; in practice the two-tier ring above is simpler and good enough, so I would name-drop the EH and stick with the ring unless precision demands otherwise.

3. Maintaining Top-K: Full Sort -> Min-Heap -> Space-Saving

We can now estimate any article’s windowed count. But to return the top K we must know which ids are heavy - and we refuse to sort 100M.

Approach A: Sort all articles by count each query (naive)

Enumerate every article, estimate its window count, sort, take K.

Where it breaks: enumerating 100M ids and estimating each, every second, is ~100M * (bucket-sum) operations per query - hundreds of millions of ops/sec of pure waste, 99.99% of it on articles with a handful of shares that will never rank. And a CMS cannot even enumerate ids; you would need a separate list of all 100M ids to iterate. Dead end.

Approach B: A min-heap of size K

Keep a min-heap of size K keyed by count. On each share, look up the article’s current window estimate; if it beats the heap’s minimum (or is already in the heap), update. The heap root is the K-th place; anything below it is ignored.

This is far better - O(log K) per update, K tiny - but it has a real flaw for windows: an article can climb without being shared right now. Counts also decay as buckets expire, so an article sitting in the heap can silently fall below the true K-th place, and a rising article that was evicted earlier has no way back in because we stopped tracking it. A plain size-K heap has no memory of “almost made it” candidates, so under a sliding window it drifts out of correctness.

Approach C: Space-Saving (Stream-Summary) - bounded, guaranteed heavy hitters

The fix is the Space-Saving algorithm, which keeps a bounded set of m monitored counters (m = a few thousand, well above K) with a clever eviction rule that guarantees the true top-K are captured for skewed streams (and share streams are extremely skewed - a few viral articles dominate).

Space-Saving: monitor m counters (id -> {count, error})

on share(X):
  if X is monitored:            count[X] += 1
  else if fewer than m ids:     add X with count=1, error=0
  else:                         # replace the current MINIMUM-count id Y
     evicted = min-count id Y
     count[X]   = count[Y] + 1          # X inherits the floor, +1
     error[X]   = count[Y]              # remember it may be overcounted by this
     drop Y, insert X

query top-K: return the m ids sorted by count, take K
  guaranteed_rank = count - error  (worst-case true count)

Why it works: a genuinely heavy article, once it enters the monitored set, keeps getting incremented and quickly rises above the eviction floor, so it is never dropped. Only light, transient ids churn through the minimum slot. The error field bounds how wrong each count could be, so you can even certify “X is definitely in the top-K.” We keep a Space-Saving summary per window (5m, 1h, 24h), fed by the same share stream, sized m at 10x-100x K.

Combining parts 1-3 per shard: the share stream feeds (a) the per-bucket CMS ring, which answers “what is X’s count over this window” for any candidate, and (b) the per-window Space-Saving summary, which answers “who are the candidate heavy hitters.” The summary proposes the candidates; the sketch-ring gives their precise windowed counts at merge time. Heap operations are O(log m), memory is a few MB, and 100M cold articles never touch either structure beyond a fleeting min-slot visit.

4. Distributed Merge: Getting the Global Top-K Right

One node cannot ingest 100K shares/sec and hold every window’s structures forever, so we shard by hash(article_id) across M nodes (part of the HLD). Each shard maintains its own CMS rings and Space-Saving summaries for its articles. Now: how do you get the global top-K from M local top-K lists?

Approach A: Take each shard’s top-K, merge, done (the naive, and subtly wrong, merge)

Ask each shard for its local top-K, union them, re-rank, take K.

Where this is wrong: because we shard by hash(article_id), each article lives entirely on one shard, so a shard’s local count for its own article is the global count - good. But an article that is globally #40 might be only, say, #63 on its own shard (that shard happens to hold several bigger articles), so it never appears in that shard’s local top-K and is missed entirely from the merge. Taking exactly top-K from each shard silently drops true global winners that rank below K locally. Classic distributed top-K trap.

Approach B: Over-fetch - pull top-m per shard, merge with exact counts

Two fixes, used together:

  1. Over-fetch. Ask each shard for its local top-m where m is comfortably larger than K (e.g. m = K * some factor, or a fixed few hundred/thousand - exactly the Space-Saving summary we already keep). A global top-K article can rank at worst K-th globally; pulling a generous top-m per shard makes the probability it is missing on its home shard negligible for skewed streams. Since each article is wholly on one shard, we do not even need the article to be top-m on other shards - it only has to make its own shard’s top-m, which it easily does if it is globally heavy.
  2. Exact counts at merge. Because sharding is by article, each candidate’s count comes from exactly one shard - no summing partial counts across shards, no compounding of cross-shard error. The Aggregator collects the union of per-shard top-m candidate ids with their (shard-local, hence global) windowed counts, sorts, and takes the true global top-K.
Aggregator (per window, ~1x/sec):
  candidates = {}
  for shard in shards:
     for (id, count, error) in shard.top_m(window):   # local = global count
        candidates[id] = (count, error)                # no cross-shard sum needed
  global_top_k = sort(candidates by count desc)[:K]
  cache.set("top:" + window, global_top_k, ttl=1s)

Because we partition by hash(article_id), the merge is a plain union-and-sort with no double counting - the reason we chose article-hash sharding over, say, sharding by time is precisely to keep each article’s count whole on one node. The table below is the trade:

Shard keyCount of an articleMerge correctnessVerdict
Hash of article_idcomplete on one shardunion of top-m, exact per candidateRight - keeps counts whole
Round-robin / by timesplit across all shardsmust sum partial counts, errors compound, hot bucketRejected - fragments counts
Take exactly top-K per shardcompletemisses globally-heavy-but-locally-#K+1 articlesRejected - drops winners

The residual approximation is (a) the CMS overcount (bounded, tiny for heavy hitters) and (b) the Space-Saving guarantee (exact for sufficiently skewed streams, which trending is). We report the list as “trending,” not as an audited count, which is exactly the product.

API Design & Data Schema

Ingestion API (the write plane)

POST /v1/shares
  {
    "article_id": "a_9f2c",
    "category": "sports",        # optional, for per-category boards
    "region": "eu",              # optional, for per-region boards
    "ts": 1752451200             # server stamps if absent
  }
  202 Accepted    -> counted asynchronously (~1-2s), fire-and-forget

Notes:
  - No durable write blocks the ack; a share is a statistical drop.
  - Batched ingest also supported: POST /v1/shares:batch with an array,
    to amortize request overhead from high-volume producers.

Serving / read API (the data plane)

GET /v1/top?window=5m&k=100&category=sports
  window   enum   5m | 1h | 24h        (the three fixed sliding windows)
  k        int    1..100 (capped)
  category string optional; region optional -> selects the board
  200 OK   (served from the ~1s Top-K cache, identical for all callers)
  {
    "window": "5m",
    "generated_at": 1752451205,
    "top": [
      { "rank": 1, "article_id": "a_9f2c", "count": 184203, "error": 12 },
      { "rank": 2, "article_id": "a_1c77", "count": 173540, "error": 8  }
    ]
  }
  - "count" is the approximate windowed share count (CMS estimate).
  - "error" is the Space-Saving overcount bound; true count is in
    [count - error, count]. Usually shown only in debug.

Data model: in-memory structures + a durable rebuild log

There is no operational SQL/NoSQL table on the hot path - the counting structures live in RAM, because the access pattern is “increment 100K/sec and read an aggregate,” which no disk-backed store serves at this latency. The only persisted thing is the raw stream, for rebuild and offline analytics.

In-memory per Counter Shard (the fast, derived structures):

Per shard, per window granularity (5m/1h share the fine ring; 24h coarse):

  CMS ring:   array of per-bucket Count-Min Sketches
     fine  ring: 360 buckets x 16 MB  (10s buckets, covers up to 1h)
     coarse ring: 288 buckets x 16 MB (5min buckets, covers 24h)
     each bucket = uint32[d=4][w=2^20], conservative-update

  Space-Saving summary (one per window: 5m, 1h, 24h):
     hash map  id -> {count:uint64, error:uint64}
     + a min-structure (bucket-list / heap) over counts for O(1) min + evict
     m ≈ 10,000 monitored ids  (>> K)

  Total per shard ≈ (360 + 288) * 16 MB + a few MB ≈ ~10 GB
     -> pick M shards so per-shard memory is comfortable (e.g. 8-16 shards).

Durable raw share log (off the hot path, for rebuild + analytics):

share_event   (append-only, to Kafka -> object store / warehouse)
  article_id   TEXT
  category     TEXT
  region       TEXT
  ts           TIMESTAMP (event time)
  -- partitioned by hour; ~100 GB/day; NEVER read by the serving path.
  -- used to (a) rebuild a shard's rings after a crash by replaying the
  --   last 24h, and (b) run exact offline top-K for auditing the approx.

Why RAM structures and not a database: the operation is “increment approximate counters 100K times/sec and read a windowed aggregate in milliseconds.” A B-tree or LSM store cannot ingest that write rate into 100M keys, and a GROUP BY ... ORDER BY LIMIT over a day of events is a scan, not a millisecond read. Sketches and heaps in RAM are the only thing that fits; the durable log is what makes those RAM structures disposable.

Top-K cache:

key:   top:{window}:{category?}:{region?}     e.g. top:5m, top:1h:sports
value: serialized global top-K list
TTL:   ~1s (the Aggregator recomputes and overwrites each second)

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each:

1. Memory blowup from exact per-key counters (breaks first, conceptually). 100M articles times ~300 time buckets is ~240 GB of mostly-zero counters. Fix: Count-Min Sketch - memory fixed by target error (~16 MB/bucket), independent of the 100M key count. Trade a tiny bounded overcount for ~50x less memory. This is the move that makes the whole thing possible.

2. Ingest throughput at viral spikes (~50K-100K/sec). Synchronous, durable-per-share counting cannot keep up and would blow the ack budget. Fix: fire-and-forget ingest (ack after enqueue, not after a durable write) plus Kafka as a shock absorber - a spike drains at whatever rate consumers sustain, trading a couple seconds of freshness lag for never blocking. Sharding spreads the count updates across M nodes.

3. Sliding-window expiry. A “last 5 minutes” count must forget old shares without re-scanning raw data. Fix: time-bucketed CMS rings - a window count is the sum over its buckets, and old buckets rotate out and get zeroed, so the window slides for free. Two tiers (10s fine, 5min coarse) balance precision against ring size.

4. Sorting the full key space for top-K. Enumerating and ranking 100M articles per query is pure waste. Fix: Space-Saving keeps a bounded top-m summary of heavy hitters per window, so we only ever consider a few thousand candidates. The cold tail never enters the top-K path.

5. Hot article / hot partition. A single viral article can be a huge fraction of all shares, hammering one Kafka partition and one shard. Fix: the count is still a cheap increment, so one hot key is tolerable; if a single article truly saturates a partition, split its counting with a random suffix (hash(article_id + rand[0..s)) into s sub-counters, summed at read) to spread the increments, accepting a trivial merge step. The top-K read is one cached answer for everyone, so read-side hot keys are a non-issue.

6. Distributed merge dropping global winners. Taking exactly top-K per shard misses articles that rank below K on their home shard but are globally heavy. Fix: over-fetch top-m per shard (m » K, exactly the Space-Saving summary), union, and re-rank on exact per-article counts. Article-hash sharding keeps each count whole, so the merge never sums partial counts.

7. Aggregator as a single point of failure. One merger recomputing the cache is a SPOF. Fix: run multiple stateless Aggregators; each can independently pull shard top-m and refresh the cache (last writer wins, they compute the same thing). Shards themselves run replicas fed by the same Kafka partitions (or restored from the durable log) for HA.

8. Shard loss loses in-flight counts. The structures are in RAM; a crash drops recent counts. Fix: the durable raw log lets a restarting shard replay the last 24 hours to rebuild its rings and summaries. During the gap, reads degrade to slightly stale from cache. We accept losing a few seconds of un-replayed counts because the product is an approximate aggregate - this is the durability trade we made up front.

9. Clock skew across ingest/shard nodes. Bucketing by wall clock across machines can misplace shares near boundaries. Fix: bucket by the event ts stamped at ingest (single clock domain per ingest node, NTP-synced), and give buckets a small grace overlap so a share arriving slightly late still lands in the right bucket. Second-level skew is invisible against 5m/1h/24h windows.

Wrap-Up

The trade-offs that define this design:

  • Approximate over exact. We use a Count-Min Sketch specifically because its memory is fixed by the error we tolerate, not by the 100M articles, turning 240 GB of exact counters into ~16 MB per bucket. We pay a tiny, bounded, one-directional overcount that is noise for the heavy hitters we actually rank. This single choice is what makes counting a firehose over 100M keys feasible.
  • Buckets over rescans. Sliding windows expire by rotating time-bucketed sketches out of a ring and summing the rest, so “last 5 minutes” forgets old shares with no delete job and no scan of raw data - the window slides because the ring head moves.
  • Bounded top-K over full sort. Space-Saving keeps only the few thousand heavy-hitter candidates per window, so we never enumerate or sort the 100M cold tail; the sketch-ring then gives those candidates precise windowed counts at merge time.
  • Article-hash shard over time shard. We partition by hash(article_id) so each article’s count stays whole on one node, making the distributed merge a plain union-and-sort with exact per-candidate counts - and we over-fetch top-m per shard so a globally-heavy-but-locally-#K+1 article is never dropped.
  • Throughput over per-event durability. We ack shares fire-and-forget and count asynchronously, keeping only a durable raw log for rebuild, because the product is a statistical aggregate where losing a few seconds of shares is a rounding error - the opposite of a system of record.

One-line summary: a fire-and-forget share firehose sharded by article hash into per-bucket Count-Min Sketch rings and per-window Space-Saving summaries - answering the last 5-minute, 1-hour, and 24-hour top-K by rotating buckets to expire old counts, over-fetching heavy-hitter candidates per shard, and merging on exact per-article counts served from a one-second cache - trading a tiny bounded count error for real-time top-K over 100M articles and a billion shares a day.