Mediana ze strumienia danych i scalanie k-way
Utrzymają Państwo dwa kopce — maksymalny dla mniejszej połowy i minimalny dla większej — aby aktualizować medianę w O(log n), a także scalą k posortowanych list za pomocą kopca.
Mediana ze strumienia danych i scalanie k-way to bezpłatna lekcja Coding Interview Prep na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej Coding Interview Prep, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs Coding Interview Prep zawiera 4 lekcji w sumie.
Problem mediany ze strumienia danych
Find Median from Data Stream (LeetCode #295) wymaga wydajnej obsługi dwóch operacji: addNum(num) do dodawania liczby oraz findMedian() do zwracania bieżącej mediany. Mediana listy o parzystej długości jest średnią dwóch środkowych wartości. Posortowana lista aktualizowana metodą naiwną zapewnia wstawianie w czasie O(n) i odczyt mediany w czasie O(1). Optymalne rozwiązanie wykorzystuje dwa kopce, zapewniając wstawianie w czasie O(log n) i odczyt mediany w czasie O(1).
import heapq
# Strategy: maintain two halves of the data
# max_heap: lower half (stores negated values for max behavior)
# min_heap: upper half
# Invariant: len(max_heap) == len(min_heap) or len(max_heap) == len(min_heap) + 1
# Invariant: max(max_heap) <= min(min_heap)
# Median:
# odd count: max_heap[0] (top of lower half)
# even count: average of tops of both halves
print('Two-heap strategy for O(log n) insert, O(1) median')Implementacja MedianFinder z użyciem dwóch kopców
Należy utrzymywać kopiec maksymalny dla dolnej połowy oraz kopiec minimum dla górnej połowy. Zawsze trzeba zapewnić, aby kopiec maksymalny miał tyle samo elementów co kopiec minimum lub o jeden element więcej. Podczas dodawania liczby należy dodać ją do kopca maksymalnego, a następnie przywrócić równowagę, przenosząc jego wierzchołek do kopca minimum, jeśli jest większy od minimum w tym kopcu, oraz w razie potrzeby wyrównać rozmiary kopców.
import heapq
class MedianFinder:
def __init__(self):
self.lo = [] # max-heap (negated) for lower half
self.hi = [] # min-heap for upper half
def addNum(self, num):
heapq.heappush(self.lo, -num) # push to lower half
# Ensure max of lower <= min of upper
if self.hi and -self.lo[0] > self.hi[0]:
heapq.heappush(self.hi, -heapq.heappop(self.lo))
# Balance sizes: lo can have at most 1 more than hi
if len(self.lo) > len(self.hi) + 1:
heapq.heappush(self.hi, -heapq.heappop(self.lo))
elif len(self.hi) > len(self.lo):
heapq.heappush(self.lo, -heapq.heappop(self.hi))
def findMedian(self):
if len(self.lo) > len(self.hi):
return -self.lo[0] # odd count: top of lower half
return (-self.lo[0] + self.hi[0]) / 2
mf = MedianFinder()
for n in [1, 2, 3, 4, 5]: mf.addNum(n)
print(mf.findMedian()) # 3.0Prześledzenie kroków MedianFinder
Zrozumienie, dlaczego niezmiennik dwóch kopców jest zachowany, ma kluczowe znaczenie podczas wyjaśniania rozwiązania na rozmowie kwalifikacyjnej. Prześledźmy krok po kroku dodawanie elementów [5, 15, 1, 3]. Po każdym wstawieniu należy przywrócić równowagę, tak aby dolny kopiec maksymalny zawierał mniejszą połowę elementów. Niezmiennik gwarantuje, że warunek max(lo) <= min(hi) jest zawsze spełniony, dzięki czemu mediana jest łatwo dostępna na wierzchołku jednego lub obu kopców.
import heapq
# Manual trace for [5, 15, 1, 3]:
# add 5: lo=[-5] hi=[] median=5
# add 15: lo=[-5] hi=[15] median=(5+15)/2=10
# add 1: lo=[-5,-1] hi=[15] median=5
# add 3: lo=[-5,-3,-1] hi=[15] -- lo too big
# -> lo=[-5,-3] hi=[1,15] -- wait, wrong direction
# Actually:
# add 1: push to lo -> lo=[-5,-1], then 1>lo? No, -lo[0]=5>15? No
# lo has 2, hi has 1: balance -> move lo top to hi
# lo=[-1], hi=[5,15]
# Median = (-lo[0] + hi[0])/2 = (1+5)/2 = 3
mf2 = MedianFinder()
for n, expected in [(5, 5.0), (15, 10.0), (1, 5.0), (3, 4.0)]:
mf2.addNum(n)
print(f'After adding {n}: median={mf2.findMedian()} (expected ~{expected})')Mediana w przesuwającym się oknie
Sliding Window Median (LeetCode #480) to trudniejszy wariant problemu: należy znaleźć medianę każdego okna o rozmiarze k przesuwanego po tablicy. Podejście z dwoma kopcami można rozszerzyć o zbiór leniwego usuwania, aby obsługiwać elementy wypadające z okna. Gdy element opuszcza okno, należy oznaczyć go w zbiorze usuwania, a gdy znajdzie się na wierzchołku któregoś z kopców, usunąć go.
import heapq
def median_sliding_window(nums, k):
lo = [] # max-heap (negated)
hi = [] # min-heap
removed = {}
result = []
def balance():
# Move valid tops to correct side
while lo and removed.get(-lo[0], 0) > 0:
removed[-lo[0]] -= 1; heapq.heappop(lo)
while hi and removed.get(hi[0], 0) > 0:
removed[hi[0]] -= 1; heapq.heappop(hi)
for i, num in enumerate(nums):
heapq.heappush(lo, -num)
heapq.heappush(hi, -heapq.heappop(lo))
if len(hi) > len(lo): heapq.heappush(lo, -heapq.heappop(hi))
if i >= k:
out = nums[i - k]
removed[out] = removed.get(out, 0) + 1
balance()
if len(lo) > len(hi): heapq.heappush(hi, -heapq.heappop(lo))
if i >= k - 1:
if len(lo) > len(hi): result.append(float(-lo[0]))
else: result.append((-lo[0] + hi[0]) / 2.0)
return result
print(median_sliding_window([1,3,-1,-3,5,3,6,7], 3)) # [1,-1,-1,3,5,6]Scalanie k-way: problem
Merge K Sorted Lists (LeetCode #23) to fundamentalny problem znajdujący zastosowanie w sortowaniu zewnętrznym, scalaniu baz danych i systemach rozproszonych. Mając k posortowanych list jednokierunkowych zawierających łącznie n węzłów, należy scalić je w jedną posortowaną listę. Naiwne podejście, polegające na scalaniu dwóch list naraz, ma złożoność O(kn), a podejście dziel i zwyciężaj — O(n log k). Podejście oparte na kopcu przetwarza każdy węzeł dokładnie raz, wykonując O(log k) pracy na węzeł, co daje łącznie O(n log k).
class ListNode:
def __init__(self, val=0, next=None):
self.val = val
self.next = next
# Build a linked list from a Python list
def build_list(arr):
dummy = ListNode(0)
curr = dummy
for val in arr:
curr.next = ListNode(val)
curr = curr.next
return dummy.next
# Convert linked list to Python list for printing
def to_list(head):
result = []
while head:
result.append(head.val)
head = head.next
return result
print('K-way merge: O(n log k) using a min-heap of k heads')Scalanie k-way z kopcem minimum
Na początku należy umieścić w kopcu pierwszy węzeł każdej listy. W każdym kroku należy pobrać minimum, dodać je do wyniku i umieścić w kopcu następny węzeł z tej listy, jeśli istnieje. Kopiec ma zawsze co najwyżej k elementów — po jednym początku dla każdej aktywnej listy. Ponieważ łącznie przetwarzamy n węzłów, wykonując dla każdego O(log k) operacji na kopcu, całkowity czas działania wynosi O(n log k), a pamięć zajmowana przez kopiec — O(k).
import heapq
def merge_k_lists(lists):
dummy = ListNode(0)
curr = dummy
heap = []
for i, node in enumerate(lists):
if node:
heapq.heappush(heap, (node.val, i, node))
while heap:
val, i, node = heapq.heappop(heap)
curr.next = node
curr = curr.next
if node.next:
heapq.heappush(heap, (node.next.val, i, node.next))
return dummy.next
lists = [
build_list([1, 4, 5]),
build_list([1, 3, 4]),
build_list([2, 6])
]
result = merge_k_lists(lists)
print(to_list(result)) # [1, 1, 2, 3, 4, 4, 5, 6]Najmniejszy przedział obejmujący K list
Smallest Range (LeetCode #632) znajduje najmniejszy przedział [lo, hi], w którym znajduje się co najmniej jeden element z każdej z k posortowanych list. Należy użyć kopca minimum zainicjalizowanego pierwszym elementem każdej listy i śledzić bieżące maksimum. Przedział należy zawężać, zawsze przechodząc do następnego elementu listy zawierającej bieżące minimum. Należy zakończyć, gdy któraś z list zostanie wyczerpana.
import heapq
def smallest_range(nums):
heap = []
current_max = float('-inf')
for i, lst in enumerate(nums):
heapq.heappush(heap, (lst[0], i, 0))
current_max = max(current_max, lst[0])
best = [float('-inf'), float('inf')]
while heap:
current_min, list_idx, elem_idx = heapq.heappop(heap)
if current_max - current_min < best[1] - best[0]:
best = [current_min, current_max]
if elem_idx + 1 >= len(nums[list_idx]):
break # one list exhausted
next_val = nums[list_idx][elem_idx + 1]
heapq.heappush(heap, (next_val, list_idx, elem_idx + 1))
current_max = max(current_max, next_val)
return best
print(smallest_range([[4,10,15,24,26],[0,9,12,20],[5,18,22,30]]))
# [20, 24]K-ty najmniejszy element w macierzy
Kth Smallest Element in a Sorted Matrix (LeetCode #378): mamy macierz n×n, w której każdy wiersz i każda kolumna są posortowane. Należy znaleźć k-ty najmniejszy element. Każdy wiersz można potraktować jako posortowaną listę i użyć scalania k-way z kopcem. Alternatywnie można zastosować wyszukiwanie binarne w zakresie wartości. Podejście z kopcem ma złożoność O(k log n), więc jest wydajne, gdy k jest małe; wyszukiwanie binarne ma złożoność O(n log(max-min)) i lepiej sprawdza się przy dużych wartościach k.
import heapq
def kth_smallest_matrix(matrix, k):
n = len(matrix)
heap = [(matrix[0][0], 0, 0)]
count = 0
visited = {(0, 0)}
while heap:
val, r, c = heapq.heappop(heap)
count += 1
if count == k:
return val
# Push right neighbor
if c + 1 < n and (r, c+1) not in visited:
heapq.heappush(heap, (matrix[r][c+1], r, c+1))
visited.add((r, c+1))
# Push bottom neighbor
if r + 1 < n and (r+1, c) not in visited:
heapq.heappush(heap, (matrix[r+1][c], r+1, c))
visited.add((r+1, c))
return -1
matrix = [[1,5,9],[10,11,13],[12,13,15]]
print(kth_smallest_matrix(matrix, 8)) # 13Dwa kopce dla statystyk kroczących
Wzorzec dwóch kopców można uogólnić poza medianę. Można go użyć do utrzymywania kwantyla kroczącego (np. 25. percentyla): rozmiar dolnego kopca należy ustalić tak, aby przechowywał p*n elementów, a górnego tak, aby przechowywał (1-p)*n elementów. Po każdym dodaniu elementu należy przywrócić równowagę tak jak wcześniej. Wzorzec ten pojawia się w problemach dotyczących statystyk strumieniowych, w których jednocześnie potrzebne są wydajne wstawianie i zapytania o kwantyl.
import heapq
# Generalised two-heap for arbitrary quantile p
# lo contains floor(p * count) elements
# hi contains the remaining elements
class QuantileFinder:
def __init__(self, p):
self.p = p # quantile (e.g., 0.5 for median)
self.lo = [] # max-heap
self.hi = [] # min-heap
self.count = 0
def add(self, num):
self.count += 1
heapq.heappush(self.lo, -num)
heapq.heappush(self.hi, -heapq.heappop(self.lo))
# Target: lo should have floor(p * count) elements
target_lo = int(self.p * self.count)
while len(self.lo) < target_lo:
heapq.heappush(self.lo, -heapq.heappop(self.hi))
while len(self.lo) > target_lo:
heapq.heappush(self.hi, -heapq.heappop(self.lo))
def quantile(self):
return -self.lo[0] if self.lo else self.hi[0]
qf = QuantileFinder(0.5) # median
for n in [1, 2, 3, 4, 5, 6]: qf.add(n)
print(qf.quantile()) # 3 (median of 1-6)K najbliższych punktów względem początku układu
K Closest Points to Origin (LeetCode #973) wykorzystuje kopiec maksymalny o rozmiarze k. Należy dodać do niego kwadrat odległości każdego punktu, aby uniknąć obliczania pierwiastka. Gdy rozmiar kopca przekroczy k, należy usunąć punkt najbardziej oddalony. Pozostałe k punktów to k punktów najbliższych. Złożoność wynosi O(n log k). Alternatywnie można użyć algorytmu quickselect o średniej złożoności O(n), ale rozwiązanie z kopcem jest prostsze do poprawnego zaimplementowania i wyjaśnienia podczas rozmowy kwalifikacyjnej.
import heapq
def k_closest(points, k):
heap = [] # max-heap via negation
for x, y in points:
dist_sq = x*x + y*y
heapq.heappush(heap, (-dist_sq, x, y))
if len(heap) > k:
heapq.heappop(heap) # remove farthest
return [[x, y] for _, x, y in heap]
points = [[1,3], [-2,2], [5,8], [0,1], [-1,-1]]
print(k_closest(points, 2))
# Two closest to origin: [0,1] (dist=1) and [-1,-1] (dist=2)
# Verify by distances:
for x, y in points:
print(f'({x},{y}): dist^2 = {x*x+y*y}')Analiza czasu i pamięci dla dwóch kopców
Podejście z dwoma kopcami dla mediany zapewnia O(log n) dla każdej operacji addNum oraz O(1) dla findMedian. Zajętość pamięci wynosi O(n), ponieważ trzeba przechowywać wszystkie elementy. Scalanie k-way ma złożoność czasową O(n log k) i zajmuje O(k) pamięci na kopiec. Są to wartości bliskie optymalnym: można dowieść dolnego ograniczenia Omega(n log k) dla scalania k-way opartego na porównaniach, co pokazuje, że rozwiązanie z kopcem jest optymalne asymptotycznie. Podczas rozmów kwalifikacyjnych należy zawsze jasno podawać te złożoności.
# Complexity summary for heap applications:
# Problem | Time per op | Space
# ----------------------|--------------|------
# MedianFinder.addNum | O(log n) | O(n)
# MedianFinder.find | O(1) | -
# Merge k sorted lists | O(n log k) | O(k)
# Kth smallest matrix | O(k log n) | O(n)
# K closest points | O(n log k) | O(k)
# Task scheduler | O(n log 26) | O(26)
# Kth largest stream | O(log k) | O(k)
# Sliding window median | O(n log k) | O(k)
print('Heap problems: identify k (heap size) vs n (input size)')Szybkie sprawdzenie
Proszę sprawdzić swoje zrozumienie zagadnień Data Structures & Algorithms — Coding Interview Prep omówionych w tej lekcji.
Podsumowanie lekcji
W tej lekcji poznali Państwo: MedianFinder z dwoma kopcami, zapewniający wstawianie w czasie O(log n) i odczyt mediany w czasie O(1), scalanie k-way z użyciem kopca minimum w czasie O(n log k) i przy zużyciu O(k) pamięci, a także rozszerzenia obejmujące medianę w przesuwającym się oknie, najmniejszy przedział i k najbliższych punktów. W następnej części poznamy reprezentacje grafów oraz przygotowanie do ich przeszukiwania.
Często zadawane pytania
Czy lekcja „Mediana ze strumienia danych i scalanie k-way” jest bezpłatna?
Tak — pełny tekst „Mediana ze strumienia danych i scalanie k-way” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu Coding Interview Prep, przejdź na CoddyKit PRO. Kurs Coding Interview Prep zawiera 4 lekcji w sumie.
Co nauczysz się w „Mediana ze strumienia danych i scalanie k-way”?
Utrzymają Państwo dwa kopce — maksymalny dla mniejszej połowy i minimalny dla większej — aby aktualizować medianę w O(log n), a także scalą k posortowanych list za pomocą kopca. Ćwiczysz Coding Interview Prep z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć Coding Interview Prep?
Nie wymagamy żadnego doświadczenia. Coding Interview Prep w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.
Ile czasu zajmuje lekcja „Mediana ze strumienia danych i scalanie k-way”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji Coding Interview Prep?
Tak. Każda lekcja Coding Interview Prep zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Własność kopca i reprezentacja tablicowa
- Heapify, push i pop od podstaw
- heapq w Pythonie i sztuczki z kopcem maksymalnym
- Mediana ze strumienia danych i scalanie k-way