DSAPrep
EasyHeap / Priority Queue

Kth Largest Element In a Stream

Design a class to find the kth largest element in a stream of numbers.

Implement KthLargest: KthLargest(int k, int[] nums) initializes the object with the integer k and the stream of numbers nums. int add(int val) appends the number val to the stream and returns the element representing the kth largest element in the stream so far.

Example 1

Input: ["KthLargest", "add", "add", "add", "add", "add"], [[3, [4, 5, 8, 2]], [3], [5], [10], [9], [4]]
Output: [null, 4, 5, 5, 8, 8]
Explanation: KthLargest(3, [4,5,8,2]); after each add, the 3rd largest of the pool so far is returned.

Example 2

Input: ["KthLargest", "add", "add", "add", "add"], [[4, [7, 7, 7, 7, 8, 3]], [2], [10], [9], [9]]
Output: [null, 7, 7, 7, 8]

Constraints

  • 0 <= nums.length <= 10^4
  • 1 <= k <= nums.length + 1
  • -10^4 <= nums[i] <= 10^4
  • -10^4 <= val <= 10^4
  • At most 10^4 calls will be made to add.
View original on LeetCode ↗

We only ever need to answer “what is the kth largest value seen so far?” We do not need the whole pool sorted — just fast access to the smallest of the top k elements. A min-heap capped at size k does exactly this: its root is always the kth largest value, since everything smaller than that root has already been pushed out.

Sort on Every Call

Time O(n log n) per addSpace O(n)

The naive approach: keep a list of every value seen, and on each add, append the new value, sort, and read off the kth largest.

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


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

Correctness: trivially correct — sorting descending and indexing k-1 is the definition of the kth largest.

Complexity: each add re-sorts the entire growing list, costing O(n log n) where n is the number of elements seen so far. Over m calls this is O(m·n log n) — far too slow for the constraint of up to 10^4 calls.

Min-Heap of Size k

OptimalTime O(log k) per add, O(n log k) to buildSpace O(k)

Maintain a min-heap that never holds more than k elements — the k largest values seen so far. Its smallest element (the root) is, by definition, the kth largest overall. On add, push the new value, and if the heap grew past size k, pop the smallest (it can’t be a top-k value anymore).

import heapq


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

Tracing KthLargest(3, [4, 5, 8, 2]), then add(3), add(5), add(10) — heap contents shown sorted for readability (the actual array order inside the heap differs, but the root, index 0, is always the true minimum):

4
0
5
1
8
2
k = 3root = 4
1 / 7
comparingresult

Init: heapify [4,5,8,2], then pop the smallest until only 3 remain. Heap holds the 3 largest: root (min) = 4.

Correctness: the heap only ever holds the k largest elements seen so far — any element smaller than the current root has been evicted because it cannot be among the top k. The root of a min-heap is its minimum, which is exactly the kth largest of the full pool.

Complexity: each add does at most one push and one pop, each O(log k) since the heap never exceeds size k. The constructor heapifies up to n initial elements in O(n) then pops down to size k in O((n-k) log k). This is optimal — you cannot answer “kth largest so far” faster than O(log k) amortized per insertion without extra structure.