Heaps & Priority Queues
Always-get-the-smallest in O(log n). Top-K, merging sorted streams, and the two-heaps running-median pattern.
A priority queue answers one question efficiently: “What is the next most important item?” The priority might be the smallest number, largest score, earliest deadline, or shortest tentative distance. A heap is the array-backed data structure that makes that queue fast.
The trade-off is focused. A heap is not sorted; it only guarantees that the best item is at the top. That is exactly enough for top-K, scheduling, streaming medians, merging sorted lists, and shortest-path algorithms.
The mental model
A binary heap is a complete binary tree stored in an array. For a min-heap, every parent is less than or equal to its children, so the minimum is always at index
0. In Python, heapq implements a min-heap.
import heapq
heap = [5, 1, 9, 3]
heapq.heapify(heap) # O(n), rearranges in place
heap[0] # O(1) peek at the smallest
heapq.heappush(heap, 2) # O(log n)
smallest = heapq.heappop(heap) # O(log n)
heappush bubbles a new item up until the heap property is restored. heappop removes the root, moves the last item to the root, then bubbles it down. The height of a complete binary tree is O(log n), so both operations are O(log n). Peeking is O(1) because the best item is already at the root.
Python has no max-heap API in heapq. Negate numeric priorities when you need the largest item first:
max_heap = []
for x in [5, 1, 9, 3]:
heapq.heappush(max_heap, -x)
largest = -heapq.heappop(max_heap)
When each element has a priority and a payload, push tuples like (priority, value). Python compares tuples lexicographically, so ties fall through to the next field. Add a counter if payloads are not comparable.
i live at 2i+1 and 2i+2.Pattern recognition
Variations
Worked problems
Notice the difference between “sort once” and “keep the right frontier while you scan.” Heaps shine in the second shape.
Kth Largest Element in an Array
Given an unsorted array and an integer k, return the kth largest value. You do not need to return the values above it.
Approach. Keep a min-heap of the largest k values seen so far. The heap root is the smallest among those k, which is exactly the current kth largest. When a new value arrives, push it and remove the smallest if the heap grew past size k.
Show solution
import heapq
def find_kth_largest(nums, k):
heap = []
for x in nums:
heapq.heappush(heap, x)
if len(heap) > k:
heapq.heappop(heap)
return heap[0]Complexity. O(n log k) time and O(k) space. Sorting would be O(n log n); the heap avoids ordering values that cannot affect the kth largest answer.
Top K Frequent Elements
Given integers and k, return any k values with the highest frequencies.
Approach. Count first, then keep a min-heap of (frequency, value) pairs. Whenever the heap exceeds size k, remove the least frequent value among the current candidates. At the end, the heap contains the top k frequencies.
Show solution
from collections import Counter
import heapq
def top_k_frequent(nums, k):
counts = Counter(nums)
heap = []
for value, freq in counts.items():
heapq.heappush(heap, (freq, value))
if len(heap) > k:
heapq.heappop(heap)
return [value for freq, value in heap]Complexity. O(n + m log k) time and O(m) space for m distinct values. The count map costs O(n); the heap work is bounded by distinct values, not total length.
Find Median from Data Stream
Design a structure that receives numbers one at a time and returns the median of all numbers seen so far.
Approach. Split the stream into two halves. low is a max-heap implemented with negated numbers and stores the smaller half. high is a min-heap and stores the larger half. Keep sizes balanced so low is either the same size as high or one larger. The median is then one root or the average of two roots.
Show solution
import heapq
class MedianFinder:
def __init__(self):
self.low = [] # max-heap via negatives
self.high = [] # min-heap
def add_num(self, num):
heapq.heappush(self.low, -num)
heapq.heappush(self.high, -heapq.heappop(self.low))
if len(self.high) > len(self.low):
heapq.heappush(self.low, -heapq.heappop(self.high))
def find_median(self):
if len(self.low) > len(self.high):
return -self.low[0]
return (-self.low[0] + self.high[0]) / 2Complexity. add_num is O(log n) time and find_median is O(1). Space is O(n) for the stored stream. The two heap roots are the only values needed to answer the median query.