BeginnerPython · 25 min

Heap & Top-K: Beginner

Here is the single thing that separates a candidate who knows heap problems from one who just knows sorting: when K is much smaller than N, you do not need to sort the entire list. You only need to maintain the top K elements at any given time. That is what a heap does. Python gives you heapq, and heapq is a min-heap only. That one sentence, 'Python only gives you a min-heap, but you can simulate a max-heap by negating the values,' is the gotcha that filters out candidates who only studied the concept but never wrote the code. Learn it cold. The heap pattern shows up in phone screens at every major tech company because it directly maps to what data engineers do every day: finding the top-K rows, the slowest queries, the highest-revenue customers.

What you will be able to do

Use heapq confidently: heappush, heappop, heapify, nlargest, nsmallest

Maintain a size-K min-heap to find top-K largest elements in O(n log k)

Solve Kth largest element and explain both approaches to the interviewer

Use tuple heaps for priority-ordered processing

Connect heap patterns to real DE work: top revenue customers, slowest queries

heapq Module Fundamentals

Daily Life

Interviews

Use heapq confidently for any heap operation

What you absolutely must know before any heap interview question:

▸Python's heapq is a MIN-heap only. heappop() returns the SMALLEST element
▸To simulate a max-heap, negate all values before pushing and negate again on pop
▸heapq.heapify(list) transforms a list in-place in O(n), beating a full sort
▸heapq.nlargest(k, iterable) and nsmallest(k, iterable) are convenience wrappers
▸heappush and heappop are both O(log n). That log n cost is the key to all heap efficiency

Every heap interview problem in Python starts with one fact: heapq only gives you a min-heap. The root is always the smallest element. When you heappop(), you get the smallest. When you heappush(), the heap rebalances to maintain that invariant in O(log n). This is not a bug; it is by design. The reason Python only provides a min-heap is that a max-heap is trivially simulated by negating values, and providing both would add library surface area for no real gain. The interviewer knows this, and they will sometimes ask you to explicitly explain it.

	import heapq

	# --- Basic min-heap operations ---
	min_heap = []
	heapq.heappush(min_heap, 5)
	heapq.heappush(min_heap, 2)
	heapq.heappush(min_heap, 8)
	heapq.heappush(min_heap, 1)

	print(heapq.heappop(min_heap)) # 1 — smallest out first
	print(min_heap[0]) # peek at smallest without popping

	# --- heapify: convert existing list in O(n) ---
	data = [5, 2, 8, 1, 9, 3]
	heapq.heapify(data) # in-place, O(n)
	print(data[0]) # 1

	# --- Simulating a max-heap by negating ---
	max_heap = []
	for val in [5, 2, 8, 1, 9, 3]:
	heapq.heappush(max_heap, -val) # negate before push

	largest = -heapq.heappop(max_heap) # negate again on pop
	print(largest) # 9

	# --- Convenience functions ---
	nums = [5, 2, 8, 1, 9, 3, 7]
	print(heapq.nlargest(3, nums)) # [9, 8, 7]
	print(heapq.nsmallest(3, nums)) # [1, 2, 3]

nlargest vs sorted: When to Use Each

heapq.nlargest(k, data) is O(n log k). sorted(data, reverse=True)[:k] is O(n log n). When k is much smaller than n, nlargest wins by a large margin. When k approaches n, nlargest degrades and sorted is simpler. The crossover point where sorted becomes competitive is roughly k > n/2, but for most real-world top-K queries where you want the top 10 or top 100 out of millions of rows, nlargest is strictly better. In interviews, mention this tradeoff explicitly. It shows you think about scalability.

✓sorted()[:k]

O(n log n) always
Sorts the entire collection
Simpler code: one line
Better only when k is close to n

✗heapq.nlargest(k)

O(n log k) time
Processes stream without storing all
Ideal when k << n
Better for top-10 of 10 million rows

Say this in the interview:

▸"I could sort and slice, which is O(n log n), but since k is much smaller than n, I'll maintain a heap of size k for O(n log k). On a dataset of 10 million rows with k=10, that's roughly 10 million * 4 operations vs 10 million * 23 operations, so about 6x faster."

TIP

heapq.nlargest and nsmallest accept a key= argument just like sorted(). heapq.nlargest(5, employees, key=lambda e: e.salary) gives you the 5 highest-paid employees without sorting the full list. Use this pattern constantly in DE work.

Top-K Largest and Smallest: The Canonical Pattern

Daily Life

Interviews

Implement top-K with O(n log k) heap pattern

This is the most important heap pattern you will ever learn for interviews. Finding the top-K largest elements from a list of N elements. The naive approach: sort descending, take first K. O(n log n). The heap approach: maintain a min-heap of size exactly K. For every element you process, if it is larger than the heap's minimum (the root), push it in and pop the minimum out. At the end, the heap contains exactly the K largest elements. Time: O(n log k). Space: O(k).

The Size-K Min-Heap Pattern for Top-K Largest

This feels counterintuitive at first. You are finding the K LARGEST elements, but you are using a MIN-heap. Here is why: the min-heap keeps the K largest elements you have seen so far, and its root is the smallest of those K elements. That root is your eviction candidate. Every new element competes against the current minimum. If the new element is bigger, it evicts the smallest of the top K. If it is smaller, it gets ignored. You never need to see the full sorted list.

	import heapq

	def top_k_largest(nums, k):
	"""Find K largest elements. O(n log k) time, O(k) space."""
	if k >= len(nums):
	return sorted(nums, reverse=True) # edge case

	# Build a min-heap of the first k elements
	heap = nums[:k]
	heapq.heapify(heap) # O(k)

	# Process remaining elements
	for num in nums[k:]:
	if num > heap[0]: # larger than current minimum?
	heapq.heapreplace(heap, num) # pop min, push new (atomic, faster)

	return sorted(heap, reverse=True) # optional: return sorted

	# Example: top 3 from 8 elements
	nums = [3, 1, 4, 1, 5, 9, 2, 6]
	print(top_k_largest(nums, 3)) # [9, 6, 5]

	# heapreplace is faster than heappop + heappush
	# Use it when you know the heap is non-empty

heapq.heapreplace(heap, item) is an atomic pop-then-push that is faster than calling heappop() and heappush() separately. Use it whenever you know the heap is non-empty and you want to swap the minimum for a new value. In a tight loop over millions of records, this matters.

Why log k Matters When k << n

n = 10,000,000 rows in a customer revenue table

k = 10 (top 10 customers)

sorted approach: O(n log n) = ~230 million operations

heap approach: O(n log k) = ~33 million operations, 7x faster

At 100 million rows, this is the difference between seconds and minutes

	import heapq

	def top_k_smallest(nums, k):
	"""Find K smallest elements. O(n log k) time, O(k) space."""
	# Use a MAX-heap of size k (negate values)
	# Root is the largest of our k smallest candidates — eviction target
	heap = [-num for num in nums[:k]]
	heapq.heapify(heap)

	for num in nums[k:]:
	if num < -heap[0]: # smaller than current max of our k-smallest?
	heapq.heapreplace(heap, -num)

	return sorted(-x for x in heap)

	# Alternatively — use nsmallest directly:
	print(heapq.nsmallest(3, [3, 1, 4, 1, 5, 9, 2, 6])) # [1, 1, 2]

TIP

In an interview, always state the complexity contrast upfront: 'Naive sort is O(n log n). Since k is small, I'll maintain a size-k heap for O(n log k). This is the canonical top-K pattern and it matters a lot when n is in the hundreds of millions.' Two sentences. Instant credibility.

Kth Largest Element: Two Approaches

Daily Life

Interviews

Solve Kth largest with both approaches and explain tradeoffs

LeetCode 215 is one of the most frequently asked heap problems at FAANG and FAANG-adjacent companies. Find the Kth largest element in an unsorted array. It is deceptively simple, but interviewers use it to filter candidates who know the theory from candidates who know when to apply which tool. There are two approaches you need to know: sort (simple, O(n log n)) and heap (efficient, O(n log k)). You should be able to code both and explain when each is appropriate.

Approach 1: Sort

	def find_kth_largest_sort(nums, k):
	"""Simple. O(n log n) time, O(1) or O(n) space depending on sort."""
	return sorted(nums, reverse=True)[k - 1]

	# Example: 2nd largest of [3, 2, 1, 5, 6, 4]
	print(find_kth_largest_sort([3, 2, 1, 5, 6, 4], 2)) # 5

The sort approach is fine for small n or when you need to impress with simplicity. Say it first. 'The straightforward approach is to sort descending and return index k-1. O(n log n) time.' Then offer the better solution. Interviewers appreciate that you know the easy solution before jumping to the complex one. It shows you are not just pattern-matching; you are reasoning.

Approach 2: Min-Heap of Size K

	import heapq

	def find_kth_largest_heap(nums, k):
	"""Efficient. O(n log k) time, O(k) space."""
	# Maintain a min-heap of exactly k elements
	# The root (minimum of the heap) is the kth largest overall
	heap = []
	for num in nums:
	heapq.heappush(heap, num)
	if len(heap) > k:
	heapq.heappop(heap) # evict smallest; keep only top k
	return heap[0] # root = kth largest

	# Example
	print(find_kth_largest_heap([3, 2, 1, 5, 6, 4], 2)) # 5
	print(find_kth_largest_heap([3, 2, 3, 1, 2, 4, 5, 5, 6], 4)) # 4

The insight: after processing all elements, the heap contains the K largest values seen. The smallest of those K values (the root of the min-heap) is exactly the Kth largest. It is sitting right there at heap[0]. No sorting needed. Time: O(n log k). Space: O(k). If k=2, the heap only ever holds 2 elements and every push-pop pair is O(log 2) = O(1) effectively. This is the difference that matters at scale.

What to say when the interviewer asks 'can you do better than O(n log n)?'

▸"Yes. I can maintain a size-k min-heap as I scan the array. Every element is either pushed or discarded in O(log k). Total time is O(n log k), which is asymptotically better when k is much smaller than n."
▸Bonus: mention QuickSelect (O(n) average) if they push further, but do not lead with it unless asked

Approach	Time	Space	When to Use
sorted()[:k]	O(n log n)	O(n)	k close to n, or n is small, or code clarity matters
Min-heap of size k	O(n log k)	O(k)	k << n, streaming data, memory-constrained
QuickSelect	O(n) avg, O(n²) worst	O(1)	n is huge, k is exact, can tolerate worst case
heapq.nlargest(k)	O(n log k)	O(k)	Same as min-heap; use when code brevity is fine

The DE use case that comes up constantly: find the top-10 highest revenue customers from a table with 50 million rows. You cannot sort 50 million rows just to get 10. You maintain a size-10 heap. The heap approach is not academic. It is what you do when the data does not fit in sorted memory.

TIP

In the DE interview context, frame Kth largest as: 'This maps directly to finding the 10th percentile query latency or the top-10 revenue customers. The heap approach lets me solve it in a single scan with O(k) memory, which is critical when the dataset is too large to sort in memory.'

Heap with Tuples: Priority and Tie-Breaking

Daily Life

Interviews

Use tuple heaps safely with correct tie-breaking

Python heaps compare tuples lexicographically: first by the first element, then by the second if there is a tie, then by the third. This is incredibly useful for priority queues where you want to order by one field and break ties by another. It is also a gotcha: if two tuples have the same priority value and the second element is an uncomparable type (like a custom object without __lt__), heapq will raise a TypeError. Knowing this cold is how you avoid a humiliating bug in a live interview.

Priority Queue with (priority, item) Tuples

	import heapq

	# Basic (priority, item) pattern
	tasks = []
	heapq.heappush(tasks, (3, 'low priority task'))
	heapq.heappush(tasks, (1, 'high priority task'))
	heapq.heappush(tasks, (2, 'medium priority task'))

	while tasks:
	priority, task = heapq.heappop(tasks)
	print(f'Priority {priority}: {task}')
	# Priority 1: high priority task
	# Priority 2: medium priority task
	# Priority 3: low priority task

	# --- Tie-breaking with a counter ---
	import itertools

	counter = itertools.count() # unique sequence number
	heap = []

	heapq.heappush(heap, (1, next(counter), 'task A')) # (priority, seq, item)
	heapq.heappush(heap, (1, next(counter), 'task B')) # same priority, earlier seq wins
	heapq.heappush(heap, (2, next(counter), 'task C'))

	priority, seq, task = heapq.heappop(heap)
	print(task) # 'task A' — same priority, task A came first

The (priority, counter, item) pattern is the canonical safe way to use heaps with tuples. The counter guarantees that no two tuples ever have the same first two elements, which means heapq never needs to compare the item directly. Without the counter, if two tasks have the same priority and the item is a custom object that does not implement __lt__, Python raises a TypeError. This is a live-coding land mine. The counter sidesteps it completely.

Lexicographic Comparison: Useful and Occasionally a Gotcha

	import heapq

	# Useful: sort events by (timestamp, event_id) for stable ordering
	events = [
	(1714000003, 'event_C'),
	(1714000001, 'event_A'),
	(1714000001, 'event_B'), # same timestamp, 'B' > 'A' alphabetically
	]
	heapq.heapify(events)
	print(heapq.heappop(events)) # (1714000001, 'event_A') -- stable!
	print(heapq.heappop(events)) # (1714000001, 'event_B')
	print(heapq.heappop(events)) # (1714000003, 'event_C')

	# Gotcha: custom objects without __lt__ will crash
	class Job:
	def __init__(self, priority, name):
	self.priority = priority
	self.name = name

	heap = []
	heapq.heappush(heap, (1, Job(1, 'job_A'))) # OK
	heapq.heappush(heap, (1, Job(1, 'job_B'))) # CRASH on comparison!

The Tuple Comparison Gotcha

▸If two tuples tie on the first element, Python compares the second element
▸If the second element is a custom object without __lt__, you get: TypeError: '<' not supported
▸Fix: always insert a unique counter as the second element: (priority, counter, item)
▸This guarantees no tuple comparison ever reaches the custom object

Safe Pattern
Readable Pattern
Inversion Pattern

TIP

Interviewers love asking about the tie-breaking problem because it exposes whether you actually wrote heap code or just read about it. Drop the counter pattern naturally: 'I'll use (priority, seq, item) with an itertools.count() to avoid comparison errors on ties.' That one sentence signals you have actually debugged this in production.

Basic DE Applications

Daily Life

Interviews

Apply heap patterns to real DE problems confidently

Here is where you convert a coding answer into a data engineering answer. Heap and top-K problems are not just LeetCode exercises. They show up constantly in DE work: finding the most frequent log errors to prioritize, identifying the slowest queries for optimization, finding the K closest events to a target timestamp for alignment. Every time you frame your heap solution in terms of real DE problems, the interviewer writes 'strong domain understanding' on the scorecard.

Top-K Most Frequent Log Errors

	import heapq
	from collections import Counter

	def top_k_frequent_errors(log_lines, k):
	"""Find K most frequent error types from log stream.
	O(n log k) time. Classic phone screen problem.
	"""
	error_counts = Counter(log_lines) # O(n)
	# nlargest with key extracts top-k by count
	return heapq.nlargest(k, error_counts, key=error_counts.get)

	errors = [
	'ConnectionTimeout', 'NullPointer', 'ConnectionTimeout',
	'DiskFull', 'ConnectionTimeout', 'NullPointer', 'OOMError'
	]
	print(top_k_frequent_errors(errors, 2)) # ['ConnectionTimeout', 'NullPointer']

The Counter + nlargest pattern is the most common real-world heap problem in DE. Log aggregation, error monitoring, and usage analytics all reduce to this: count frequencies, find the top K. In production, you might be doing this over a Spark DataFrame with groupBy().count().orderBy(desc()).limit(k). That is the distributed version of exactly this algorithm, and knowing the single-machine Python version makes you dangerous in either context.

Finding the K Slowest Queries

	import heapq

	def k_slowest_queries(query_log, k):
	"""Find k slowest queries from a query log.
	query_log: list of (duration_ms, query_text, timestamp)
	Returns the k entries with highest duration.
	"""
	# nlargest with key=lambda selects top-k by first tuple element
	return heapq.nlargest(k, query_log, key=lambda x: x[0])

	query_log = [
	(120, 'SELECT * FROM orders WHERE ...', '2024-01-01 10:00'),
	(4500, 'SELECT customer_id, SUM(revenue) FROM ...', '2024-01-01 10:01'),
	(89, 'SELECT COUNT(*) FROM events', '2024-01-01 10:02'),
	(2300, 'SELECT * FROM sessions JOIN ...', '2024-01-01 10:03'),
	(7100, 'SELECT DISTINCT user_id FROM ...', '2024-01-01 10:04'),
	]

	slowest = k_slowest_queries(query_log, 2)
	for duration, query, ts in slowest:
	print(f'{duration}ms: {query[:40]}...')

K Closest Timestamps to a Target

	import heapq
	from datetime import datetime

	def k_closest_timestamps(timestamps, target, k):
	"""Find k timestamps closest to the target.
	Useful for: aligning events from different sources,
	joining streams with time skew, watermark analysis.
	"""
	# Use nsmallest with key=absolute distance to target
	return heapq.nsmallest(
	k,
	timestamps,
	key=lambda ts: abs((ts - target).total_seconds())
	)

	target = datetime(2024, 1, 1, 12, 0, 0)
	events = [
	datetime(2024, 1, 1, 11, 55, 0), # 5 min before
	datetime(2024, 1, 1, 12, 30, 0), # 30 min after
	datetime(2024, 1, 1, 11, 58, 0), # 2 min before
	datetime(2024, 1, 1, 13, 0, 0), # 1 hour after
	datetime(2024, 1, 1, 12, 1, 0), # 1 min after
	]
	closest = k_closest_timestamps(events, target, 2)
	print(closest) # [12:01, 11:58]

The three DE heap scenarios to have ready in every interview:

▸Top-K frequent: Counter + nlargest(k, counts, key=counts.get), for log errors, query patterns, user events
▸Top-K by metric: nlargest(k, records, key=lambda r: r.metric), for slowest queries, highest revenue, most errors
▸K closest: nsmallest(k, records, key=lambda r: abs(r.value - target)), for timestamp alignment, watermark skew analysis

✓Do

Always state the O(n log k) complexity and contrast with O(n log n) sort
Use heapq.nlargest/nsmallest for clean one-liners in interviews when brevity helps
Explicitly mention the DE analogy after solving: 'same as ORDER BY revenue DESC LIMIT 10'
Handle the edge case: if k >= len(data), return all elements sorted

✗Don't

Never sort the entire collection just to take the top K when n is large
Don't forget to negate values when simulating a max-heap with numeric data
Don't use heappush on an unhashable or uncomparable object without a counter guard

TIP

After solving any top-K problem, bridge to DE immediately: 'In production, I use this for identifying the top-10 error codes in our data quality pipeline. We count errors per type over a rolling window, then heap-select the top 10 to prioritize on-call investigation.' That one sentence ends the interview on a strong note.

❯❯❯PUTTING IT ALL TOGETHER

> You are in a phone screen. The interviewer asks: 'Given a list of log entries with error types and counts, find the 5 most frequent error types as efficiently as possible.'

You say: 'I'll use a Counter for O(n) frequency counting, then heapq.nlargest for O(n log k) selection. Total: O(n log k), far better than sorting which is O(n log n). For k=5 and n=10 million, that's roughly a 5x improvement.'

You write Counter + nlargest in under 2 minutes. You handle the edge case where k >= number of distinct errors. You name the complexity unprompted.

You bridge: 'This exact pattern powers our production error monitoring dashboard. We count error codes over a 1-hour rolling window and surface the top 10 to the on-call engineer. Same algorithm, production scale.'

KEY TAKEAWAYS

Python heapq is min-heap only: negate values to simulate a max-heap. Know this before walking in.

O(n log k) vs O(n log n): maintain a size-k heap instead of sorting. The difference is massive when k << n.

Size-k min-heap for top-K largest: root is always the Kth largest and serves as the eviction threshold. Every new element competes against it.

Tuple heaps need a counter: (priority, counter, item) with itertools.count() avoids TypeError on ties.

DE applications: top-K frequent errors, slowest queries, closest timestamps. Bridge to production after every solution.

Stop sorting the whole list. Keep only what you need.

Category: Python
Difficulty: beginner
Duration: 25 minutes
Challenges: 0 hands-on challenges

Topics covered: heapq Module Fundamentals, Top-K Largest and Smallest: The Canonical Pattern, Kth Largest Element: Two Approaches, Heap with Tuples: Priority and Tie-Breaking, Basic DE Applications

Lesson Sections

heapq Module Fundamentals (concepts: pyHeapqModule, pyMinHeap, pyMaxHeapSimulation)
Every heap interview problem in Python starts with one fact: heapq only gives you a min-heap. The root is always the smallest element. When you heappop(), you get the smallest. When you heappush(), the heap rebalances to maintain that invariant in O(log n). This is not a bug; it is by design. The reason Python only provides a min-heap is that a max-heap is trivially simulated by negating values, and providing both would add library surface area for no real gain. The interviewer knows this, and t
Top-K Largest and Smallest: The Canonical Pattern (concepts: pyTopKLargest, pyTopKSmallest, pyHeapReplace)
This is the most important heap pattern you will ever learn for interviews. Finding the top-K largest elements from a list of N elements. The naive approach: sort descending, take first K. O(n log n). The heap approach: maintain a min-heap of size exactly K. For every element you process, if it is larger than the heap's minimum (the root), push it in and pop the minimum out. At the end, the heap contains exactly the K largest elements. Time: O(n log k). Space: O(k). The Size-K Min-Heap Pattern f
Kth Largest Element: Two Approaches (concepts: pyKthLargest, pyMinHeapTopK, pyQuickSelect)
LeetCode 215 is one of the most frequently asked heap problems at FAANG and FAANG-adjacent companies. Find the Kth largest element in an unsorted array. It is deceptively simple, but interviewers use it to filter candidates who know the theory from candidates who know when to apply which tool. There are two approaches you need to know: sort (simple, O(n log n)) and heap (efficient, O(n log k)). You should be able to code both and explain when each is appropriate. Approach 1: Sort The sort approa
Heap with Tuples: Priority and Tie-Breaking (concepts: pyHeapTuples, pyTieBreaking, pyIterToolsCount)
Python heaps compare tuples lexicographically: first by the first element, then by the second if there is a tie, then by the third. This is incredibly useful for priority queues where you want to order by one field and break ties by another. It is also a gotcha: if two tuples have the same priority value and the second element is an uncomparable type (like a custom object without __lt__), heapq will raise a TypeError. Knowing this cold is how you avoid a humiliating bug in a live interview. Prio
Basic DE Applications (concepts: pyTopKFrequent, pySlowestQueries, pyKClosest)
Here is where you convert a coding answer into a data engineering answer. Heap and top-K problems are not just LeetCode exercises. They show up constantly in DE work: finding the most frequent log errors to prioritize, identifying the slowest queries for optimization, finding the K closest events to a target timestamp for alignment. Every time you frame your heap solution in terms of real DE problems, the interviewer writes 'strong domain understanding' on the scorecard. Top-K Most Frequent Log