Design Top K Rankings System (App Store / Amazon Bestsellers)

Event Ingestion (Purchase and Download Services)

Events enter the pipeline here, where keeping the ingestion path asynchronous ensures client writes and downstream reads remain fast.

• Validate incoming events through verified purchase confirmation and deduplicate by device identifier within a 24-hour window for free downloads
• Publish to the Kafka ranking-events topic using item_id as the partition key to guarantee item-level ordering
• Return HTTP 202 Accepted immediately because ranking aggregation is fully asynchronous
• Fraud filter: Velocity caps per IP and account are enforced in Flink before events qualify for heap updates

Stream Processing (Apache Flink): The Core Engine

This is where event aggregation happens, with the choice between exact and approximate counting serving as the critical architectural decision.

Approach 1: Exact Count Pipeline (for small-to-medium scale)

Flink Pipeline Topology (Dimension-Aware Multi-Board Aggregation):
  Source: Kafka topic 'ranking-events' (64 partitions distributing ingest throughput)
  KeyBy: (category, geography, metric, item_id) for item-level event accumulation
  Window: Tumbling or sliding window (e.g., 5-minute sliding or 24-hour tumbling)
  Aggregate: Incremental count accumulator per item within each dimension combination
  Keyed Process / Top K: Independent min-heap of size K per logical leaderboard key:
    Partition Worker
      ├── Dimension Key A (e.g., games:us:daily:downloads) → Min-Heap (size K)
      ├── Dimension Key B (e.g., productivity:global:hourly:revenue) → Min-Heap (size K)
      └── ...
  Merge Stage: Downstream reduce operator groups partial heaps by matching leaderboard key
  Sink: Redis Sorted Sets updated per board (ranking:{category}:{geography}:{period}:{metric})

Approach 2: Approximate Count (for massive scale): Count-Min Sketch + Heap

When tracking millions of unique items, maintaining exact counters for every entity in memory becomes cost-prohibitive.

Count-Min Sketch:

• Probabilistic data structure consisting of d hash functions across a matrix of w counters
• On an event for item X, hash the item key with each of the d functions and increment the corresponding d counters
• To query the frequency of item X, evaluate all d mapped counters and take their minimum value
• Space: O(w x d), requiring typically 2 MB of memory for under 0.1% error rate
• Error: The sketch always overcounts and never undercounts, with error bounded mathematically by ε = e / w

Min-Heap for Top K:

interface RankedItem {
  itemId: string;
  count: number;
}

// Maintain a min-heap of size K over incoming streaming events for a specific leaderboard
function updateTopKHeap(heap: MinHeap<RankedItem>, item: RankedItem, k: number): void {
  if (heap.size() < k) {
    heap.insert(item);
  } else if (item.count > heap.peek().count) {
    heap.poll();
    heap.insert(item);
  }
}

Combined Streaming Approach:

1. Count-Min Sketch tracks approximate counts across all items with minimal memory consumption
2. A Min-Heap of size K maintains the running set of current top candidates
3. When a new event arrives:
   • Increment the item count across sketch counters
   • Query the minimum counter estimate to check if the item qualifies for the top-K heap
   • If the count exceeds the heap root, insert or update the item in the heap
4. Periodically flush the local heap state to Redis sorted sets

Multi-Dimensional Keyed State Architecture:

Because the system supports multiple dimensions such as category, geography, time period, and metric (potentially yielding 12K to 20K independent ranking lists), streaming aggregation is keyed by the specific leaderboard dimensions:

Partition Worker
  ├── Dimension Key A (e.g., games:us:daily:downloads) → Top-K Min-Heap State
  ├── Dimension Key B (e.g., productivity:global:hourly:revenue) → Top-K Min-Heap State
  └── Dimension Key C (e.g., education:uk:weekly:ratings) → Top-K Min-Heap State

Kafka partitioning provides distributed ingest throughput across 64 partitions. Flink keyed state using KeyBy(category, geography, metric, item_id) maintains independent aggregation state for each relevant leaderboard. Crucially, the system does not maintain a single heap that incorrectly mixes unrelated leaderboards together. Downstream, local Top-K heap results from partition workers are merged exclusively for their corresponding category, geography, metric, and time window.

Time-Windowed Rankings

The choice between sliding and tumbling windows fundamentally changes how trending content feels to users, making it important to justify your selection.

# Time-windowed aggregation configurations
hourly_rankings:
  window_type: "sliding"
  duration: "1 hour"
  slide_interval: "5 minutes"
  use_case: "Near-real-time trending leaderboards"

daily_rankings:
  window_type: "tumbling"
  duration: "24 hours"
  use_case: "Official daily top lists"

weekly_rankings:
  window_type: "tumbling"
  duration: "7 days"
  use_case: "Weekly chart summaries"

all_time_rankings:
  window_type: "cumulative"
  duration: "unbounded"
  use_case: "All-time popular items without window resets"

Exponential Decay for Trending Calculations:

Instead of rigid fixed windows, exponential decay weights recent activity higher than historical events:

# Exponential decay scoring formula:
# score = sum(event_value * math.exp(-decay_rate * age_in_hours))
# decay_rate (lambda) = 0.1 yields a half-life of approximately 7 hours
# Recent events contribute substantially higher weight than historical events
import math

def compute_decay_score(event_value: float, age_in_hours: float, decay_rate: float = 0.1) -> float:
    return event_value * math.exp(-decay_rate * age_in_hours)

Redis (Ranking Cache)

Pre-computed rankings reside in memory so client reads never require on-the-fly sorting at query time.

Each ranking leaderboard is backed by a Redis Sorted Set with atomic update operations:

# Structure: Key = ranking:{category}:{geography}:{period}:{metric}, Score = metric score, Member = item_id

# Fetch top 10 items in games by downloads in the US for today
ZREVRANGE ranking:games:US:daily:downloads 0 9 WITHSCORES

# Look up an individual item's current weekly rank across global overall downloads
ZREVRANK ranking:overall:global:weekly:downloads app123

# Atomically update or insert an item score from Flink flush
ZADD ranking:games:US:daily:downloads 15432 app123

Ranking Service (Read API)

The read API delivers pre-computed leaderboards from cache, keeping computationally heavy sorting off the critical hot path.

• Stateless API instances serve read requests directly from Redis cluster replicas
• Querying ZREVRANGE ranking:{category}:{geography}:{period}:{metric} 0 K-1 returns the top-K leaderboard in under 5 milliseconds
• ZREVRANK executes fast single-item rank lookups across multiple leaderboards concurrently
• Historical point-in-time queries route to Cassandra snapshots generated by the hourly Spark batch process

Batch Pipeline (Spark): Reconciliation

An hourly batch reconciliation pipeline corrects any mathematical drift introduced by the approximate real-time stream processing layer.

• Hourly Spark batch jobs read raw, immutable event logs from Cassandra bucketed partitions in parallel (and alternatively from cloud object storage such as S3 in Parquet format), avoiding single monolithic partition scans
• The job computes exact rankings across all dimensions without probabilistic approximations
• It reconciles existing Redis sorted sets, correcting any Count-Min Sketch overcount drift
• The pipeline persists immutable historical ranking snapshots into Cassandra to support point-in-time queries

Event Bus Design (Kafka)

The event bus decouples ingestion producers from downstream processing consumers and absorbs burst traffic spikes.

topic: "ranking-events"
partitions: 64
partition_key: "item_id" # guarantees ordering per item within a partition
retention: "7 days"      # batch reconciliation replays from Cassandra when needed
replication_factor: 3
min_insync_replicas: 2

producer:
  idempotence: true # enable.idempotence=true to prevent duplicate counts
  payload:
    event_id: "string (UUID)"
    event_type: "purchase | download"
    item_id: "string"
    category: "string"
    value: "number"
    timestamp: "string (ISO 8601)"
    metadata: "object"

consumer_groups:
  flink_aggregator:
    description: "Maintains dimension-keyed Top-K min-heaps and Count-Min Sketches per leaderboard, flushing to Redis every 60 seconds"
  event_archiver:
    description: "Performs batch inserts into Cassandra ranking_events partitioned by date, hour, and deterministic bucket"
  fraud_detector:
    description: "Evaluates velocity and per-IP thresholds before events count toward rankings"

operational_paths:
  sync_path: "Validate payload, publish to ranking-events topic, and return 202 Accepted immediately"
  async_path: "Flink stream aggregation, Redis board update, and Spark hourly reconciliation"
  dead_letter_queue: "ranking-events-dlq with retry limit of 3, alerting when consumer lag exceeds 5 minutes"