Part IV · Search, Selection and Optimisation Pattern 9 3 problems

Top ‘K’ Elements (Heaps)

You almost never need the data sorted. You need its extremes. A heap of size K gives you those for O(n log K) instead of O(n log n), and it works on a stream you cannot hold in memory.

Sorting to find the top three of a million items does a million times more ordering work than the question asked for. A heap is the data structure that answers “what is the smallest thing I am currently keeping” in O(1) and lets you replace it in O(log K). That single operation is the whole pattern.

Contents

  1. When to use
  2. Core idea
  3. Working with heapq
  4. The templates
  5. Common mistakes
  6. Kth Largest Element in an Array
  7. Top K Frequent Elements
  8. K Closest Points to Origin
  9. Recap

When to use

The trigger. The question asks for the K largest, K smallest, K most frequent, or K closest items, or for a running median or k-th value. The give-away extras are “from a stream”, “the data does not fit in memory”, or a very large n with a small k.
SignalWhat it looks like
K language“top K”, “K-th largest”, “K most frequent”, “K nearest”.
Streaming“as numbers arrive”, “design a class with an add method”, “you cannot store everything”.
Repeated extremes“merge K sorted lists”, “the next smallest each time”, “schedule the earliest-finishing task”.
Two-sided balance“median of a data stream” needs two heaps facing each other. Worth knowing as the classic follow-up.

When a heap is not the answer

Core idea

The counter-intuitive part: to keep the K largest items, use a min-heap. The heap holds the K best seen so far, and its root is the weakest of them. That root is the admission threshold. A new item only earns a place if it beats the weakest survivor, and admitting it evicts exactly that root. The heap never grows past K.
stream 9 2 7 8 8 > root? min-heap, K = 3 2 9 7 root = weakest kept = 2 yes, replace 7 9 8 2 evicted, new root = 7 the root is both the answer to “K-th largest” and the admission gate
Figure 9.1 — A min-heap of size K used as a filter. The root doubles as the threshold and, at the end, as the K-th largest value.

The cost comparison worth quoting

ApproachTimeSpaceNotes
Sort everything, take KO(n log n)O(n)Simplest. Fine when n is small or K is near n.
Min-heap of size KO(n log K)O(K)Works on a stream. The default answer.
Heapify all, pop K timesO(n + K log n)O(n)Better when K is tiny and you already hold all the data.
QuickselectO(n) averageO(1)O(n²) worst case. Needs mutable, in-memory data.
Bucket by countO(n)O(n)Only when the key is a bounded integer, such as a frequency.

Working with heapq

Python’s heapq is a min-heap over a plain list. Four facts cover almost everything.

NeedDo this
A max-heapPush -value, negate again on the way out. There is no max-heap in the standard library.
Build from a listheapq.heapify(items) is O(n), cheaper than n pushes at O(n log n).
Pop then push, as one actionheapq.heapreplace(heap, item). One sift instead of two.
Order by a keyPush tuples: (sort_key, tiebreak, payload). Tuples compare left to right.
The tuple-comparison trap. If two tuples tie on the first element, Python compares the second, and if that is an object with no ordering it raises TypeError. Always make the second element something comparable, such as an insertion counter or a coordinate, and keep unorderable payloads in third place or later.
import heapq

heap: list[tuple[int, int, str]] = []
heapq.heappush(heap, (priority, counter, task_name))   # counter breaks ties safely
heapq.nlargest(k, iterable) and nsmallest do exactly this pattern internally, and are the right production choice. In an interview, write the heap loop, then say “in real code I would call heapq.nlargest”. Both halves of that sentence score.

The templates

Template A — keep the K largest
import heapq


def k_largest(nums: list[int], k: int) -> list[int]:
    """The K largest values, unordered. Min-heap: the root is the weakest kept."""
    heap: list[int] = []

    for value in nums:
        if len(heap) < k:
            heapq.heappush(heap, value)
        elif value > heap[0]:              # beats the weakest survivor
            heapq.heapreplace(heap, value)

    return heap

For the K smallest, flip both: use a max-heap by negating, and admit when value < -heap[0].

Template B — K best by a computed key
def k_best(items: list[str], k: int) -> list[str]:
    """K items with the largest score(item)."""
    heap: list[tuple[float, int, str]] = []

    for index, item in enumerate(items):
        entry = (score(item), index, item)   # index is the safe tie-break

        if len(heap) < k:
            heapq.heappush(heap, entry)
        elif entry > heap[0]:
            heapq.heapreplace(heap, entry)

    return [item for _, _, item in heap]

Common mistakes

The problems

1. Kth Largest Element in an Array Medium

Problem

Return the k-th largest element in an unsorted array. Note this is the k-th largest in sorted order, not the k-th distinct value.

Approach

Solution: heap

import heapq


def find_kth_largest(nums: list[int], k: int) -> int:
    """The k-th largest value in nums, counting duplicates.

    Args:
        nums: Integers, unsorted.
        k: 1-based rank from the top, with 1 <= k <= len(nums).

    Returns:
        The k-th largest value.

    Raises:
        ValueError: If k is out of range.

    Example:
        >>> find_kth_largest([3, 2, 1, 5, 6, 4], 2)
        5
    """
    if not 1 <= k <= len(nums):
        raise ValueError(f"k must be in 1..{len(nums)}, got {k}")

    # Seed with the first k values. heapify is O(k), k pushes would be O(k log k).
    heap = nums[:k]
    heapq.heapify(heap)

    for value in nums[k:]:
        if value > heap[0]:
            heapq.heapreplace(heap, value)

    # The root is the weakest of the k largest, which is the k-th largest.
    return heap[0]

Solution: quickselect

import random


def _partition(nums: list[int], left: int, right: int, pivot_index: int) -> int:
    """Lomuto partition. Returns the pivot's final resting index."""
    pivot = nums[pivot_index]
    nums[pivot_index], nums[right] = nums[right], nums[pivot_index]

    store = left
    for i in range(left, right):
        if nums[i] < pivot:
            nums[store], nums[i] = nums[i], nums[store]
            store += 1

    nums[store], nums[right] = nums[right], nums[store]
    return store


def find_kth_largest_quickselect(nums: list[int], k: int) -> int:
    """O(n) expected time. Mutates nums.

    The k-th largest sits at index len(nums) - k once the array is
    partitioned around it, so we only recurse into the half that contains
    that index.
    """
    target = len(nums) - k
    left, right = 0, len(nums) - 1

    while left <= right:
        # A random pivot is what turns the O(n^2) worst case into a
        # vanishingly unlikely one.
        pivot_index = _partition(nums, left, right, random.randint(left, right))

        if pivot_index == target:
            return nums[pivot_index]
        if pivot_index < target:
            left = pivot_index + 1
        else:
            right = pivot_index - 1

    raise ValueError("k out of range")
Why quickselect is O(n) on average. Quicksort recurses into both halves, giving T(n) = 2T(n/2) + n, which is O(n log n). Quickselect recurses into one half: T(n) = T(n/2) + n, and the geometric series n + n/2 + n/4 + … sums to 2n. That one-sentence comparison is the answer the interviewer is fishing for.
HeapO(n log k) time, O(k) spaceQuickselectO(n) average, O(1) space

Which to offer

SituationPick
Data arrives as a stream, or n is huge and k is smallHeap
All data in memory, mutation allowed, k arbitraryQuickselect
You need a guaranteed worst caseHeap. Quickselect degrades to O(n²).
Production Pythonheapq.nlargest(k, nums)[-1]

Edge cases to raise

Say this out loud: “A min-heap, not a max-heap, because I need constant-time access to the weakest item I am keeping so I know what to evict.”

2. Top K Frequent Elements Medium

Problem

Given an array nums and an integer k, return the k most frequent elements, in any order.

Approach

Solution: heap

import heapq
from collections import Counter


def top_k_frequent(nums: list[int], k: int) -> list[int]:
    """The k most frequent values in nums, in no particular order.

    Args:
        nums: Input values.
        k: How many to return, with 1 <= k <= number of distinct values.

    Returns:
        The k values with the highest counts.

    Example:
        >>> sorted(top_k_frequent([1, 1, 1, 2, 2, 3], 2))
        [1, 2]
    """
    counts = Counter(nums)
    heap: list[tuple[int, int]] = []          # (frequency, value)

    for value, frequency in counts.items():
        if len(heap) < k:
            heapq.heappush(heap, (frequency, value))
        elif frequency > heap[0][0]:
            heapq.heapreplace(heap, (frequency, value))

    return [value for _, value in heap]

Solution: bucket by count, O(n)

from collections import Counter


def top_k_frequent_buckets(nums: list[int], k: int) -> list[int]:
    """Same answer in linear time, by indexing buckets on the count itself.

    A value cannot appear more than len(nums) times, so counts are bounded
    and can be used directly as array indices. That removes the log factor.
    """
    counts = Counter(nums)

    # buckets[f] holds every value that occurs exactly f times.
    buckets: list[list[int]] = [[] for _ in range(len(nums) + 1)]
    for value, frequency in counts.items():
        buckets[frequency].append(value)

    out: list[int] = []
    for frequency in range(len(nums), 0, -1):
        for value in buckets[frequency]:
            out.append(value)
            if len(out) == k:
                return out

    return out

Walkthrough of the bucket version

nums = [1, 1, 1, 2, 2, 3], k = 2. Counts are {1: 3, 2: 2, 3: 1}. Buckets: index 1 holds [3], index 2 holds [2], index 3 holds [1]. Walking down from index 6: nothing at 6, 5, 4; at 3 take 1; at 2 take 2; we now have two, so return [1, 2].

HeapO(n + m log k)BucketsO(n)SpaceO(n)

The production one-liner

def top_k_frequent_stdlib(nums: list[int], k: int) -> list[int]:
    """What you would actually ship."""
    return [value for value, _ in Counter(nums).most_common(k)]

most_common uses heapq.nlargest internally when k is given, so it is the heap solution with none of the code. Write your own first, then show this.

Edge cases to raise

Say this out loud: “Counts are bounded by n, so I can use the count itself as an array index and drop the log factor entirely. That is the linear-time version.”

3. K Closest Points to Origin Medium

Problem

Given a list of points on a plane and an integer k, return the k points closest to the origin. The distance is the usual Euclidean distance.

Two observations before any code

Solution

import heapq


def k_closest(points: list[list[int]], k: int) -> list[list[int]]:
    """The k points nearest the origin, in no particular order.

    Args:
        points: [x, y] integer coordinates.
        k: How many to return, with 1 <= k <= len(points).

    Returns:
        The k closest points.

    Example:
        >>> sorted(k_closest([[1, 3], [-2, 2], [5, 8], [0, 1]], 2))
        [[-2, 2], [0, 1]]
    """
    # Negated squared distance turns heapq's min-heap into a max-heap, so the
    # root is the furthest point we are currently keeping. x and y are the
    # tie-break, which keeps every comparison on plain integers.
    heap: list[tuple[int, int, int]] = []

    for x, y in points:
        entry = (-(x * x + y * y), x, y)

        if len(heap) < k:
            heapq.heappush(heap, entry)
        elif entry[0] > heap[0][0]:      # closer than the worst one kept
            heapq.heapreplace(heap, entry)

    return [[x, y] for _, x, y in heap]

Walkthrough

points = [[1,3],[-2,2],[5,8],[0,1]], k = 2. Squared distances are 10, 8, 89, 1.

pointheap root (furthest kept)action
[1, 3]10heap not full, push
[-2, 2]8heap not full, push
[5, 8]891089 > 10, reject
[0, 1]1101 < 10, evict [1, 3]

Result: [[-2, 2], [0, 1]].

TimeO(n log k)SpaceO(k)

The alternatives, ranked

ApproachTimeWhen
Sort by squared distance, slice kO(n log n)Simplest. Fine unless n is very large.
Max-heap of size kO(n log k)Streaming points, or n far bigger than k.
Quickselect on squared distanceO(n) averageAll points in memory and you want the best average time.
heapq.nsmallest(k, points, key=...)O(n log k)Production.

Edge cases to raise

Say this out loud: “I compare squared distances, because square root is monotone and skipping it keeps everything in exact integers. And since I want the K smallest, the size-K heap has to be a max-heap, which in Python means negating.”

Recap

The six things to carry forward

Where this goes next

Pattern 10, Subsets and Backtracking, goes the other way. Instead of shrinking the problem to K items, it enumerates an exponential space on purpose, and the skill is pruning it before it eats the clock.


8 — Depth-First Search 10 — Subsets (Backtracking)