Everyone thinks App Store rankings are a SELECT app_id, COUNT(*) FROM downloads GROUP BY app_id ORDER BY count DESC LIMIT 100. Count the downloads, sort, take the top 100, done. Then you look at a real store and the assumption falls apart in three separate ways at once. First, “trending” is not the same as “most downloaded” - a five-year-old app with a huge install base racks up more downloads per minute than a brand-new hit, yet the new hit is what is trending, so trending has to measure velocity, not volume. Second, there is not one chart, there is a chart per country times per category times per window (top free, top paid, top grossing), which is tens of thousands of distinct top-K lists that all have to stay fresh. Third, the windows are not all the same shape - “top 100 trending every minute” is a fast rolling score, “top 1000 daily” and “weekly” are rolling windows that must expire old events, and “top 1000 all-time” is a monotonic cumulative counter that must never forget anything. One naive GROUP BY cannot be all three.

So the problem is really: compute many thousands of top-K boards, each over a different notion of “score,” at a billion events a day, refreshing the fast ones every minute, without recomputing a day of events on every query and without storing an exact counter for every (app, country, category, bucket) combination. The design comes down to four decisions: how you define and compute a trending score that rewards acceleration rather than raw count, how you keep rolling windows (daily, weekly) expiring old events while all-time keeps a running total, how you avoid materializing tens of thousands of boards from scratch (hierarchical rollup plus lazy materialization), and how you shard and merge so the top-K per board is correct and cheap. Let me do it properly.

Functional Requirements (FR)

In scope:

  • Record an event. A download, install, in-app purchase, or app open is counted as a ranking signal. This is the firehose - a billion events a day.
  • Top 100 trending, refreshed every minute. The K=100 apps whose download velocity is accelerating fastest right now, recomputed at least once per minute so a launch shows up almost immediately.
  • Top 1000 daily, weekly, and all-time. Larger K=1000 boards over three window shapes: the trailing 24 hours, the trailing 7 days, and cumulative since launch.
  • Per-country and per-category boards. Charts exist per storefront (roughly 175 countries) and per category (roughly 30 categories), plus an “all categories” overall chart. Same machinery, many more keys.
  • Chart variants: top free, top paid, top grossing. Free and paid rank on install count; grossing ranks on revenue. Same top-K pipeline, different score input.
  • Near-real-time freshness for trending. An event should move an app on the trending board within about a minute. Daily and weekly can tolerate a minute or two of lag; all-time can tolerate a few minutes.

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

  • Exact per-app download totals to the unit. For a chart the product is the ranked order, not an audited count. Being off by a fraction of a percent never changes who is in the top 100 in a way anyone cares about. Financial-grade revenue accounting for grossing lives in a separate billing system; we consume a revenue number from it.
  • Arbitrary historical windows. “Top apps between week 3 and week 5 of last year” is an offline analytics query against a warehouse, not this system. We serve fixed windows: trending (minute), daily, weekly, all-time.
  • Per-app rank on demand below K. This is not the leaderboard problem (“where does my app rank if it is 40,000th”). We return the top K per board; ranks below K are a separate query.
  • Fraud and manipulation. Bot-driven install farms and review rings are a separate abuse pipeline. We assume events arriving here are already scrubbed, and note where that gate sits.
  • Editorial overrides. Human-curated “Apps We Love” placement is a CMS feature layered on top, not part of the ranking engine.

The one decision that drives everything: there is no single “count” - there are several score functions (velocity, rolling count, cumulative count, revenue) over the same event stream, each feeding thousands of per-country per-category top-K boards. So the design is dominated by a shared counting substrate, a trending score built on decayed rates, a split between rolling and cumulative windows, and a materialization strategy that does not recompute every board from scratch.

Non-Functional Requirements (NFR)

  • Scale: roughly 5M distinct apps. 1B events/day. Average ingest about 11.6K events/sec, but launches are spiky - a hit game or a viral app drives bursts, so size for a peak of about 50K events/sec. Read traffic (people browsing charts) is high, but any given board’s answer is identical for everyone in that country and category, so it is trivially cacheable and collapses to roughly one recompute per board per minute.
  • Latency: chart read p99 under about 20ms (served from cache, sub-millisecond hit). Event-ingest ack p99 under about 10ms - an event is fire-and-forget, we ack fast and count asynchronously.
  • Freshness: about 60 seconds from an event to the trending board reflecting it; a minute or two for daily and weekly; a few minutes for all-time. None of these is synchronous.
  • Availability: 99.9%+. Charts are a prominent, front-and-center surface. They should degrade to slightly stale rather than error. Losing a few seconds of counts during a failover is acceptable.
  • Consistency: eventual and approximate by design. Two nodes may briefly disagree on rank 97 versus 98, or on the exact velocity of an app. That is fine. What is not fine is a dead app showing as #1, or the daily board failing to expire yesterday’s events.
  • Durability: medium. Individual events do not need per-event durability on the counting path (losing 50ms of installs shifts a chart by a rounding error), but the all-time cumulative totals and revenue must be durable and exact-enough, because they persist forever and errors there compound instead of aging out. We split durability by window shape: rolling windows are cheap and disposable, cumulative totals are persisted and reconciled.

Back-of-the-Envelope Estimation (BoE)

Let me ground the design in numbers.

Traffic:

Distinct apps:                   5,000,000     (5M)
Events per day:                  1,000,000,000 (1B)   (installs, opens, purchases)

Average ingest QPS = 1B / 86,400 sec ≈ 11,574 events/sec
Launch spikes (a hit drops, everyone downloads in one hour):
Peak ingest QPS (say 4-5x avg)   ≈ 50,000 events/sec

Reads (chart browses): very high, but ONE answer per (country, category,
  chart, window). Every user in the US Games top-free chart sees the SAME
  list -> collapses to ~1 recompute per board per minute, served from cache.

The daily average of about 11.6K/sec is the honest steady state; the design must survive the roughly 50K/sec launch burst. The read side is a non-event because everyone browsing the same chart gets the same list.

How many boards is “a chart per country per category”?

Countries / storefronts:                 ~175
Categories (+ "all categories"):          ~31
Chart types (free / paid / grossing):       3
Windows (trending / daily / weekly / all): 4

Boards = 175 * 31 * 3 * 4 ≈ 65,000 distinct top-K lists.

Not all are equally hot: US + a few big storefronts and a handful of
categories are 90% of reads. The long tail (Games in Andorra, weekly)
is browsed rarely. This skew is what makes lazy materialization work.

Sixty-five thousand boards is the number that kills the naive “compute each board independently every minute” approach. If each recompute scanned events, that is 65K scans per minute. The design instead computes a small shared substrate and derives boards from it.

Counting memory - is exact affordable here?

Unlike a 100M-key firehose, apps number ~5M, and each app charts in only
a handful of the ~175 countries. So the live (app, country) pairs are
sparse - say each app is active in ~10 countries on average:
  5M apps * ~10 countries = 50M (app, country) counting keys.

Per key we hold rolling-window buckets (see deep dive):
  daily  ring: 24 * 4 = 96  fifteen-minute buckets
  weekly ring: 7  * 24 = 168 hourly buckets (reuse daily's aggregates)
  trending: a small decayed rate state (a few counters)
  Round it to ~300 small counters per key.

  50M keys * ~300 * 4 bytes ≈ 60 GB   -> sharded across nodes, fits.

So here exact counting is affordable - 5M apps is small enough that we do not need a Count-Min Sketch to fit in memory. We keep exact per-bucket counters, sharded. (If the key space blew up - say per-device or per-session keys - we would reach for sketches; I note that as the escape hatch but do not need it at 5M apps.)

All-time cumulative and revenue (must be durable):

Global all-time count per app:     5M * 8 bytes = 40 MB
Per-country all-time (sparse):     50M pairs * 8 bytes = 400 MB
Revenue totals (grossing):         50M pairs * 8 bytes = 400 MB
  -> tiny, but PERSISTED (durable KV), not just RAM. Errors here never
     age out, so they are backed by a durable store + periodic reconcile.

Storage / bandwidth:

Chart response: 1000 entries * ~60 bytes ≈ 60 KB, served from cache.
Raw event: ~60-100 bytes on the wire.
Ingest bandwidth at 50K/sec * 100 B ≈ 5 MB/sec. Modest.
Durable raw log (rebuild + analytics): 1B/day * 100 B = 100 GB/day
  -> cheap object storage / warehouse, OFF the hot path.

Bandwidth is a non-issue. The scarce resources are ingest throughput (absorbing 50K events/sec of count updates) and recompute cost (deriving 65K boards without scanning per board). Size for those.

High-Level Design (HLD)

The system splits into a write path that takes the event firehose, shards it by app, and folds each event into exact per-bucket counters plus a decayed trending-rate state; a scoring/rollup layer that turns raw counts into the four score functions and rolls app-level counts up the country and category hierarchy; and a read path that materializes the top-K per board (hot boards eagerly each minute, cold boards lazily on demand) and serves from cache.

              WRITE PATH (event firehose, ~50K/sec peak)
   ┌──────────────────────────────────────────────────────────────────┐
   │  Devices / Store  --event-->  API GW / LB  --> Event Ingest       │
   │                                                 (stateless)        │
   │                                202 Accepted (fire-and-forget)      │
   └───────────────────────────────────┬───────────────────────────────┘
                                        │ produce {app_id, country, cat,
                                        │          type, revenue?, ts}
                              ┌─────────▼──────────────────┐
                              │  Kafka (partitioned by      │
                              │  hash(app_id), N parts)     │ <- same app
                              └─────────┬──────────────────┘    same shard
                                        │
          ┌─────────────────────────────┼─────────────────────────────┐
          ▼                             ▼                             ▼
 ┌──────────────────┐        ┌──────────────────┐        ┌──────────────────┐
 │ Counter Shard 0  │        │ Counter Shard 1  │  ...   │ Counter Shard M  │
 │ ---------------- │        │                  │        │                  │
 │ per (app,country)│        │ exact bucket     │        │ exact bucket     │
 │ - daily ring     │        │ rings + trending │        │ rings + trending │
 │ - weekly ring    │        │ decayed rate     │        │ decayed rate     │
 │ - trending EWMA  │        │ + all-time total │        │ + all-time total │
 │ - all-time total │        │   (-> durable KV)│        │   (-> durable KV)│
 └────────┬─────────┘        └────────┬─────────┘        └────────┬─────────┘
          │  each shard exposes local top-m per (country,cat,window,score) │
          └─────────────────────────────┬─────────────────────────────────┘
                                         ▼   ROLLUP + MERGE
                              ┌─────────────────────────┐
                              │  Rollup / Aggregator     │  every minute:
                              │  - fan-in shard top-m    │  hot boards eager
                              │  - roll app -> category   │  cold boards lazy
                              │    -> "all categories"    │  (on first read)
                              │  - merge -> board top-K   │
                              └───────────┬─────────────┘
                                 ┌────────▼─────────┐
                                 │  Board Cache     │  key = country:cat:
                                 │  (Redis / CDN)   │  chart:window
                                 └──────────────────┘

  (side path) Kafka --tee--> Durable Raw Log (object store / warehouse)
                             for rebuild + all-time reconcile + analytics
  Durable KV (all-time totals, revenue) <-- persisted, reconciled nightly

Request flow - an event (the write hot path):

  1. A device or the store backend sends POST /v1/events with app_id, country, category, type (install / open / purchase), and revenue for purchases. The Event Ingest service is stateless: stamp a server-side ts, produce to Kafka, return 202 Accepted. No durable write blocks the ack.
  2. Kafka partitions by hash(app_id) so every event for an app lands on the same partition and therefore the same counter shard. Each app’s counts stay whole on one node - the routing decision that makes the merge correct later.
  3. The Counter Shard consumer folds the event into the current bucket of the app’s daily and weekly rings (per country), updates the trending decayed-rate state, and increments the durable all-time total (and revenue, for purchases). It does this per (app, country); the category is carried on the event so rollup can group by it.
  4. Within about a minute the counters and rate states reflect the event, so the next rollup sees it.

Request flow - a chart read:

  1. A user opens “US, Games, Top Free, trending.” The app calls GET /v1/charts?country=us&category=games&chart=free&window=trending&k=100. The Read Service returns the cached board for that key - one precomputed answer for everyone. Sub-millisecond.
  2. The cache is kept warm by the Rollup/Aggregator. For hot boards (big countries and categories) it recomputes every minute: pull each shard’s local top-m candidates for the relevant (country, window, score), roll app-level scores up into category and “all categories” buckets, merge across shards, take the top-K, overwrite the cache. For cold boards (long-tail country plus category plus weekly) it materializes lazily on first read and caches the result for a few minutes, so we never spend minute-by-minute compute on boards nobody is watching.

The key architectural insight: counting is exact and sharded-by-app, scores are derived (velocity, rolling, cumulative, revenue), boards roll up a hierarchy (app to category to overall), and only hot boards are materialized eagerly. We never scan events per board, we never store a separate counter per board (boards are derived from per-app counts), and the four windows use the mechanism that fits each (decayed rate, bucket ring, or persisted total).

Component Deep Dive

Four hard parts: (1) defining a trending score that measures velocity, not volume; (2) making rolling windows (daily, weekly) expire while all-time keeps a running total; (3) materializing 65K boards without recomputing each from scratch; and (4) sharding and merging the top-K correctly. I walk each from naive to scalable.

1. The Trending Score: Raw Count -> Rate -> Decayed Velocity -> Acceleration

“Trending” is where most candidates go wrong, because the obvious answer is quietly incorrect.

Approach A: Rank by downloads in the last minute (the naive instinct)

Count installs in the trailing minute, sort, take the top 100.

Where it breaks: this is a popularity chart, not a trending chart. A giant incumbent app with 100M installed users gets more installs per minute from pure background churn than a brand-new app in its breakout hour. So the “trending” board is just the “biggest apps” board with more noise, and the new hit - the thing the chart exists to surface - never appears. Raw count in a short window is dominated by baseline volume, not by change.

Approach B: Rank by growth rate (this window vs the previous window)

Compute the install rate now and compare to the recent baseline: score = rate_now / rate_baseline, or the delta rate_now - rate_baseline. An app doubling its rate scores high even if its absolute numbers are small.

This is the right idea - trending is about acceleration - but the naive form has two problems. A ratio explodes for tiny apps (an app going from 2 to 8 installs a minute is “4x” but is not trending anything), and a single spiky minute makes the score jitter wildly minute to minute. We need to reward genuine acceleration while damping noise and small-count nonsense.

Approach C: Exponentially decayed velocity with a baseline and a floor

Keep two exponentially weighted moving averages (EWMAs) of the install rate per (app, country): a fast one (short half-life, say 10 minutes) capturing “right now” and a slow one (long half-life, say 24 hours) capturing the baseline. The trending score is how far the fast rate has pulled above the slow baseline, normalized and floored so tiny apps do not blow up.

Per (app, country), maintain two decayed counters updated on each event
and decayed by elapsed time (no per-event scan, just multiply-and-add):

  on event at time t:
     dt_fast = t - last_update
     fast = fast * exp(-lambda_fast * dt_fast) + 1      # lambda_fast: ~10min
     slow = slow * exp(-lambda_slow * dt_fast) + 1      # lambda_slow: ~24h
     last_update = t

  trending_score = (fast - slow) / sqrt(slow + C)

  - fast - slow: absolute acceleration (how much the recent rate exceeds
    the long-run baseline). New apps have slow ~ 0, so their whole fast
    rate counts as acceleration - exactly what we want.
  - divide by sqrt(slow + C): normalize so a huge incumbent needs a huge
    absolute jump to score high, while keeping small apps from exploding.
    C is a floor constant (a minimum baseline) that kills tiny-count noise.
  - EWMA decay damps single-minute spikes automatically.

Why EWMA rather than fixed buckets for trending: an EWMA is O(1) state per key (two floats plus a timestamp) and needs no ring of buckets, because “recent” is encoded in the decay, not in which buckets you sum. It updates in constant time on each event and can be read at any instant by applying the decay for the elapsed time. That is far cheaper than summing bucket rings every minute for 5M apps, and it gives a smooth score.

The score is intentionally not a count. Two apps with identical trending scores can have wildly different install totals; that is correct, because the chart ranks momentum. We expose the underlying install count separately for the daily/weekly/all-time boards, which do rank on volume.

One subtlety: EWMA state is cheap but is derived, not a ledger. If a shard restarts, we rebuild the fast/slow EWMAs by replaying the trailing 24h of events from the durable log (the slow half-life means only the last day matters). We never need to persist the EWMA itself.

2. Rolling Windows vs Cumulative: Bucket Ring vs Running Total

The daily, weekly, and all-time boards all rank on volume, but the windows are fundamentally different shapes, and treating them the same is a mistake.

Approach A: Treat every window as “sum events in the window” (naive)

Store every event with a timestamp; on query, scan the trailing 24h / 7d / all-time and aggregate.

Where it breaks: the all-time scan is the entire event history forever, and even the weekly scan is billions of rows. You cannot re-aggregate that per minute per board. And “all-time” as a windowed scan is absurd - it is a monotonic total that only ever grows; scanning history to recompute a number that never decreases is pure waste. Rolling and cumulative windows want opposite mechanisms.

Approach B: Bucket rings for rolling, a persisted running total for cumulative

Split by window shape:

DAILY (trailing 24h)  -> ring of 96 fifteen-minute buckets per (app,country)
  count_daily(app,country) = sum over all 96 buckets
  When wall clock crosses a 15-min boundary, advance ring head, zero the
  new slot (reuse memory). The bucket 24h1m old is simply no longer summed
  -> the window slides for free, no delete job, no scan.

WEEKLY (trailing 7d) -> ring of 168 hourly buckets per (app,country)
  Reuse the daily 15-min buckets by rolling four of them into each hourly
  bucket as they age, so we do not keep two full-resolution rings.
  count_weekly = sum over all 168 hourly buckets.

ALL-TIME (cumulative) -> a single running total per (app,country), PERSISTED
  on event: total += 1   (and revenue += amount for grossing)
  No ring, no expiry. It is a durable counter in a KV store, updated
  in place. Reading it is O(1); it never needs recomputation.
  • Rolling windows expire by rotation. The ring head moving is the entire expiry mechanism. Query cost is bounded - sum 96 or 168 small integers for a candidate app, not a scan of raw events.
  • Cumulative is a durable counter, not a window. Because it never decreases and never forgets, it wants a persisted running total updated in place, not a bucket ring. This is also the one number that must be exact-enough and durable, so it lives in a durable KV store (backed by the raw log for reconcile), while the rolling rings live in RAM and are disposable.
  • Two resolutions, not two full rings. We keep 15-minute buckets fresh for daily and roll them up to hourly for weekly as they cross the 24h line, so weekly costs 168 small counters, not a second high-resolution ring.

Approach C: Handling the all-time board’s size and drift

All-time ranks 5M apps by a monotonic total, and the top 1000 is very stable minute to minute (the #1 app of all time does not change on a Tuesday). So we do not recompute the all-time top 1000 every minute from 5M totals. Instead:

  • Maintain a bounded top-N candidate set (say top 5000 by all-time total) per (country, category), updated incrementally: when an app’s total crosses the current #5000’s total, it enters the candidate set; otherwise it cannot be in the top 1000 and is ignored.
  • Recompute the all-time board from that small candidate set every few minutes. An app 4-millionth by all-time downloads can never jump into the top 1000 in one minute, so ignoring it is safe.
  • Reconcile the durable all-time totals against the raw log nightly to correct any drift from dropped or double-counted events, since cumulative errors never age out.

The rolling boards get the opposite treatment: they do churn (a launch can enter the daily top 1000 fast), so they are recomputed from shard top-m every minute.

3. Materializing 65K Boards: Per-Board Compute -> Hierarchical Rollup + Lazy Materialization

We can now score any app for any window. But there are about 65,000 boards, and computing each independently is the scaling wall.

Approach A: Compute each board from scratch every minute (naive)

For each (country, category, chart, window), gather the relevant apps, score, sort, take K.

Where it breaks: 65K boards recomputed every minute is 65K sort-and-merge jobs a minute, most of them for boards nobody is looking at (weekly Games chart for a tiny storefront). And it redoes shared work: the “US, all categories” board and the “US, Games” board both need US app scores; computing them separately re-derives the same per-app numbers. The work is quadratic in the wrong dimensions.

Approach B: Hierarchical rollup - compute the substrate once, derive boards

Boards form a hierarchy. Category boards roll up into the “all categories” board; and every board is a filter-and-sort over the same per-app scores. So compute the shared substrate once:

Level 0 (substrate): per (app, country) scores for each window/score-fn,
  already maintained by the counter shards.

Level 1: group apps by category within a country, keep a per-category
  top-m candidate list. An app's category is fixed metadata, so this is a
  group-by, not a recompute of scores.

Level 2 ("all categories"): merge the per-category top-m lists within a
  country into the country-wide top-m. Because top-K of a union is a
  subset of the per-group top-m (with sufficient m), we never rescan all
  apps - we merge a few thousand candidates.

Boards are then a projection:
  (country, category, chart, window) -> filter substrate by chart type
  (free/paid/grossing selects the score input), take top-K.

The trending, daily, weekly, and all-time boards for a given (country, category) all read the same candidate set, differing only in which score they sort by. So one candidate gather per (country, category) feeds four windows times three chart types at once. That collapses the 65K independent jobs into roughly 175 * 31 = ~5400 candidate gathers, each producing a dozen boards.

Approach C: Lazy materialization for the cold long tail

Even 5400 gathers per minute is wasteful when 90% of reads hit a few hundred hot boards. So split by heat:

HOT boards (big countries x big categories, ~few hundred):
  materialized EAGERLY every minute by the Rollup workers. Always warm.

COLD boards (long tail, ~64,000):
  materialized LAZILY on first read:
    on GET, check cache; on miss, gather that board's candidates from the
    substrate, compute top-K, cache with a 2-5 min TTL. Subsequent reads
    in that window hit cache. If nobody reads it, we never compute it.
  • Heat is measured, not guessed. A rolling read-rate counter per board key promotes a board to “hot” (eager) if its read QPS crosses a threshold, and demotes it when it goes quiet. Launch-day of a hit game in a small country promotes that country’s board automatically.
  • Cold-read latency. The first cold read pays a gather-and-sort (tens of ms over a few thousand candidates), which is acceptable for a rarely-visited board and warm for everyone after. We can also precompute the substrate (candidate lists) for all countries every minute but only materialize the final board on read, so even a cold first read is a quick merge, not a scan.

This is the move that makes 65K boards tractable: compute a shared, hierarchical substrate a few thousand times a minute, and turn substrate into final boards eagerly only for the hot few hundred and lazily for the rest.

4. Sharding and the Distributed Merge

One node cannot ingest 50K events/sec and hold every app’s rings, EWMAs, and totals, so we shard by hash(app_id) across M nodes. Each shard owns its apps’ counters. How do we get a correct global top-K per board from M shards?

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

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

Where this is wrong: an app that is globally #400 on the US Games daily board might rank only, say, #620 on its own shard (that shard happens to hold several bigger apps), so it never appears in that shard’s local top-K and is silently dropped from the merge. Taking exactly top-K per shard loses global winners that rank below K locally. Classic distributed top-K trap.

Approach B: Over-fetch top-m per shard, merge on exact per-app scores

Two fixes, together:

  1. Over-fetch. Ask each shard for its local top-m where m is comfortably larger than K (say m = 5000 for a K=1000 board). Because we shard by app, an app lives entirely on one shard, so its shard-local score is its global score - it only has to make its own shard’s top-m, which any globally-heavy app easily does. A generous m makes a missed global winner negligibly unlikely for skewed chart streams.
  2. Exact scores at merge. Each candidate’s score comes from exactly one shard - no summing partial counts across shards, no compounding error. The Rollup collects the union of per-shard top-m candidates with their scores, groups by category, rolls up, sorts, takes top-K per board.
Rollup (per country, per window/score, ~1x/min for hot; on-read for cold):
  candidates = {}
  for shard in shards:
     for (app_id, category, score) in shard.top_m(country, window, scorefn):
        candidates[app_id] = (category, score)      # local = global score
  # hierarchical rollup:
  for cat in categories:
     board[country][cat] = topK(filter(candidates, category=cat))
  board[country]["all"] = topK(candidates)          # merge across categories
  cache.set(key(country, cat, chart, window), board, ttl=...)

Because we partition by hash(app_id), each app’s score stays whole on one node, so the merge is a plain union-and-sort with no double counting. The alternative shard keys are worse:

Shard keyAn app’s scoreMerge correctnessVerdict
Hash of app_idcomplete on one shardunion of top-m, exact per candidateRight - keeps scores whole
By countrycomplete within a countryfine per-country, but one hot country (US) overloads one shardRejected - hot-country skew
Round-robin / by timesplit across all shardsmust sum partial counts, errors compoundRejected - fragments scores
Exactly top-K per shardcompletemisses globally-heavy-but-locally-#K+1 appsRejected - drops winners

The residual approximation is the EWMA smoothing on trending and the top-m truncation, both bounded and both invisible for the heavy hitters that actually chart. We serve the result as a “chart,” not an audited ledger - exactly the product.

API Design & Data Schema

Ingestion API (the write plane)

POST /v1/events
  {
    "app_id": "app_7f3a",
    "country": "us",
    "category": "games",
    "type": "install",           # install | open | purchase | update
    "revenue": 4.99,             # present only for purchases (grossing)
    "ts": 1752624000            # server stamps if absent
  }
  202 Accepted    -> counted asynchronously (~1 min), fire-and-forget

Notes:
  - No durable write blocks the ack for install/open counting.
  - Purchases ALSO write to the durable revenue ledger (grossing must be
    exact-enough), so purchase acks confirm the durable enqueue.
  - Batched ingest: POST /v1/events:batch with an array, to amortize
    request overhead from high-volume store backends.

Serving / read API (the data plane)

GET /v1/charts?country=us&category=games&chart=free&window=trending&k=100
  country  string  storefront code
  category string  category or "all"
  chart    enum    free | paid | grossing
  window   enum    trending | daily | weekly | all
  k        int     capped at 100 for trending, 1000 for daily/weekly/all
  200 OK   (served from the board cache, identical for all callers)
  {
    "country": "us", "category": "games", "chart": "free",
    "window": "trending", "generated_at": 1752624060,
    "top": [
      { "rank": 1, "app_id": "app_7f3a", "score": 8123.4, "installs_24h": 51200 },
      { "rank": 2, "app_id": "app_1c90", "score": 7740.1, "installs_24h": 48010 }
    ]
  }
  - For window=trending, "score" is the decayed-velocity score;
    "installs_24h" is shown for context, not for ranking.
  - For daily/weekly/all, ranking is by the volume count (or revenue for
    grossing); "score" carries that count.

Data model: RAM structures, a durable KV, and a rebuild log

The rolling-window counting lives in RAM (the access pattern is “increment 50K/sec and read an aggregate,” which no disk store serves at this latency). The cumulative totals and revenue live in a durable KV because they persist forever. The raw stream is persisted for rebuild and reconcile.

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

Per (app_id, country) owned by this shard:
  daily ring:   uint32[96]    (15-min buckets, trailing 24h)
  weekly ring:  uint32[168]   (hourly buckets, trailing 7d, rolled up)
  trending:     { fast:float, slow:float, last_ts:int }  (EWMA state)
  app metadata: category, chart eligibility (cached from a metadata svc)

Per (country, window, scorefn) the shard maintains a bounded top-m
  candidate structure (a min-heap of size m over the local scores) so it
  can answer shard.top_m() without scanning all its apps.

Sizing: ~50M (app,country) pairs / M shards. With M ~ 16-32 shards,
  each holds a couple million keys * ~300 counters -> a few GB RAM.

Durable KV (cumulative, must persist):

all_time_total   key = (app_id, country) -> { installs:uint64, revenue:uint64 }
  - updated in place on each event (buffered + flushed, not per-event fsync)
  - backed by the raw log for nightly reconcile
  - powers the all-time and grossing-all-time boards
  Store: a durable KV (e.g. RocksDB-backed / DynamoDB), NOT the RAM shard.

Durable raw event log (off hot path, rebuild + reconcile + analytics):

event   (append-only, Kafka -> object store / warehouse)
  app_id     TEXT
  country    TEXT
  category   TEXT
  type       TEXT
  revenue    DECIMAL   NULL
  ts         TIMESTAMP
  -- partitioned by hour; ~100 GB/day; never read by the serving path.
  -- used to (a) rebuild a shard's rings + EWMAs by replaying trailing 7d,
  --   (b) nightly-reconcile the durable all-time totals, (c) offline audit.

Board cache:

key:   board:{country}:{category}:{chart}:{window}   e.g. board:us:games:free:trending
value: serialized top-K list
TTL:   ~60s for eager hot boards (overwritten each minute),
       ~2-5min for lazy cold boards (recomputed on read after expiry)

SQL vs NoSQL: there is no relational hot path. Rolling counts are RAM structures (no store serves 50K writes/sec into tens of millions of keys at millisecond read). Cumulative totals are a durable KV (point-key increments, no joins, no scans - a NoSQL KV fits exactly; a B-tree SQL row per key would work but buys nothing over a KV here). App metadata (category, name, eligibility) is a small relational table (millions of rows, richly queried by the catalog service) cached into the shards. The raw log is columnar files in a warehouse for scans. Each store matches its access pattern.

Bottlenecks & Scaling

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

1. Trending computed as raw count (breaks correctness first). Ranking by installs-in-a-minute makes trending a popularity chart; new hits never surface. Fix: decayed-velocity score - two EWMAs (fast and slow) per (app, country), score = normalized acceleration (fast - slow) / sqrt(slow + C). O(1) state per key, no bucket ring, smooth and small-count-safe.

2. Recomputing a day of events per board (breaks at query time). Scanning the trailing window per board per minute is billions of rows. Fix: bucket rings for rolling windows (sum 96 or 168 small counters, expire by rotation) and a persisted running total for all-time. Never scan raw events on the serving path.

3. 65K boards computed independently. One sort-and-merge per board per minute, most for boards nobody watches, redoing shared work. Fix: hierarchical rollup (compute per-(country,category) candidate substrate once, derive all four windows and three charts from it) plus lazy materialization (eager for the hot few hundred, on-read for the cold long tail). Collapses 65K jobs to a few thousand substrate gathers.

4. Ingest throughput at launch spikes (~50K/sec). Synchronous, durable-per-event counting blows the ack budget. Fix: fire-and-forget ingest (ack after enqueue) plus Kafka as a shock absorber - a spike drains at consumer rate, trading a minute of freshness for never blocking. Sharding spreads updates across M nodes. (Purchases still confirm the durable revenue enqueue.)

5. Sorting the full 5M key space for top-K. Enumerating and ranking all apps per board is waste; 99% are cold. Fix: bounded top-m per shard per (country, window) - a size-m min-heap over local scores - so the merge only ever considers a few thousand candidates. The cold tail never enters the top-K path.

6. Hot app / hot partition. A viral launch is a huge fraction of all events, hammering one Kafka partition and one shard. Fix: the update is a cheap increment plus an EWMA touch, so one hot key is tolerable. If a single app truly saturates a partition, split its counting with a random suffix (hash(app_id + rand[0..s)) into s sub-counters, summed at read), accepting a trivial merge. Chart reads are one cached answer per board, so read-side hot keys are a non-issue.

7. Hot country skew. The US storefront dwarfs small ones; sharding by country would overload one node. Fix: shard by app, not country, so the US load spreads across all shards. The rollup groups by country at merge time from evenly-loaded shards.

8. All-time board and cumulative drift. All-time totals persist forever; dropped or double-counted events never age out, and recomputing the top 1000 from 5M totals every minute is wasteful. Fix: incremental top-N candidate set for all-time (an app enters only when it crosses the current #5000), recomputed every few minutes; and nightly reconcile of the durable totals against the raw log to correct drift.

9. Distributed merge dropping winners. Exact top-K per shard misses globally-heavy-but-locally-#K+1 apps. Fix: over-fetch top-m per shard (m » K), union, re-rank on exact per-app scores. App-hash sharding keeps each score whole, so no partial-count summing.

10. Rollup and cache as single points of failure. One Rollup worker refreshing the cache, or one cache node, is a SPOF. Fix: run multiple stateless Rollup workers (each pulls shard top-m and refreshes independently - last writer wins, they compute the same thing) and a replicated cache (Redis cluster / CDN). Shards run replicas fed by the same Kafka partitions (or restored from the raw log) for HA.

11. Shard loss loses in-flight rolling counts. The rings and EWMAs are in RAM; a crash drops recent counts. Fix: the durable raw log lets a restarting shard replay the trailing 7 days to rebuild rings and EWMAs; the durable KV holds the all-time totals so those survive independently. During the gap, reads degrade to slightly stale from cache. We accept losing a few seconds of un-replayed rolling counts because rolling charts are approximate aggregates.

12. Clock skew across nodes. Bucketing by wall clock across machines misplaces events near boundaries. Fix: bucket by the event ts stamped at ingest (single clock domain per ingest node, NTP-synced), with a small grace overlap so a slightly-late event lands in the right bucket. Sub-second skew is invisible against minute, day, and week windows.

Wrap-Up

The trade-offs that define this design:

  • Velocity over volume for trending. The trending board ranks a decayed-velocity score (fast - slow) / sqrt(slow + C), not a raw count, because the product is momentum - a breakout hit must beat a bigger, flatter incumbent. Two EWMAs give O(1) state per key and a smooth, small-count-safe score with no bucket ring.
  • Right mechanism per window shape. Daily and weekly are rolling windows that expire by rotating a bucket ring; all-time is a monotonic total kept as a persisted, durable counter and refreshed from a bounded candidate set. Treating cumulative as a windowed scan would be absurd; treating rolling as a running total would never forget.
  • Derive boards, do not recompute them. Sixty-five thousand per-country per-category per-window boards come from one shared per-(country,category) candidate substrate rolled up a hierarchy, materialized eagerly for the hot few hundred and lazily on-read for the cold tail. We never scan events per board and never store a counter per board.
  • App-hash shard over country or time shard. Partitioning by hash(app_id) keeps each app’s score whole on one node and spreads the hot US storefront across all shards, making the merge a plain union-and-sort; over-fetching top-m per shard keeps a globally-heavy-but-locally-#K+1 app from being dropped.
  • Split durability by window shape. Rolling counts are disposable RAM rebuilt from the raw log; cumulative totals and revenue are durable and nightly-reconciled, because their errors never age out. Ingest is fire-and-forget for counting throughput, exact-enough for grossing.

One-line summary: an app-hash-sharded event firehose feeding exact per-bucket rolling rings, decayed-velocity EWMAs, and durable cumulative totals - deriving tens of thousands of per-country per-category top-K boards from one hierarchical candidate substrate, materialized eagerly for hot charts and lazily for the cold tail - so top 100 trending refreshes every minute and top 1000 daily, weekly, and all-time stay fresh over a billion events a day, trading exactness for momentum-aware, real-time rankings.