Reason About Top K Frequent Values at Scale

Problem Statement

When monitoring trading systems or high-throughput API gateways, identifying the top k most frequent identifiers (such as stock symbols, IP addresses, or user IDs) is a critical operation.

You need to implement a class TopKTracker that processes a stream of string tokens on a single machine using bounded memory where possible, and reason through its distributed aggregation.

Requirements

1. Single-Node Execution (top_k_frequent): Given an iterable of strings stream and an integer k, compute the k most frequent strings along with their counts.

   • Count token frequencies using a hash map.
   • Extract the top k elements using a min-heap of size at most k rather than sorting the entire vocabulary.
   • Return a list of tuples (token, frequency) ordered by frequency descending.
   • If multiple tokens have identical frequencies, their relative order does not matter unless specified, but all top k slots must contain strictly the highest frequency items.

2. Scaling Logic (Conceptual / Implementation): Explain and implement the aggregation function merge_partitioned_counts(partition_counts: list[dict[str, int]], k: int) -> list[tuple[str, int]] where multiple workers process disjoint key partitions versus arbitrary stream splits.

Function Signature

from typing import Iterable, List, Tuple, Dict

class TopKTracker:
    def top_k_frequent(self, stream: Iterable[str], k: int) -> List[Tuple[str, int]]:
        pass

    def merge_partitioned_counts(
        self, 
        partition_counts: List[Dict[str, int]], 
        k: int
    ) -> List[Tuple[str, int]]:
        pass

Examples

tracker = TopKTracker()
stream = ["AAPL", "MSFT", "AAPL", "GOOG", "AAPL", "MSFT", "AMZN"]
k = 2

# AAPL count: 3, MSFT count: 2, GOOG count: 1, AMZN count: 1
result = tracker.top_k_frequent(stream, k)
# Output: [("AAPL", 3), ("MSFT", 2)]

# Worker 1 saw a random chunk of logs, Worker 2 saw another chunk
p1 = {"AAPL": 2, "MSFT": 1}
p2 = {"AAPL": 1, "GOOG": 3, "MSFT": 2}

result = tracker.merge_partitioned_counts([p1, p2], k=2)
# Merged counts: AAPL: 3, MSFT: 3, GOOG: 3
# Any 2 of AAPL, MSFT, GOOG with count 3 is valid

Constraints

• 1 <= k <= 10^4
• Total stream tokens N <= 10^7
• Unique tokens U <= 10^6
• Memory overhead for the heap selection step must be O(k) space.