Data Structures
  • Build priority queue with O(log n) insert/extract for task scheduling
  • Find top K elements in O(n log k) using min heap of size K
  • Track running median with two heaps: max heap for smaller half, min heap for larger

Heap Applications

Heaps solve specific problems efficiently. Let's see where they shine.

Priority Queue

The classic heap application. Tasks with priorities. Process highest priority first.

import heapq

class PriorityQueue:
    def __init__(self):
        self.heap = []
        self.counter = 0  # Tie-breaker for same priority
    
    def push(self, priority, task):
        # Negative priority for max heap behavior
        heapq.heappush(self.heap, (-priority, self.counter, task))
        self.counter += 1
    
    def pop(self):
        if not self.heap:
            return None
        priority, _, task = heapq.heappop(self.heap)
        return task
    
    def is_empty(self):
        return len(self.heap) == 0

# Usage
pq = PriorityQueue()
pq.push(3, "Low priority")
pq.push(10, "High priority")
pq.push(5, "Medium priority")

print(pq.pop())  # "High priority"
print(pq.pop())  # "Medium priority"
print(pq.pop())  # "Low priority"

Insert and extract in O(log n). Perfect for task scheduling, event-driven simulation.

Top K Elements

Find K largest (or smallest) elements.

Naive: Sort, take first K. O(n log n).

Heap: Use min heap of size K. O(n log k).

import heapq

def top_k_largest(nums, k):
    # Min heap of size k
    heap = nums[:k]
    heapq.heapify(heap)
    
    for num in nums[k:]:
        if num > heap[0]:  # Larger than smallest in heap
            heapq.heapreplace(heap, num)
    
    return heap

print(top_k_largest([3, 1, 5, 12, 2, 11], 3))  # [5, 12, 11]

Keep K largest in heap. Root is Kth largest. Each comparison and replacement is O(log k).

For K smallest, use max heap.

Merge K Sorted Lists

Given K sorted lists, merge into one sorted list.

Naive: Concatenate, sort. O(n log n) where n is total elements.

Heap: Use min heap with one element from each list. O(n log k).

import heapq

def merge_k_sorted(lists):
    heap = []
    result = []
    
    # Initialize heap with first element from each list
    for i, lst in enumerate(lists):
        if lst:
            heapq.heappush(heap, (lst[0], i, 0))  # (value, list_index, element_index)
    
    while heap:
        val, list_idx, elem_idx = heapq.heappop(heap)
        result.append(val)
        
        # Add next element from same list
        if elem_idx + 1 < len(lists[list_idx]):
            next_val = lists[list_idx][elem_idx + 1]
            heapq.heappush(heap, (next_val, list_idx, elem_idx + 1))
    
    return result

lists = [
    [1, 4, 7],
    [2, 5, 8],
    [3, 6, 9]
]
print(merge_k_sorted(lists))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

Heap size is at most K. Each of n elements pushed and popped once. O(n log k).

Running Median

Track median as numbers come in.

Use two heaps:

  • Max heap for smaller half
  • Min heap for larger half

Median is either root of max heap (if odd count) or average of both roots (if even).

import heapq

class MedianFinder:
    def __init__(self):
        self.small = []  # Max heap (negate values)
        self.large = []  # Min heap
    
    def add_num(self, num):
        # Add to max heap (small)
        heapq.heappush(self.small, -num)
        
        # Balance: largest from small must be ≤ smallest from large
        if self.small and self.large and (-self.small[0] > self.large[0]):
            val = -heapq.heappop(self.small)
            heapq.heappush(self.large, val)
        
        # Balance sizes: differ by at most 1
        if len(self.small) > len(self.large) + 1:
            val = -heapq.heappop(self.small)
            heapq.heappush(self.large, val)
        
        if len(self.large) > len(self.small):
            val = heapq.heappop(self.large)
            heapq.heappush(self.small, -val)
    
    def find_median(self):
        if len(self.small) > len(self.large):
            return -self.small[0]
        return (-self.small[0] + self.large[0]) / 2.0

mf = MedianFinder()
mf.add_num(1)
mf.add_num(2)
print(mf.find_median())  # 1.5

mf.add_num(3)
print(mf.find_median())  # 2

Each add is O(log n). Find median is O(1). Better than sorting after each addition (O(n log n)).

Kth Largest Element in Stream

Maintain Kth largest as elements come in.

Use min heap of size K.

import heapq

class KthLargest:
    def __init__(self, k, nums):
        self.k = k
        self.heap = nums
        heapq.heapify(self.heap)
        
        # Keep only k largest
        while len(self.heap) > k:
            heapq.heappop(self.heap)
    
    def add(self, val):
        heapq.heappush(self.heap, val)
        if len(self.heap) > self.k:
            heapq.heappop(self.heap)
        return self.heap[0]

kth = KthLargest(3, [4, 5, 8, 2])
print(kth.add(3))  # 4
print(kth.add(5))  # 5
print(kth.add(10)) # 5
print(kth.add(9))  # 8

Heap size never exceeds K. Each operation O(log k).

Task Scheduler with Cooldown

Schedule tasks. Each task type has cooldown period.

Use heap to track when each task can run next.

import heapq
from collections import Counter

def least_interval(tasks, n):
    counts = Counter(tasks)
    heap = [-count for count in counts.values()]  # Max heap
    heapq.heapify(heap)
    
    time = 0
    
    while heap:
        temp = []
        for _ in range(n + 1):
            if heap:
                temp.append(heapq.heappop(heap))
        
        for count in temp:
            count += 1  # Processed one instance
            if count < 0:  # Still instances remaining
                heapq.heappush(heap, count)
        
        time += len(temp) if not heap else n + 1
    
    return time

print(least_interval(['A','A','A','B','B','B'], 2))  # 8
# A -> B -> idle -> A -> B -> idle -> A -> B

Dijkstra's Shortest Path

Find shortest paths from source to all nodes.

Use min heap for unvisited nodes by distance.

import heapq

def dijkstra(graph, start):
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    
    heap = [(0, start)]  # (distance, node)
    
    while heap:
        dist, node = heapq.heappop(heap)
        
        if dist > distances[node]:
            continue
        
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(heap, (new_dist, neighbor))
    
    return distances

graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': []
}

print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Each edge examined once. Heap operations O(log V). Total: O((V + E) log V).

When to Use Heaps

Need repeated min/max access: Priority queues, event simulation.

Top K problems: K largest, K smallest. O(n log k) vs O(n log n) sorting.

Merge sorted sequences: K-way merge. O(n log k).

Streaming data with order: Running median, Kth largest in stream.

When NOT to Use Heaps

Need all elements sorted: Heap only gives min/max. Use sorting or BST.

Random access by value: Heaps don't support efficient search. Use hash table or BST.

Fixed size, all elements known: Sorting might be simpler.

The Versatility

Heaps solve:

  • Priority-based scheduling
  • Top K in O(n log k)
  • Merging sorted data
  • Streaming statistics
  • Graph algorithms (Dijkstra, Prim)

Understanding heap applications means recognizing when you need efficient min/max access or top K elements.

Master heaps, and you'll handle priority-based problems elegantly.