Kth Largest in a Stream: The Persistent Bounded Heap

"Kth Largest Element in a Stream" is a Easy-level staple in Amazon, Meta, Databricks loops, and it is a textbook fit for the Top-K and Heaps pattern — one of the nine patterns that cover the bulk of what FAANG coding interviews actually test. Part 3 of 11 in the Top-K and Heaps arc.

The Problem

Design a class initialized with k and an initial array, whose add(val) method appends a value and returns the k-th largest seen so far (duplicates count). The streaming version of Kth Largest — and the cleanest proof of why the bounded-heap formulation is the right one to internalize.

KthLargest(3, [4, 5, 8, 2])
add(3) -> 4;  add(5) -> 5;  add(10) -> 5;  add(9) -> 8

Recognizing the Top-K and Heaps Pattern

When a problem asks for the k largest, smallest, closest, or most frequent — or for repeated access to an extreme while data changes — a heap gives O(log n) insertion and O(1) access to the extreme. The signature trick: keep a bounded heap of size k with the opposite polarity (a min-heap to track the k largest), evicting the root on overflow. Python's heapq is a min-heap; negate values for max behavior.

A stream forbids re-sorting per query. The size-k min-heap is a persistent summary: it holds exactly the k largest so far, its root is the running answer, and each update costs O(log k). When state must survive across calls, the heap is the class.

The Approach

Constructor: heapify the seed array, pop down to k elements. add: push, pop if over k, return the root. Guaranteed at least k elements by the time add is called per the problem contract — but pushing first keeps even that assumption unnecessary.

The heapq.heappushpop fusion (push then pop in one O(log k) operation) is the polish move when the heap is already full — name it, use it if comfortable.

Python Solution

import heapq


class KthLargest:
    """Streaming k-th largest via a persistent size-k min-heap."""

    def __init__(self, k: int, nums: list[int]) -> None:
        self.k = k
        self.heap = list(nums)
        heapq.heapify(self.heap)
        while len(self.heap) > k:
            heapq.heappop(self.heap)

    def add(self, val: int) -> int:
        heapq.heappush(self.heap, val)
        if len(self.heap) > self.k:
            heapq.heappop(self.heap)
        return self.heap[0]

Complexity

• Time: O(log k) per add — one bounded push and at most one pop
• Space: O(k) — the heap never exceeds k + 1 momentarily

Interview Tips and Follow-Ups

• Explain why a max-heap is wrong here despite 'largest' in the title — you need cheap access to the smallest of the top k. This inversion is the pattern's core idea.
• Constructor cost O(n + (n−k) log n) via heapify-then-pop vs O(n log k) via repeated add — worth one sentence if asked.
• Real-system version: top-k metrics over sliding windows adds eviction by time — say how the heap alone stops sufficing (lazy deletion or a different sketch).

If this clicked, continue the Top-K and Heaps arc in the Technical Interview category. One hundred questions, nine patterns, all in Python.