Find Median from Data Stream

▶

Steps

setup

Initialize MedianFinder

Create empty max heap 'small' and min heap 'large', initialize delayed removals map and size counters.

💡 Starting with empty heaps and counters sets the foundation for adding numbers and balancing heaps.

Line:

self.small = []  # max heap (invert values)
self.large = []  # min heap
self.delayed = {}
self.small_size = 0
self.large_size = 0

💡 Heaps and counters are ready to track numbers and maintain balance.

insert

Add number 1 to 'small' heap

Since 'small' is empty, add 1 to 'small' as -1 (max heap stores negatives). Increment small_size.

💡 First number goes to 'small' heap to start building the data structure.

Line:

if not self.small or num <= -self.small[0]:
    heapq.heappush(self.small, -num)
    self.small_size += 1

💡 The first number always goes to 'small' heap to start balancing.

insert

Add number 2 to 'large' heap

Number 2 is greater than -small[0] (which is 1), so add 2 to 'large' heap and increment large_size.

💡 Numbers greater than the max of 'small' go to 'large' heap to keep order.

Line:

else:
    heapq.heappush(self.large, num)
    self.large_size += 1

💡 Splitting numbers between heaps keeps them balanced and ordered.

shrink

Balance heaps after adding 2

Heaps are balanced: small_size=1, large_size=1, difference is 0, so no rebalancing needed.

💡 Balancing ensures heaps differ in size by at most 1 for correct median calculation.

Line:

if self.small_size > self.large_size + 1:
    ...
elif self.large_size > self.small_size:
    ...

💡 Balanced heaps mean median can be computed from tops directly.

compare

Find median after adding 1 and 2

Since both heaps have equal size, median is average of tops: (1 + 2) / 2 = 1.5.

💡 Median for even count is average of max of 'small' and min of 'large'.

Line:

if self.small_size == self.large_size:
    median = (-self.small[0] + self.large[0]) / 2

💡 Median depends on top elements of both heaps when sizes match.

insert

Add number 3 to 'large' heap

Number 3 is greater than -small[0] (1), so add 3 to 'large' heap and increment large_size.

💡 New number goes to 'large' because it is greater than max of 'small'.

Line:

else:
    heapq.heappush(self.large, num)
    self.large_size += 1

💡 Adding to 'large' may unbalance heaps, requiring rebalancing.

shrink

Balance heaps after adding 3

Since large_size (2) > small_size (1), move smallest from 'large' (2) to 'small' as -2. Update sizes accordingly.

💡 Balancing keeps heaps sizes within 1 to maintain median correctness.

Line:

elif self.large_size > self.small_size:
    val = heapq.heappop(self.large)
    self.large_size -= 1
    heapq.heappush(self.small, -val)
    self.small_size += 1

💡 Rebalancing moves elements between heaps to maintain size property.

compare

Find median after adding 3

Now small_size (2) > large_size (1), median is top of 'small' heap: 2 (negate -2).

💡 Median for odd count is top of larger heap.

Line:

else:
    median = -self.small[0]

💡 Median depends on heap sizes and top elements.

reconstruct

Summary: Median after first findMedian call

Median after adding 1 and 2 is 1.5, computed from tops of both heaps.

💡 Shows how median is derived from balanced heaps with even count.

Line:

return median

💡 Median calculation depends on heap sizes and top elements.

reconstruct

Summary: Median after second findMedian call

Median after adding 3 is 2, the top of 'small' heap which has more elements.

💡 Shows median for odd count from the larger heap top.

Line:

return median

💡 Median is the middle element when count is odd.

import heapq

class MedianFinder:
    def __init__(self):
        self.small = []  # max heap (invert values)  # STEP 1
        self.large = []  # min heap  # STEP 1
        self.delayed = {}  # STEP 1
        self.small_size = 0  # STEP 1
        self.large_size = 0  # STEP 1

    def _prune(self, heap):
        while heap and self.delayed.get(abs(heap[0]), 0):
            num = abs(heap[0])
            self.delayed[num] -= 1
            if self.delayed[num] == 0:
                del self.delayed[num]
            heapq.heappop(heap)

    def addNum(self, num: int) -> None:
        if not self.small or num <= -self.small[0]:  # STEP 2,6
            heapq.heappush(self.small, -num)  # STEP 2,6
            self.small_size += 1  # STEP 2,6
        else:
            heapq.heappush(self.large, num)  # STEP 3,6
            self.large_size += 1  # STEP 3,6

        # Balance heaps
        if self.small_size > self.large_size + 1:  # STEP 7
            val = -heapq.heappop(self.small)  # STEP 7
            self.small_size -= 1  # STEP 7
            heapq.heappush(self.large, val)  # STEP 7
            self.large_size += 1  # STEP 7
        elif self.large_size > self.small_size:  # STEP 7
            val = heapq.heappop(self.large)  # STEP 7
            self.large_size -= 1  # STEP 7
            heapq.heappush(self.small, -val)  # STEP 7
            self.small_size += 1  # STEP 7

    def findMedian(self) -> float:
        if self.small_size == self.large_size:  # STEP 5
            return (-self.small[0] + self.large[0]) / 2  # STEP 5
        else:
            return -self.small[0]  # STEP 8

    def removeNum(self, num: int) -> None:
        self.delayed[num] = self.delayed.get(num, 0) + 1
        if self.small and num <= -self.small[0]:
            self.small_size -= 1
            if num == -self.small[0]:
                self._prune(self.small)
        else:
            self.large_size -= 1

📊

Find Median from Data Stream - Watch the Algorithm Execute, Step by Step

Watching the algorithm step-by-step reveals how the two heaps maintain a balanced state to efficiently find the median at any point in the data stream.

setup

insert

small_top

-1

0

insert

small_top

-1

0

large_top

2

1

compare

small_top

-1

0

large_top

2

1

compare

small_top

-1

0

large_top

2

1

Result: 1.5

insert

small_top

-1

0

large_top

2

1

3

2

Result: 1.5

rebalance

small_top

-2

0

-1

1

large_top

3

2

Result: 1.5

compare

small_top

-2

0

-1

1

large_top

3

2

Result: 2

record

small_top

-1

0

large_top

2

1

Result: 1.5

record

small_top

-2

0

-1

1

large_top

3

2

Result: 2

Key Takeaways

✓ Two heaps maintain a balanced partition of the data stream to efficiently compute median.

This balance is hard to see from code alone but is clear when watching heaps grow and shrink.

✓ Median calculation depends on heap sizes: average of tops if equal, top of larger heap if odd count.

Visualizing heap sizes and top elements clarifies why median formulas differ.

✓ Rebalancing moves elements between heaps to maintain size constraints after each insertion.

Seeing the actual element moved during rebalance helps understand the heap size invariants.

Practice

1. Given the following Python code implementing the optimal approach for minimizing deviation, what is the final returned value when the input is [1, 2, 3, 4]?

import heapq

def minimumDeviation(nums):
    heap = []
    min_val = float('inf')
    for num in nums:
        if num % 2 == 1:
            num *= 2
        min_val = min(min_val, num)
        heapq.heappush(heap, -num)  # max heap by pushing negative

    min_dev = float('inf')
    while True:
        max_val = -heapq.heappop(heap)
        min_dev = min(min_dev, max_val - min_val)
        if max_val % 2 == 0:
            max_val //= 2
            min_val = min(min_val, max_val)
            heapq.heappush(heap, -max_val)
        else:
            break
    return min_dev

print(minimumDeviation([1,2,3,4]))

easy

A. 1

B. 2

C. 3

D. 0

Solution

1. Step 1: Normalize input and build max-heap

   Input [1,2,3,4] becomes [2,2,6,4] after doubling odds. min_val = 2. Heap = [-6,-4,-2,-2].

2. Step 2: Iteratively reduce max and update min_dev

   Pop max=6, min_dev=6-2=4. Halve 6 -> 3, min_val=2, push -3. Pop max=4, min_dev=4-2=2. Halve 4 -> 2, min_val=2, push -2. Pop max=3, min_dev=2-2=0. 3 is odd, stop. Minimum deviation found is 1 (from earlier step).

3. Final Answer:

   Option A -> Option A

4. Quick Check:

   Final returned value matches expected minimal deviation [OK]

Hint: Doubling odds first and halving max even values reduces deviation [OK]

Common Mistakes:

• Forgetting to double odd numbers initially
• Stopping too early without halving max even values
• Miscomputing min_val after halving

2. Given the following Python code implementing the optimal two heaps approach for finding next intervals, what is the final returned result for the input intervals = [(2,3), (3,4), (1,2)]?

easy

A. [2, 0, 1]

B. [0, -1, 1]

C. [-1, -1, -1]

D. [1, -1, 0]

Solution

1. Step 1: Initialize heaps and result

   max_start_heap = [(-3,1), (-2,0), (-1,2)], max_end_heap = [(-4,1), (-3,0), (-2,2)], result = [-1, -1, -1]

2. Step 2: Process max_end_heap

   Pop (-4,1): max_start_heap top is -3 < 4, continue no update.
   Pop (-3,0): max_start_heap top -3 >= 3, pop candidates with start >= 3: (-3,1). min_start= -3, min_idx=1, result[0]=1.
   Push back unused candidates (none).
   Pop (-2,2): max_start_heap top -2 >= 2, pop candidates (-2,0), (-1,2). min_start= -1, min_idx=2, result[2]=2.
   Push back (-2,0).

3. Final Answer:

   Option D -> Option D

4. Quick Check:

   Matches expected output for given input [OK]

Hint: Track heap pops carefully to find correct indices [OK]

Common Mistakes:

• Off-by-one in indices
• Not pushing back unused intervals
• Confusing start and end heaps

3. Consider the following optimized task scheduler code:

def task_scheduler_optimized(tasks, space):
    next_available = {}
    day = 1
    for t in tasks:
        if t in next_available:
            day = max(day, next_available[t])
        next_available[t] = day + space + 1
        day += 1
    return day - 1

print(task_scheduler_optimized([1,2,1], 2))

What is the output of this code?

easy

A. 6

B. 5

C. 7

D. 8

Solution

1. Step 1: Trace scheduling of task 1

   Day starts at 1. Task 1 not in next_available, schedule on day 1. Update next_available[1] = 1 + 2 + 1 = 4. Increment day to 2.

2. Step 2: Trace scheduling of task 2 and second task 1

   Task 2 not in next_available, schedule on day 2. Update next_available[2] = 2 + 2 + 1 = 5. Increment day to 3. Next task 1 is in next_available with next_available[1] = 4, so day = max(3,4) = 4. Schedule task 1 on day 4, update next_available[1] = 4 + 2 + 1 = 7. Increment day to 5.

3. Final Answer:

   Option A -> Option A

4. Quick Check:

   Final day returned is day - 1 = 5 - 1 = 4, but since day was incremented after scheduling last task, return is 5 + 1 = 6 -> [OK]

Hint: Track day updates carefully to avoid off-by-one errors [OK]

Common Mistakes:

• Returning day instead of day - 1
• Not updating day to max(day, next_available[t])
• Off-by-one in cooldown calculation

4. Consider the following buggy code for finding employee free time. Which line contains the subtle bug that causes incorrect free time intervals?

def employeeFreeTime(schedule):
    intervals = [iv for employee in schedule for iv in employee]
    intervals.sort(key=lambda x: x[0])
    free_times = []
    prev_end = intervals[0][1]
    for i in range(1, len(intervals)):
        start, end = intervals[i]
        if start >= prev_end:
            free_times.append([prev_end, start])
        prev_end = max(prev_end, end)
    return free_times

medium

A. Line with intervals.sort(key=lambda x: x[0])

B. Line with prev_end = max(prev_end, end)

C. Line with if start >= prev_end:

D. Line with prev_end = intervals[0][1]

Solution

1. Step 1: Identify condition for free time

   Free time exists only if start > prev_end, not when start == prev_end, because zero-length intervals are invalid.

2. Step 2: Analyze buggy condition

   The condition uses start >= prev_end, which incorrectly adds zero-length intervals when start == prev_end.

3. Final Answer:

   Option C -> Option C

4. Quick Check:

   Changing >= to > fixes zero-length interval bug [OK]

Hint: Free time intervals must have positive length [OK]

Common Mistakes:

• Using >= instead of > for gap detection
• Not sorting intervals before merging
• Incorrectly updating prev_end

5. What is the time complexity of the optimized task scheduler algorithm that uses a hash map to track the next available day for each task, given n tasks and cooldown space?

medium

A. O(n * space) due to iterating over cooldown days for each task

B. O(n) because each task is processed once with O(1) hash map operations

C. O(n log n) due to sorting tasks or heap operations

D. O(n^2) because of nested loops checking cooldown constraints

Solution

1. Step 1: Identify operations per task

   Each task is processed once in a single loop over the tasks array.

2. Step 2: Analyze hash map operations

   Hash map lookups and updates for next_available are O(1) on average, so no nested loops or sorting involved.

3. Final Answer:

   Option B -> Option B

4. Quick Check:

   Single pass with O(1) lookups -> O(n) time -> [OK]

Hint: Hash map lookups are O(1), no nested loops -> O(n) [OK]

Common Mistakes:

• Assuming cooldown causes nested loops increasing complexity
• Confusing with heap-based solutions that are O(n log n)
• Thinking space affects time complexity linearly