Heap β Priority Queue Implementation
After completing this topic, you will be able to:
Explain the structure and working principle of a heap, and implement a priority queue using Python's heapq to solve problems that require "quickly retrieving the smallest (or largest) value."
Problem: Quickly Retrieve the "Most Urgent" Item
Consider a hospital emergency room. Patients are treated based on the severity of their symptoms, not the order in which they arrived (a queue). How can we implement a "priority queue" like this?
- Sorted list: Insertion requires sorting every time β O(n)
- Unsorted list: Finding the minimum requires a full scan β O(n)
Heaps handle both insertion and minimum value retrieval in O(log n) time.
Heap Rules
The rules of a Min Heap are simple:
A parent node is always less than or equal to its children
1 β The smallest value is always at the root
/ \
3 5
/ \ / \
7 4 8 6Since the smallest value is always at the root, the minimum value can be checked in O(1) time.
Difference from BST: A BST has "left < parent < right," but a heap only maintains "parent < children." The order between siblings doesn't matter.
Representing it as an Array
A heap is a tree, but in practice, it is stored as an array:
Index: [0, 1, 2, 3, 4, 5, 6]
Value: [1, 3, 5, 7, 4, 8, 6]For a parent node at index i:
Left child = 2*i + 1
Right child = 2*i + 2
Parent = (i - 1) // 2The children of index 0 are 1 and 2; the children of index 1 are 3 and 4; the children of index 2 are 5 and 6. The parent-child relationship can be determined by index calculations without pointers.
Insertion and Deletion Principles
Insertion (heappush): Add to the end of the array and move up by comparing with the parent (sift up):
Initial: [1, 3, 5, 7, 4, 8, 6]
Insert 2: [1, 3, 5, 7, 4, 8, 6, 2]
β Index 7's parent (3) = index 3 (value 7)
2 < 7 β swap: [1, 3, 5, 2, 4, 8, 6, 7]
β Index 3's parent = index 1 (value 3)
2 < 3 β swap: [1, 2, 5, 3, 4, 8, 6, 7]
β Index 1's parent = index 0 (value 1)
2 > 1 β stopDeletion (heappop): Remove the root, place the last element at the root, and move down by comparing with children (sift down). Both operations only move through the height of the tree, so they are O(log n).
Python heapq
Python provides a min-heap with the heapq module:
import heapq
# Use an empty list as the heapheap = []
# Insertion β O(log n)heapq.heappush(heap, 5)heapq.heappush(heap, 3)heapq.heappush(heap, 7)heapq.heappush(heap, 1)
print(heap) # [1, 3, 7, 5] β Internal order follows heap rulesprint(heap[0]) # 1 β The minimum value is always at index 0
# Extract the minimum value β O(log n)smallest = heapq.heappop(heap)print(smallest) # 1print(heap) # [3, 5, 7]Just remember heappush and heappop.
Converting an Existing List to a Heap
data = [9, 1, 4, 7, 2, 8, 3]heapq.heapify(data) # O(n) β Faster than sorting (O(n log n))print(data) # [1, 2, 3, 7, 9, 8, 4]heapify converts the list into a heap in place.
Implementing a Priority Queue
import heapq
class PriorityQueue: def __init__(self): self.heap = [] def push(self, priority, item): heapq.heappush(self.heap, (priority, item)) def pop(self): return heapq.heappop(self.heap)[1] def is_empty(self): return len(self.heap) == 0
# Emergency room exampleer = PriorityQueue()er.push(3, "Patient with a cold")er.push(1, "Patient in cardiac arrest") # Priority 1 is the most urgenter.push(2, "Patient with a fracture")
print(er.pop()) # "Patient in cardiac arrest"print(er.pop()) # "Patient with a fracture"print(er.pop()) # "Patient with a cold"The first element of the tuple is the comparison criterion. Smaller numbers have higher priority.
When You Need a Max Heap
Python's heapq only provides a min-heap. If you need a max-heap, put the values as negative numbers:
# Max heap effectmax_heap = []for val in [3, 1, 5, 2, 4]: heapq.heappush(max_heap, -val)
print(-heapq.heappop(max_heap)) # 5 (the largest value)print(-heapq.heappop(max_heap)) # 4Useful Functions
data = [7, 3, 9, 1, 5, 8, 2]
# The 3 smallestprint(heapq.nsmallest(3, data)) # [1, 2, 3]
# The 3 largestprint(heapq.nlargest(3, data)) # [9, 8, 7]
# Merge multiple sorted lists into onea = [1, 4, 7]b = [2, 5, 8]c = [3, 6, 9]print(list(heapq.merge(a, b, c))) # [1, 2, 3, 4, 5, 6, 7, 8, 9]Time Complexity Summary
| Operation | Time Complexity |
|---|---|
| Insertion (heappush) | O(log n) |
| Extract Minimum (heappop) | O(log n) |
| Check Minimum (heap[0]) | O(1) |
| Heapify | O(n) |
| Purpose | Data Structure |
|---|---|
| Search by exact key | Hash table |
| Maintain sorted order | BST |
| Quickly get the minimum/maximum value only | Heap |
Heap Sort
You can sort using a heap:
def heap_sort(arr): heapq.heapify(arr) # O(n) return [heapq.heappop(arr) for _ in range(len(arr))]
data = [7, 3, 9, 1, 5]print(heap_sort(data)) # [1, 3, 5, 7, 9]Time complexity is O(n log n), the same as quicksort and merge sort. In practice, Python's sorted() (Timsort) is faster, but understanding the principle of heap sort is important.
Real-World Example β Top K Problem
"Find the 10 largest values out of 1 million data points":
import heapqimport random
data = [random.randint(0, 1_000_000) for _ in range(1_000_000)]
# β Sort everything β O(n log n)top10_sort = sorted(data, reverse=True)[:10]
# β
Use a heap β O(n log k)top10_heap = heapq.nlargest(10, data)Sorting everything takes O(n log n), but the heap only maintains size k, so it's O(n log k). This makes a big difference when k is much smaller than n.
# Implementing it yourself shows the principlemin_heap = []for val in data: if len(min_heap) < 10: heapq.heappush(min_heap, val) elif val > min_heap[0]: heapq.heapreplace(min_heap, val)
top10 = sorted(min_heap, reverse=True)Maintain a min-heap of size 10. If a new value is greater than the minimum value in the heap, replace it. The remaining 10 values are the 10 largest in the entire dataset.
Real-World Example β Dijkstra's Algorithm
Dijkstra's algorithm, which finds the shortest path, also uses a heap:
def dijkstra(graph, start): dist = {node: float('inf') for node in graph} dist[start] = 0 heap = [(0, start)] while heap: cost, node = heapq.heappop(heap) if cost > dist[node]: continue for neighbor, weight in graph[node]: new_cost = cost + weight if new_cost < dist[neighbor]: dist[neighbor] = new_cost heapq.heappush(heap, (new_cost, neighbor)) return distYou need to repeatedly retrieve the unvisited node with the shortest distance, so using a heap makes it O((V+E) log V) instead of O(VΒ²).
Heaps are optimal for situations where "you don't need to sort everything, but you need to quickly retrieve the most important item."