Top K Frequent Elements in a Stream: Heap and Space-Saving

"Top k frequent elements" is usually asked with a fixed array. In production the input never stops: search queries, error codes, hashtags. Counts keep changing, someone asks for the top ten every few seconds, and the distinct items can outgrow memory. This article covers what an exact streaming answer costs, and how the Space-Saving algorithm gives a bounded-memory answer with guarantees you can test.

The problem it solves

For a fixed array, count with a hash map and then select: the bucket sort version does it in O(n), and the size-k heap does it in O(d log k) for d distinct items. A stream adds two requirements:

• Answer repeatedly. The top k is a query asked over and over while events keep arriving.
• Bound the memory. Ten million distinct search phrases means ten million counters for an exact tally.

The first is cheap. The second cannot be met exactly, so the question becomes which errors are acceptable.

The intuition

Exact, with changing counts. Keep the tally live: each arrival is one dictionary increment, O(1). When asked for the top k, run the size-k min-heap pass over the current counts. Maintaining a top-k heap on every arrival is awkward: heapq has no "increase this key", and an item outside the heap can overtake one inside at any moment. Re-selecting at query time is simpler and correct. The catch is memory: the tally holds every distinct item ever seen.

Approximate, with m counters. Space-Saving (Metwally, Agrawal and El Abbadi, 2005, from memory) keeps at most m counters. A known item increments its own. A new item takes a free counter if there is one. Otherwise it evicts the item with the smallest count and inherits that count plus one, remembering the inherited part as its possible overestimate. Three properties follow, and the code below checks all of them:

• The counters always sum to n, the number of events, so the smallest is at most n / m.
• Every reported count is an upper bound, and subtracting its recorded error gives a lower bound.
• Any item that occurs more than n / m times is guaranteed to hold a counter.

Heavy items are therefore never lost, and their counts are close; it is the long tail that gets blurred. Misra-Gries is the classic relative with similar guarantees, and count-min sketches are another route (from memory).

Watch it run

The animation runs the lesson's exact two-phase version on six events, three distinct labels, with only the top two wanted, so ranking all of them is wasted work. Pass one is a tally: ux is now at 1, then db at 1, ux at 2, api at 1, db at 2, ux at 3. No ordering yet, just counting. Three distinct labels with their counts; now the counts walk through a heap that is never allowed past two.

ux at 3 goes in, since there is still room. The root is the weakest keeper, not the strongest. db at 2 goes in too. api at 1 pushes the heap to three and rises straight to the root, because it is the weakest of them. The root, api at 1, is the weakest, so it is the one dropped, for good. Two held out of three distinct labels: ux and db. The heap never holds more than k; the tally still keeps every distinct label, the memory a long stream cannot afford.

The code

The exact version with a live tally and a query function, then Space-Saving on a skewed stream: 20,000 events over 1,000 labels where a few are hot and the rest form a long tail. The stream is a toy model, a seeded Zipf-like draw, not real traffic.

import heapq
import random
from collections import Counter

def top_k(counts, k):
    """Exact answer from a full tally: a k-sized heap pass, ties broken by label."""
    return [x for x, n in heapq.nsmallest(k, counts.items(), key=lambda kv: (-kv[1], kv[0]))]

counts = Counter()
for event in ["ux", "db", "ux", "api", "db", "ux"]:
    counts[event] += 1                     # O(1) per arrival
print(top_k(counts, 2))                    # ['ux', 'db']
for event in ["api", "api", "api"]:
    counts[event] += 1
print(top_k(counts, 2))                    # ['api', 'ux']  -> re-ask, get the new answer

def space_saving(stream, m):
    """At most m counters, whatever the stream does. item -> [count, overestimate]."""
    counters = {}
    for x in stream:
        if x in counters:
            counters[x][0] += 1
        elif len(counters) < m:
            counters[x] = [1, 0]
        else:                              # full: the newcomer inherits the smallest count
            victim = min(counters, key=lambda y: counters[y][0])
            floor = counters.pop(victim)[0]
            counters[x] = [floor + 1, floor]
    return counters

rng = random.Random(39)
labels = ["tag%d" % i for i in range(1000)]
weights = [1 / (i + 1) for i in range(1000)]          # a few hot labels, a long tail
stream = rng.choices(labels, weights, k=20_000)

exact = Counter(stream)
summary = space_saving(stream, 50)
approx = sorted(summary, key=lambda x: (-summary[x][0], x))[:5]
print(len(exact), len(summary))            # 985 50
print(top_k(exact, 5))                     # ['tag0', 'tag1', 'tag2', 'tag3', 'tag4']
print(approx)                              # ['tag0', 'tag1', 'tag2', 'tag3', 'tag4']
print(summary["tag0"], exact["tag0"])      # [2631, 0] 2631
print(summary["tag9"], exact["tag9"])      # [309, 176] 280

Fifty counters instead of 985, and the same top five. The hot labels never left their counters, so their counts are exact. tag9 shows the blur in the tail: reported as 309 with a possible overestimate of 176, so its true count is somewhere from 133 to 309; it is 280.

The seeded check: 400 random skewed streams with random m, each compared against an exact Counter. The summary never exceeds m counters, every count brackets the truth, the counts sum to the stream length, a summary that never filled up is exact, and every item above n / m is present:

rng = random.Random(7)
ok = True
for _ in range(400):
    universe = ["x%d" % i for i in range(rng.randint(1, 40))]
    stream = rng.choices(universe, [rng.random() ** 3 for _ in universe], k=rng.randint(0, 400))
    m = rng.randint(1, 12)
    truth = Counter(stream)                         # brute force: count everything
    summary = space_saving(stream, m)
    ok &= len(summary) <= m
    for x, (count, over) in summary.items():
        ok &= count - over <= truth[x] <= count     # never under, off by at most its error
    ok &= sum(c for c, _ in summary.values()) == len(stream)   # so the smallest is at most n / m
    if len(summary) < m:                            # never full: nothing was evicted, all exact
        ok &= all(summary[x] == [truth[x], 0] for x in truth)
    ok &= all(x in summary for x in truth if truth[x] * m > len(stream))   # heavy hitters kept
print(ok)                                           # True

The complexity

• Exact tally: O(1) per event, O(d) memory for d distinct items, O(d log k) per query.
• Space-Saving as written: O(m) memory always. A new item that evicts scans all m counters, O(m); the paper's linked "stream summary" structure, or a heap, brings that down to O(1) or O(log m) (from memory).
• Choosing m: items above n / m are guaranteed, so m is set by the smallest frequency share you must not miss, not by k.

Where it goes wrong

• Keeping the full tally forever. Fine for a few thousand labels; a memory leak for open-ended keys like URLs or queries.
• Trusting tail counts from a summary. Report the bracket, count - error to count, or only report items whose lower bound clears the threshold.
• Updating a heap in place. Changing a count inside a heapq list breaks the heap invariant silently.
• Ignoring time. "Top this hour" needs windows or decay; an all-time tally is dominated by old events.

When it shows up in interviews

As a follow-up to the array version ("now the data is a stream, and memory is limited"), and in system design as "trending hashtags", "top searched queries" or "most frequent errors". The patterns cheat sheet files it under top-k. For exact leaderboards with updates, a sorted structure is the usual answer, as in designing a leaderboard.

How to say it in an interview

"If the distinct set fits in memory, I keep an exact counter, O(1) per event, and answer each query with a size-k min-heap over the counts, O(d log k). If it doesn't, I use Space-Saving with m counters: a new item evicts the smallest counter and inherits its count plus one, recording that as its possible error. Counts are upper bounds with a known error, they always sum to n, and anything more frequent than n over m is guaranteed to be tracked, so heavy hitters are never lost."