Table of Contents

1. Introduction and Use Cases
2. Functional and Non-Functional Requirements
3. Capacity Estimation
4. Streaming Algorithms Deep Dive
5. Distributed Counting Architecture
6. Merge Strategies
7. High-Level Architecture
8. Read/Write Paths
9. Time-Decay and Freshness
10. Monitoring and Accuracy
11. Interview Questions and Answers

1. Introduction and Use Cases

The Top-K problem is one of the most fundamental problems in distributed systems and data engineering. Given a stream of events (billions per day), find the K items with the highest frequency, score, or weight. This appears everywhere in production systems: trending topics on social media, most-searched queries, highest-value customers, most-viewed videos, fraud detection, and network monitoring.

The challenge is that the data volume is too large to fit in memory and too fast to count with traditional database operations. Naive approaches — sorting all items or maintaining exact counts in a hash map — require O(N) memory where N is the number of unique items, which can be billions. Streaming algorithms solve this by making trades between accuracy and memory, providing approximate answers with provable error bounds.

Real-World Use Cases

Key Metrics

2. Functional and Non-Functional Requirements

Functional Requirements

1. Count occurrences: Track the frequency of each item in a high-volume event stream.
2. Top-K query: Return the K most frequent items with their counts, sorted by frequency.
3. Time-windowed ranking: Support ranking over configurable time windows (last hour, last day, last week).
4. Sub-topics: Support Top-K within categories or partitions (e.g., trending per country).
5. Item metadata: Return associated metadata (name, description, thumbnail) with ranked items.

Non-Functional Requirements

3. Capacity Estimation

public class TopKCapacityEstimator
{
    public static void Estimate()
    {
        long eventsPerDay = 100_000_000_000L; // 100B events/day
        long uniqueItems = 1_000_000_000L;    // 1B unique items
        int k = 1000;                          // Top 1000

        double eventsPerSecond = eventsPerDay / 86400.0;
        Console.WriteLine($"Events/sec: {eventsPerSecond:N0}");

        // Exact counting (HashMap): 1B items * 8 bytes = 8 GB
        long exactMemoryBytes = uniqueItems * 8;
        Console.WriteLine($"Exact counting memory: {exactMemoryBytes / (1024*1024*1024.0):F1} GB");

        // Count-Min Sketch: width=2^20, depth=8 = 8M counters * 8 bytes = 64 MB
        long cmsMemoryBytes = (long)Math.Pow(2, 20) * 8 * 8;
        Console.WriteLine($"Count-Min Sketch memory: {cmsMemoryBytes / (1024*1024.0):F0} MB");

        // Min-Heap for Top-K: K items * (8 bytes key + 8 bytes count + 16 bytes ptr) = 32 KB
        long heapMemoryBytes = k * 32;
        Console.WriteLine($"Min-Heap memory: {heapMemoryBytes} bytes");

        // Combined: CMS + Min-Heap = 64 MB + 32 KB ≈ 64 MB
        Console.WriteLine($"Combined memory: {(cmsMemoryBytes + heapMemoryBytes) / (1024*1024.0):F0} MB");
        Console.WriteLine($"Memory savings vs exact: {(1.0 - (cmsMemoryBytes + heapMemoryBytes) / (double)exactMemoryBytes) * 100:F1}%");
    }
}

4. Streaming Algorithms Deep Dive

Count-Min Sketch

The Count-Min Sketch (CMS) is a probabilistic data structure that estimates item frequencies using a 2D array of counters with multiple hash functions. It guarantees that estimates are never less than the true count and over-estimate by at most εN with probability δ.

public class CountMinSketch
{
    private readonly int _depth;    // Number of hash functions (rows)
    private readonly int _width;    // Number of counters per row
    private readonly long[][] _table;
    private readonly long _totalEvents;

    public CountMinSketch(int width = 1 << 20, int depth = 8)
    {
        _width = width;
        _depth = depth;
        _table = new long[depth][];
        for (int i = 0; i < depth; i++)
            _table[i] = new long[width];
    }

    public void Update(string item, long count = 1)
    {
        for (int i = 0; i < _depth; i++)
        {
            var index = GetHash(i, item) % _width;
            Interlocked.Add(ref _table[i][index], count);
        }
        Interlocked.Increment(ref _totalEvents);
    }

    public long Estimate(string item)
    {
        long min = long.MaxValue;
        for (int i = 0; i < _depth; i++)
        {
            var index = GetHash(i, item) % _width;
            min = Math.Min(min, Volatile.Read(ref _table[i][index]));
        }
        return min;
    }

    // Error bound: with probability ≥ 1 - δ, estimate ≤ true_count + ε * N
    // where N = total events, ε = e/width, δ = e^(-depth)
    public double GetErrorBound()
    {
        double epsilon = Math.E / _width;
        return epsilon * _totalEvents;
    }

    // Merge two sketches (for distributed counting)
    public static CountMinSketch Merge(CountMinSketch a, CountMinSketch b)
    {
        var merged = new CountMinSketch(a._width, a._depth);
        for (int i = 0; i < a._depth; i++)
        {
            for (int j = 0; j < a._width; j++)
            {
                merged._table[i][j] = a._table[i][j] + b._table[i][j];
            }
        }
        return merged;
    }

    private int GetHash(int hashIndex, string item)
    {
        // Use different hash seeds for each row
        return MurmurHash3.Hash(Encoding.UTF8.GetBytes(item), seed: hashIndex * 31);
    }
}

Space-Saving Algorithm

The Space-Saving algorithm maintains exactly K counters and achieves near-optimal accuracy by merging the least frequent item when a new item arrives and no slot is available.

public class SpaceSavingCounter
{
    private readonly int _k;
    private readonly SortedList<long, LinkedList<string>> _countBuckets;
    private readonly Dictionary<string, (long count, long bucketKey)> _items;

    public SpaceSavingCounter(int k = 1000)
    {
        _k = k;
        _countBuckets = new SortedList<long, LinkedList<string>>();
        _items = new Dictionary<string, (long, long)>();
    }

    public void Update(string item, long increment = 1)
    {
        if (_items.ContainsKey(item))
        {
            // Item already tracked — increment its count
            var (count, bucketKey) = _items[item];
            MoveItem(item, count, count + increment);
        }
        else if (_items.Count < _k)
        {
            // Space available — add new item with count = increment
            _items[item] = (increment, increment);
            AddToBucket(item, increment);
        }
        else
        {
            // No space — find and replace the minimum item
            var minBucket = _countBuckets.Keys[0];
            var minItem = _countBuckets[minBucket].First.Value;
            var minCount = minBucket;

            // Remove minimum item
            _countBuckets[minBucket].RemoveFirst();
            if (_countBuckets[minBucket].Count == 0)
                _countBuckets.Remove(minBucket);
            _items.Remove(minItem);

            // Add new item with count = min_count + increment
            // This is the key insight: we "inherit" the minimum count
            // This gives us an error bound of ±ε where ε is the minimum count
            var newCount = minCount + increment;
            _items[item] = (newCount, newCount);
            AddToBucket(item, newCount);
        }
    }

    public List<(string item, long count)> GetTopK()
    {
        return _countBuckets.Reverse()
            .SelectMany(kv => kv.Value.Select(item => (item, kv.Key)))
            .Take(_k)
            .ToList();
    }

    public long GetCount(string item)
    {
        return _items.TryGetValue(item, out var info) ? info.count : 0;
    }

    private void MoveItem(string item, long oldCount, long newCount)
    {
        // Remove from old bucket
        var oldBucket = _countBuckets[oldCount];
        oldBucket.Remove(item);
        if (oldBucket.Count == 0)
            _countBuckets.Remove(oldCount);

        // Add to new bucket
        _items[item] = (newCount, newCount);
        AddToBucket(item, newCount);
    }

    private void AddToBucket(string item, long count)
    {
        if (!_countBuckets.ContainsKey(count))
            _countBuckets[count] = new LinkedList<string>();
        _countBuckets[count].AddLast(item);
    }
}

Algorithm Comparison

5. Distributed Counting Architecture

Sharded Counting

At 1M+ events/sec, a single machine cannot handle the processing load. We shard the event stream by item hash, distributing counts across multiple machines. Each shard maintains its own Count-Min Sketch and Min-Heap for its portion of the item space.

public class ShardedTopKService
{
    private readonly List<CounterShard> _shards;
    private readonly int _shardCount;
    private readonly ICacheService _cache;

    public ShardedTopKService(int shardCount = 128, int k = 1000)
    {
        _shardCount = shardCount;
        _shards = Enumerable.Range(0, shardCount)
            .Select(i => new CounterShard(i, k))
            .ToList();
    }

    public void ProcessEvent(string item, long count = 1)
    {
        var shardIndex = GetShardIndex(item);
        _shards[shardIndex].Increment(item, count);
    }

    public async Task<List<(string item, long count)>> GetTopKAsync(int k = 1000)
    {
        var cacheKey = $"topk:{k}";
        var cached = await _cache.GetAsync<List<(string, long)>>(cacheKey);
        if (cached != null) return cached;

        // Collect local top-K from each shard
        var shardResults = _shards.Select(shard => shard.GetTopK(k * 2)).ToList();

        // Merge and re-rank
        var merged = shardResults
            .SelectMany(r => r)
            .GroupBy(r => r.item)
            .Select(g => (item: g.Key, count: g.Sum(r => r.count)))
            .OrderByDescending(r => r.count)
            .Take(k)
            .ToList();

        await _cache.SetAsync(cacheKey, merged, TimeSpan.FromSeconds(30));
        return merged;
    }

    private int GetShardIndex(string item)
    {
        return Math.Abs(item.GetHashCode()) % _shardCount;
    }
}

public class CounterShard
{
    private readonly int _shardId;
    private readonly CountMinSketch _cms;
    private readonly SpaceSavingCounter _heap;

    public CounterShard(int shardId, int k)
    {
        _shardId = shardId;
        _cms = new CountMinSketch(width: 1 << 18, depth: 6); // ~4MB per shard
        _heap = new SpaceSavingCounter(k);
    }

    public void Increment(string item, long count = 1)
    {
        _cms.Update(item, count);
        _heap.Update(item, count);
    }

    public List<(string item, long count)> GetTopK(int k)
    {
        // Use the heap for fast Top-K
        return _heap.GetTopK().Take(k).ToList();
    }

    public long EstimateCount(string item)
    {
        // Use CMS for accurate individual counts
        return _cms.Estimate(item);
    }
}

6. Merge Strategies

Multi-Level Merge

public class TopKMerger
{
    // Merge top-K results from multiple shards
    public List<(string item, long count)> MergeTopK(
        List<List<(string item, long count)>> shardResults, int k)
    {
        // Strategy 1: Simple frequency merge
        var merged = new Dictionary<string, long>();
        foreach (var shardResult in shardResults)
        {
            foreach (var (item, count) in shardResult)
            {
                merged.TryGetValue(item, out var existing);
                merged[item] = existing + count;
            }
        }

        return merged
            .OrderByDescending(kv => kv.Value)
            .Take(k)
            .Select(kv => (kv.Key, kv.Value))
            .ToList();
    }

    // Merge with error bounds from Count-Min Sketch
    public List<RankedItem> MergeWithConfidence(
        List<ShardTopKResult> shardResults, int k)
    {
        var itemStats = new Dictionary<string, ItemStats>();

        foreach (var result in shardResults)
        {
            foreach (var item in result.Items)
            {
                if (!itemStats.ContainsKey(item.Item))
                    itemStats[item.Item] = new ItemStats();

                var stats = itemStats[item.Item];
                stats.TotalCount += item.EstimatedCount;
                stats.ErrorBound += item.ErrorBound;
                stats.ShardCount++;
            }
        }

        return itemStats
            .Select(kv => new RankedItem
            {
                Item = kv.Key,
                EstimatedCount = kv.Value.TotalCount,
                LowerBound = kv.Value.TotalCount - kv.Value.ErrorBound,
                UpperBound = kv.Value.TotalCount + kv.Value.ErrorBound,
                Confidence = CalculateConfidence(kv.Value)
            })
            .OrderByDescending(r => r.EstimatedCount)
            .Take(k)
            .ToList();
    }
}

public class ItemStats
{
    public long TotalCount { get; set; }
    public long ErrorBound { get; set; }
    public int ShardCount { get; set; }
    public double Confidence => ShardCount / 128.0; // Higher if seen in more shards
}

Periodic Reconciliation

7. High-Level Architecture

8. Read/Write Paths

Write Path

Read Path

public class TopKQueryHandler
{
    private readonly ShardedTopKService _topKService;
    private readonly ICacheService _cache;
    private readonly IItemMetadataStore _metadata;

    public async Task<TopKResponse> GetTopKAsync(TopKRequest request)
    {
        var cacheKey = $"topk:{request.Category}:{request.Window}:{request.K}";
        var cached = await _cache.GetAsync<TopKResponse>(cacheKey);
        if (cached != null) return cached;

        var topItems = await _topKService.GetTopKAsync(request.K);

        // Enrich with metadata
        var enrichedItems = new List<RankedItem>();
        foreach (var (item, count) in topItems)
        {
            var metadata = await _metadata.GetAsync(item);
            enrichedItems.Add(new RankedItem
            {
                Item = item,
                Count = count,
                Metadata = metadata,
                Rank = enrichedItems.Count + 1
            });
        }

        var response = new TopKResponse
        {
            Items = enrichedItems,
            GeneratedAt = DateTimeOffset.UtcNow,
            Window = request.Window,
            TotalItemsCounted = await _topKService.GetTotalItemCountAsync()
        };

        await _cache.SetAsync(cacheKey, response, TimeSpan.FromSeconds(30));
        return response;
    }
}

9. Time-Decay and Freshness

Top-K ranking must account for recency. A tweet from 5 minutes ago is more relevant than one from 5 hours ago. We implement exponential time-decay to give more weight to recent events.

public class TimeDecayCounter
{
    private readonly double _halfLifeHours;
    private readonly SpaceSavingCounter _counter;

    public TimeDecayCounter(int k = 1000, double halfLifeHours = 24)
    {
        _counter = new SpaceSavingCounter(k);
        _halfLifeHours = halfLifeHours;
    }

    public void Update(string item)
    {
        // Each event contributes a weight of 1.0
        // But we adjust for time decay during query time
        _counter.Update(item, 1);
    }

    public List<(string item, double decayedCount)> GetDecayedTopK(int k)
    {
        var now = DateTimeOffset.UtcNow;
        var topK = _counter.GetTopK();

        // Apply time decay based on when each item was last seen
        return topK.Select(tuple =>
        {
            var (item, rawCount) = tuple;
            var lastSeen = GetLastSeenTime(item);
            var hoursSinceLastSeen = (now - lastSeen).TotalHours;
            var decayFactor = Math.Pow(0.5, hoursSinceLastSeen / _halfLifeHours);

            return (item, rawCount * decayFactor);
        })
        .OrderByDescending(tuple => tuple.Item2)
        .Take(k)
        .ToList();
    }
}

// Exponential decay with configurable half-life
public class ExponentialDecayFunction
{
    private readonly double _lambda;

    public ExponentialDecayFunction(double halfLifeHours)
    {
        _lambda = Math.Log(2) / halfLifeHours;
    }

    public double ComputeWeight(DateTimeOffset eventTime, DateTimeOffset queryTime)
    {
        var hoursElapsed = (queryTime - eventTime).TotalHours;
        return Math.Exp(-_lambda * hoursElapsed);
    }
}

Decay Strategies

10. Monitoring and Accuracy

Accuracy Measurement

public class TopKAccuracyMonitor
{
    private readonly IExactCounter _exactCounter; // For sampling-based verification
    private readonly ITopKService _topKService;

    public async Task<AccuracyReport> MeasureAccuracyAsync(string category, int sampleSize = 10000)
    {
        // Sample random items and compare estimated vs exact counts
        var sampleItems = await GetRandomItemsAsync(category, sampleSize);
        long totalError = 0;
        long totalExact = 0;
        int overEstimates = 0;
        int underEstimates = 0;

        foreach (var item in sampleItems)
        {
            var estimated = await _topKService.EstimateCountAsync(item);
            var exact = await _exactCounter.GetExactCountAsync(item);

            totalError += Math.Abs(estimated - exact);
            totalExact += exact;

            if (estimated > exact) overEstimates++;
            else if (estimated < exact) underEstimates++;
        }

        var avgError = (double)totalError / sampleSize;
        var avgExact = (double)totalExact / sampleSize;

        return new AccuracyReport
        {
            MeanAbsoluteError = avgError,
            RelativeError = avgExact > 0 ? avgError / avgExact : 0,
            OverEstimateRate = (double)overEstimates / sampleSize,
            UnderEstimateRate = (double)underEstimates / sampleSize,
            SampleSize = sampleSize,
            MeasuredAt = DateTimeOffset.UtcNow
        };
    }
}

Monitoring Dashboard

11. Interview Questions and Answers

Q1: How do you handle items that appear and disappear from the Top-K?

The Space-Saving algorithm naturally handles this. When a new item enters the Top-K, it displaces the current minimum. When an item stops appearing, its count remains static while other items surpass it, pushing it out of the Top-K on the next query. The key insight is that we never explicitly remove items — they naturally fall out of the ranking as their relative count decreases.

Q2: How would you handle multiple time windows (1 hour, 1 day, 1 week)?

Maintain separate Count-Min Sketch and Min-Heap instances for each time window. Each window has its own Kafka consumer group that processes events independently. Alternatively, use a single CMS with time-decayed weights: events contribute weight that decays exponentially, and the window parameter controls the half-life. This is more memory-efficient but less accurate for precise window boundaries.

Q3: How do you handle skewed data (e.g., a viral tweet with 100x normal volume)?

Implement rate limiting at the shard level: cap per-item increments to prevent a single hot item from dominating the CMS. Use exponential decay so viral items' influence decreases over time. For extreme skew, implement a "hot items" bypass path that tracks viral items separately with exact counts (they're few enough to count exactly).

Q4: How would you merge Top-K results from 128 shards accurately?

Each shard returns its local Top-2K (2x the requested K to account for items that appear in multiple shards). The merge service groups by item ID, sums counts across shards, and re-ranks. For items that appear in the CMS of multiple shards, sum the estimated counts. The error bound of the merged result is the sum of individual shard error bounds. This gives a globally consistent Top-K with the same error guarantees.

Q5: How do you handle late-arriving events that change the Top-K?

Implement a watermark-based approach in Flink. Allow a configurable lateness window (e.g., 5 minutes). Late events within this window are processed normally and update the CMS. After the watermark passes, late events are logged but not processed. The dashboard shows a "finalizing" indicator until the watermark passes. For critical rankings (e.g., billing), use a batch reconciliation job that recomputes exact counts daily.

Q6: How would you implement Top-K for a category hierarchy (e.g., trending per country per category)?

Use a hierarchical sharding strategy: primary shard by country, secondary shard by category. Each (country, category) combination maintains its own CMS and Min-Heap. The merge service can answer queries at any level: global Top-K (merge all), per-country Top-K (filter by country), or per-category Top-K (filter by category). Memory scales with the product of countries × categories, which is manageable (200 countries × 100 categories = 20K combinations).

Q7: How do you ensure the Top-K is correct within the stated error bounds?

Statistical guarantees: Count-Min Sketch provides a probabilistic upper bound on count estimates with probability 1-δ. Space-Saving provides an error bound equal to the minimum count in the heap. For production systems, run regular accuracy audits: sample random items, compare CMS estimates against exact counts (from a small representative sample stored in a database), and report the empirical error rate. Tune CMS parameters (width, depth) based on observed error rates.

Q8: How would you design this for a global social media platform with 500M daily active users?

Scale each layer: (1) Ingestion: 500 Kafka partitions to handle 2M events/sec. (2) Processing: 256 Flink task managers, each handling a subset of partitions. (3) Counting: 256 counter shards (4 CMS + Heap per machine). (4) Merge: Hierarchical merge with a coordinator per region. (5) Serving: Redis cluster with 1M+ QPS capacity. (6) Cross-region: Merge results from US, EU, APAC with 5-minute delay. Total memory: ~16 GB for CMS across all shards, well within a single rack of machines.