Ch. 18 · Data Structures & Algorithms

Heaps for Top-K and Incremental Selection

Keep the k largest items with a size-k min-heap: a traced stream, timings against sorting, heapq tie-breaking and K Closest Points.

~7 min readintermediateupdated Oct 11, 2026

“Return the k largest”, “the k most frequent”, “the k closest”: these come up constantly, in interviews and in real systems (top error messages in a log, nearest restaurants, highest-scoring results). The obvious answer is to sort everything and take the first k. That’s O(n log n), and it needs all the data in memory at once. A heap does better on both counts, and the way it does it is slightly counter-intuitive the first time you see it.

A quick refresher on what a heap gives you

A binary heap is a complete binary tree stored in an array. For the node at index i, its children are at 2i + 1 and 2i + 2, and in a min-heap every parent is no larger than its children. That one rule means the smallest element is always at index 0.

The operations you’ll use:

  • peek at the root: O(1)
  • push a new item (append, then sift up): O(log n)
  • pop the root (move the last item to the top, then sift down): O(log n)
  • heapify an existing list in place: O(n)

A heap does not keep everything sorted. Only the root is guaranteed. That’s exactly why it’s cheaper than a sorted structure, and it’s all top-k needs.

Why the k largest live in a min-heap

Here’s the counter-intuitive part. To keep the largest k values, you use a min-heap, capped at size k. The heap holds your current best k candidates. Its root is the smallest of them: the weakest survivor, the one that should be thrown out first if something better arrives. For each new value, compare it with the root. If it’s bigger, it deserves a place, and the root loses its spot. If not, ignore it.

import heapq

def top_k(stream, k):
    heap = []                                # min-heap of the k largest seen so far
    for x in stream:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:                    # beats the weakest survivor
            heapq.heapreplace(heap, x)       # pop root, push x, one sift
    return sorted(heap, reverse=True)
python

Tracing it on the stream [5, 1, 9, 3, 7, 2, 8] with k = 3, printing the heap’s array after each read:

read 5: heap=[5] root=5
read 1: heap=[1, 5] root=1
read 9: heap=[1, 5, 9] root=1
read 3: heap=[3, 5, 9] root=3
read 7: heap=[5, 7, 9] root=5
read 2: heap=[5, 7, 9] root=5
read 8: heap=[7, 8, 9] root=7
[9, 8, 7]
Text

The first three values fill the heap. When 3 arrives, it beats the root 1, so 1 is evicted. 7 beats 3. 2 loses to the root 5 and is dropped with a single comparison, no heap operation at all. 8 beats 5. At the end the heap holds 7, 8, 9, and the root, 7, is the 3rd largest value, which answers Kth Largest Element in an Array directly.

If you used a max-heap instead, the root would be the best value, which is the one you never want to evict, and you’d have no cheap way to find the worst. That’s the mistake to avoid: picking the heap type by matching the word in the question (“largest” means max-heap) rather than by asking which element you need to remove.

heapreplace pops the root and pushes the new value with one sift-down, slightly cheaper than a separate pop and push. Its cousin heappushpop pushes first and then pops, which hands the new value straight back if it’s no larger than the root. Either works here because we only call it after checking x > heap[0].

What the heap saves

Each of the n values costs at most one O(log k) heap operation, so the total is O(n log k) time and O(k) space. When k is small compared to n, log k is tiny, and the gap with sorting is real. On a million random floats with k = 10:

1,000,000 floats, k=10: sorted 155 ms, manual heap 25 ms, heapq.nlargest 9 ms
Text

Those numbers are from one run on an M1 MacBook Pro with Python 3.14 and vary a little between runs; the order of magnitude doesn’t. The hand-written heap is six times faster than sorting, and the library’s heapq.nlargest, which implements the same idea with less interpreter overhead, is faster again. In production Python, use nlargest. In an interview, write the loop so the interviewer can see the logic, then mention that the library function exists.

The memory argument often matters more than speed. A sort needs the whole input materialised. The bounded heap reads one value at a time and holds only k, so it can find the top 10 response times in a 50 GB log file streamed line by line.

When k approaches n, the advantage disappears: O(n log k) becomes O(n log n). For a single k-th element there’s also quickselect, which averages O(n) but has an O(n²) worst case and rearranges the input, a trade-off worth naming if asked for the fastest possible answer.

Heap entries, keys and ties in Python

heapq only knows how to build a min-heap, and it compares entries with <. Two consequences come up in nearly every heap problem.

First, for max-heap behaviour, negate the key. Pushing -distance makes the largest distance sit at the root.

Second, when you push tuples like (priority, item), Python compares element by element. If two priorities tie, it compares the items, and many objects can’t be compared:

tasks = [(2, {"name": "deploy"}), (2, {"name": "notify"}), (1, {"name": "build"})]
h = []
for t in tasks:
    heapq.heappush(h, t)
python
TypeError: '<' not supported between instances of 'dict' and 'dict'
Text

The fix is a tie-breaker that’s always comparable and unique, usually an increasing counter:

h = []
for seq, (prio, task) in enumerate(tasks):
    heapq.heappush(h, (prio, seq, task))     # seq breaks ties, so dicts are never compared
print([heapq.heappop(h)[2]["name"] for _ in range(len(h))])
python
['build', 'deploy', 'notify']
Text

The counter also makes equal-priority items come out in insertion order, which is usually what you want for a task queue. JavaScript has no built-in heap at all, so in a JS interview you’d either write a small array-backed one (push with sift-up, pop with sift-down, about 30 lines) or ask whether you may assume a priority queue class exists. Java has PriorityQueue, which takes a comparator.

Top K Frequent Elements

Frequency questions combine a hash map with a heap. Count first, then keep the k values with the highest counts in a size-k min-heap keyed by count:

from collections import Counter

def top_k_frequent(nums, k):
    counts = Counter(nums)
    heap = []                                   # (count, value), smallest count at the root
    for value, count in counts.items():
        heapq.heappush(heap, (count, value))
        if len(heap) > k:
            heapq.heappop(heap)                 # drop the least frequent
    return [value for count, value in sorted(heap, reverse=True)]

print(top_k_frequent([1, 1, 1, 2, 2, 3], 2), top_k_frequent([4, 4, 5, 5, 5, 6, 6, 6, 6], 1))
python
[1, 2] [6]
Text

This version pushes first and pops when the heap overflows, which is simpler to write and costs the same O(log k) per item. Counting is O(n), and the heap phase is O(u log k) for u distinct values. A bucket approach (an array indexed by count, from 1 to n) brings it to O(n) overall, a good follow-up for Top K Frequent Elements.

Interview practice: K Closest Points to Origin

Given a list of points [x, y] and an integer k, return the k points closest to the origin, in any order. Aim for better than sorting all points. Explain which heap you use and why.

Answer

You want to keep the k smallest distances, so by the same logic as before, use a max-heap of size k: its root is the farthest point you’re currently keeping, the first to evict when a closer one arrives. With heapq, that means pushing negated distances.

Compare squared distances, x² + y². Squaring preserves order for non-negative numbers, so the result is identical and you avoid floating-point square roots entirely.

def k_closest(points, k):
    heap = []                                   # max-heap by distance, via negation
    for x, y in points:
        d = x * x + y * y                       # squared distance: same order, no sqrt
        if len(heap) < k:
            heapq.heappush(heap, (-d, x, y))
        elif d < -heap[0][0]:                   # closer than the farthest one kept
            heapq.heapreplace(heap, (-d, x, y))
    return [[x, y] for _, x, y in heap]

print(k_closest([[1, 3], [-2, 2]], 1))
print(k_closest([[3, 3], [5, -1], [-2, 4]], 2))
print(k_closest([[0, 1], [1, 0], [2, 2], [-1, 0]], 3))
python
[[-2, 2]]
[[-2, 4], [3, 3]]
[[-1, 0], [1, 0], [0, 1]]
Text

In the second case, [3, 3] and [-2, 4] have squared distances 18 and 20, and [5, -1] has 26, so it’s the one left out. Time is O(n log k) and extra space O(k). The entries are tuples of numbers, so ties in distance fall through to comparing coordinates, which are always comparable; no counter needed here. If an interviewer asks about a stream of points arriving one at a time, this solution already handles it, since it never looks at a point twice. Practise on K Closest Points to Origin, and once you’re comfortable, Find Median from Data Stream uses two heaps at once.

See also

More in Data Structures & Algorithms

read ✓Data Structures & Algorithms · mid

Bit Manipulation Patterns for Interviews

Test, set and clear bits, count them with n & (n - 1), cancel pairs with XOR, enumerate subsets as masks, and avoid width bugs in Python and JS.

~8 min readread →
read ✓Data Structures & Algorithms · hard

Backtracking and Safe Pruning Rules

Backtracking as choose, explore, unchoose, and the proof each prune needs: sorted breaks, duplicate skips and the negative-number trap.

~7 min readread →
read ✓Data Structures & Algorithms · easy

Big-O Complexity and Input Growth

Read Big-O as a growth rate: count steps as n doubles, name separate inputs n and m, and spot the hidden O(n) inside one-line operations.

~6 min readread →
esc