Problem 94 · Data Engineering & Streaming

Top-K Heavy Hitters (Misra-Gries)

MediumClassic DSA & systems

01Problem

In one pass with O(k) memory, find all items occurring more than n/k times in a stream, returning candidate counts and the top candidates.

02Key insight

Keep k-1 counters; a new item when full decrements all counters (dropping zeros). Each decrement 'cancels' k distinct items, so any item with frequency > n/k survives and its count is underestimated by at most n/k. Decrement-all is amortized O(1) because total decrements <= total increments.

03Solution

solution.pypython31 lines

from typing import Hashable, Iterable


class MisraGries:
    def __init__(self, k: int) -> None:
        if k < 2:
            raise ValueError("k must be >= 2")
        self.k = k
        self.n = 0
        self.counters: dict[Hashable, int] = {}

    def add(self, item: Hashable) -> None:
        self.n += 1
        if item in self.counters:
            self.counters[item] += 1
        elif len(self.counters) < self.k - 1:
            self.counters[item] = 1
        else:
            for key in list(self.counters):  # cancel k distinct items at once
                self.counters[key] -= 1
                if self.counters[key] == 0:
                    del self.counters[key]

    def extend(self, items: Iterable[Hashable]) -> "MisraGries":
        for item in items:
            self.add(item)
        return self

    def top(self, m: int) -> list[tuple[Hashable, int]]:
        """Candidates by lower-bound count; true count is in [c, c + n/k]."""
        return sorted(self.counters.items(), key=lambda kv: -kv[1])[:m]

Try it first.

TimeO(1) amortized per item

SpaceO(k)

tests.py · 3 passing

tests.pypython10 lines

import random
from collections import Counter
rng = random.Random(3)
stream = ["a"] * 4000 + ["b"] * 2500 + [f"noise{rng.randrange(3000)}" for _ in range(3500)]
rng.shuffle(stream)
mg = MisraGries(k=10).extend(stream)
truth = Counter(stream)
assert [item for item, _ in mg.top(2)] == ["a", "b"]
assert all(0 <= truth[i] - c <= len(stream) / 10 for i, c in mg.counters.items())
assert len(mg.counters) <= 9

04Staff-level follow-up

How do you get exact counts and merge across nodes? Misra-Gries summaries merge by adding counters then subtracting the (k)-th largest value; for exact top-K do a second pass counting only the candidates, or use Space-Saving which also bounds overestimation per item.