All chapters in this track
Chapter 8 of 14
8. Heaps & Priority Queues
What Is a Heap?
A heap is a complete binary tree that satisfies the heap property:
- Min-heap: Every parent is smaller than or equal to its children. The minimum element is always at the root.
- Max-heap: Every parent is larger — the maximum is at the root.
Analogy: A hospital emergency room. Patients aren't served in order of arrival — the most critical patient always goes next. A priority queue does exactly this.
Real-world uses:
- Task schedulers — always run the highest-priority task next
- Dijkstra's shortest path — always explore the closest unvisited node
- Media streaming — buffer management, bandwidth allocation
- Event-driven simulations — process next scheduled event
Min-heap example:
1
/ \
3 2
/ \ /
7 5 4
8.1 Python's heapq
Python only provides a min-heap via the heapq module. For max-heap, negate values.
import heapq
# --- Min-heap operations ---
h: list[int] = []
heapq.heappush(h, 5) # push — O(log n)
heapq.heappush(h, 2)
heapq.heappush(h, 8)
heapq.heappush(h, 1)
print(h[0]) # peek minimum — O(1)
print(heapq.heappop(h)) # pop minimum → 1 — O(log n)
print(heapq.heappop(h)) # → 2
# --- Build heap from existing list in O(n) ---
nums = [5, 2, 8, 1, 9, 3]
heapq.heapify(nums) # transforms in-place
# --- Max-heap (negate values) ---
max_heap: list[int] = []
for x in [5, 2, 8, 1, 9]:
heapq.heappush(max_heap, -x) # store negative
largest = -heapq.heappop(max_heap) # negate back → 9
# --- nlargest / nsmallest (convenient for small k) ---
scores = [88, 95, 72, 100, 63, 91]
top3 = heapq.nlargest(3, scores) # [100, 95, 91]
bottom3 = heapq.nsmallest(3, scores) # [63, 72, 88]
8.2 Priority Queue Patterns
# --- K closest points to origin ---
import heapq, math
def k_closest(points: list[list[int]], k: int) -> list[list[int]]:
# Store (distance, point) in a min-heap
heap = [(math.sqrt(x**2 + y**2), [x, y]) for x, y in points]
heapq.heapify(heap)
return [heapq.heappop(heap)[1] for _ in range(k)]
print(k_closest([[1,3],[-2,2],[5,8],[0,1]], k=2)) # [[-2,2],[0,1]]
# --- Running median with two heaps ---
# Lower half: max-heap (negated) | Upper half: min-heap
class RunningMedian:
def __init__(self):
self.lower: list[int] = [] # max-heap (negate)
self.upper: list[int] = [] # min-heap
def add(self, num: int) -> None:
heapq.heappush(self.lower, -num)
# Balance: ensure max(lower) <= min(upper)
heapq.heappush(self.upper, -heapq.heappop(self.lower))
# Keep lower at same size or one larger
if len(self.upper) > len(self.lower):
heapq.heappush(self.lower, -heapq.heappop(self.upper))
def median(self) -> float:
if len(self.lower) > len(self.upper):
return float(-self.lower[0])
return (-self.lower[0] + self.upper[0]) / 2
rm = RunningMedian()
for x in [5, 3, 8, 4, 2]:
rm.add(x)
print(rm.median()) # 5 → 4.0 → 5 → 4.5 → 4