Kth Largest Element in a Stream

Solving tips

Key insight: the k-th largest is the minimum of the k largest, so keep a MIN-heap capped at size k — its root is always the answer.
On add, push then pop if size exceeds k; anything evicted from the top-k can never return, so the cap is safe.
Handle initial nums with fewer than k elements, and remember duplicates each take a slot ('k-th largest', not k-th distinct).
Target O(log k) per add and O(k) space; heapq.heappushpop or skipping values <= root avoids extra sifts.

Problem

Build a class that tracks the k-th largest value in a growing stream of numbers.

KthLargest(k: int, nums: List[int]) — initialize with an integer k and an initial batch of scores nums (which may contain fewer than k values).
add(val: int) -> int — append val to the stream and return the element that is currently the k-th largest counting duplicates (i.e. the k-th item of the stream sorted in descending order, not the k-th distinct value).

It is guaranteed that whenever add is called, the stream holds at least k elements.

Examples

KthLargest(3, [4, 5, 8, 2]), then add(3) -> 4, add(5) -> 5, add(10) -> 5, add(9) -> 8, add(4) -> 8 — after each insertion the 3rd largest of the whole stream is returned.
KthLargest(1, []), then add(-3) -> -3, add(-2) -> -2 — with k = 1 the answer is simply the maximum so far.
KthLargest(2, [7, 7]), then add(7) -> 7 — duplicates count separately, so the 2nd largest of [7, 7, 7] is 7.

Constraints

1 <= k <= 10^4
0 <= len(nums) <= 10^4
-10^4 <= val <= 10^4
Up to 10^4 calls to add.

Think about it first

Hint 1

You never need the whole sorted stream — after every add you only report one specific value. How much of the stream actually matters for that answer?

Hint 2

Only the k largest elements seen so far can ever influence the answer, and the answer is the smallest among those k. What structure gives you cheap access to the smallest of a set while supporting inserts?

Hint 3

Keep a min-heap capped at size k. On each add, push the value, and if the heap exceeds k elements pop the minimum. The heap's root is then exactly the k-th largest.

TL;DR

Size-k min-heap: add is O(log k) time, O(k) space.

Approach 1 — Brute force (re-sort on every add)

Store every value; on each add, sort descending and return index k - 1.

from typing import List


class KthLargest:
    def __init__(self, k: int, nums: List[int]):
        self.k = k
        self.nums = list(nums)

    def add(self, val: int) -> int:
        self.nums.append(val)
        self.nums.sort(reverse=True)
        return self.nums[self.k - 1]

Each add costs O(n log n) with n the stream length so far; with up to 10^4 adds on a stream that grows to 2·10^4 elements, that is hundreds of millions of comparison steps — the per-call sort is what the constraints are designed to punish.

Approach 2 — Min-heap of size k

The insight: the k-th largest element is the minimum of the k largest. So keep only the k largest values seen so far in a min-heap; anything that falls out of that top-k club can never re-enter it, so it is safe to throw away. The heap root is the answer at all times.

A binary min-heap is a complete binary tree (stored as an array, Python’s heapq) where every parent is ≤ its children, giving O(1) peek-min and O(log size) push/pop.

import heapq
from typing import List


class KthLargest:
    def __init__(self, k: int, nums: List[int]):
        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]

Walkthrough of KthLargest(3, [4, 5, 8, 2]):

Init: heapify [4, 5, 8, 2], then pop once (size 4 > 3) removing 2 → heap holds {4, 5, 8}, root 4.
add(3): push 3 → {3, 4, 5, 8} → pop min 3 → root 4. Return 4.
add(5): push 5 → {4, 5, 5, 8} → pop 4 → {5, 5, 8}. Return 5.
add(10): push 10 → pop 5 → {5, 8, 10}. Return 5.
add(9): push 9 → pop 5 → {8, 9, 10}. Return 8.
add(4): 4 is pushed then immediately popped (it is the min) → {8, 9, 10}. Return 8.

Matches the expected outputs 4, 5, 5, 8, 8.

Complexity: __init__ O(n log n) (or O(n + (n−k) log n) with heapify-then-pop); each add O(log k). Space O(k).

Approach 3 — Slightly slicker add (guarded push)

The insight: once the heap already has k elements, a new value only matters if it beats the current root; heapq.heappushpop does push-then-pop in one sift, and skipping values <= root avoids even that.

import heapq
from typing import List


class KthLargest:
    def __init__(self, k: int, nums: List[int]):
        self.k = k
        self.heap = heapq.nlargest(k, nums)   # at most k items
        heapq.heapify(self.heap)

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

Same asymptotics — O(log k) per add, O(k) space — but each add does at most one sift instead of two, and often zero. On the walkthrough above the behavior is identical (e.g. add(4) is rejected up front because 4 <= 8).

Common pitfalls

Using a max-heap of everything: peek gives the 1st largest, not the k-th, and finding the k-th costs k pops per query.
Forgetting that nums may start with fewer than k elements — don’t index heap[0] before trimming logic runs; the guarantee is only that add is never asked before k elements exist.
Treating “k-th largest” as k-th distinct value; duplicates each occupy a slot ([7, 7, 7] with k = 2 → 7).
Rebuilding or re-sorting per call — the whole point of the streaming setup is amortized O(log k) updates.

Pattern takeaway

When a stream question asks for “the k-th largest/smallest so far,” keep the opposite-ordered heap capped at size k: a min-heap for k-largest (root = answer), a max-heap for k-smallest. Elements evicted from the top-k can never return, so the cap is safe, and every update is O(log k) regardless of stream length.