Use heaps for top-k, k-way merge, scheduling, and streaming medians
Heaps maintain partial order to get min/max quickly; support merging/scheduling efficiently.
Use when repeatedly extracting min/max, merging k sorted inputs, or balancing two halves for median.
Size-k heap to keep best candidates
import heapq
def top_k(nums, k):
heap = []
for x in nums:
if len(heap) < k: heapq.heappush(heap, x)
elif x > heap[0]: heapq.heapreplace(heap, x)
return sorted(heap, reverse=True)
Push head of each list/array; pop/push next
import heapq
def merge_k_sorted(arrs):
heap = []
for i, arr in enumerate(arrs):
if arr: heapq.heappush(heap, (arr[0], i, 0))
res = []
while heap:
val, i, j = heapq.heappop(heap)
res.append(val)
if j+1 < len(arrs[i]): heapq.heappush(heap, (arrs[i][j+1], i, j+1))
return res
Greedy by frequency with cooldown
import heapq
from collections import Counter, deque
def least_interval(tasks, n):
freq = Counter(tasks)
heap = [-c for c in freq.values()]
heapq.heapify(heap)
time = 0
q = deque() # (ready_time, count)
while heap or q:
time += 1
if heap:
cnt = 1 + heapq.heappop(heap) # negative counts
if cnt != 0:
q.append((time + n, cnt))
if q and q[0][0] == time:
heapq.heappush(heap, q.popleft()[1])
return time
Max-heap left + min-heap right for median
import heapq
class MedianFinder:
def __init__(self):
self.small = [] # max-heap via negatives
self.large = [] # min-heap
def addNum(self, num: int) -> None:
if not self.small or num <= -self.small[0]:
heapq.heappush(self.small, -num)
else:
heapq.heappush(self.large, num)
# balance
if len(self.small) > len(self.large) + 1:
heapq.heappush(self.large, -heapq.heappop(self.small))
if len(self.large) > len(self.small):
heapq.heappush(self.small, -heapq.heappop(self.large))
def findMedian(self) -> float:
if len(self.small) > len(self.large):
return float(-self.small[0])
return (-self.small[0] + self.large[0]) / 2.0
def template(*args, **kwargs):
# General template placeholder
pass
Push/pop heap operations are logarithmic.
Heap stores elements.