Mediana da un flusso di dati e merge k-way
Mantenga due heap (un max-heap per la metà inferiore e un min-heap per quella superiore) per aggiornare la mediana in O(log n) e fonda k liste ordinate usando un heap
Mediana da un flusso di dati e merge k-way è una lezione Coding Interview Prep gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento Coding Interview Prep, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso Coding Interview Prep include 4 lezioni in totale.
Problema della mediana di uno stream di dati
Find Median from Data Stream (LeetCode #295) richiede di supportare in modo efficiente due operazioni: addNum(num) per aggiungere un numero e findMedian() per restituire la mediana corrente. In una lista di lunghezza pari, la mediana è la media dei due valori centrali. Una lista ordinata con un approccio brute-force consente un inserimento in O(n) e il recupero della mediana in O(1). La soluzione ottimale usa due heap per ottenere inserimenti in O(log n) e il recupero della mediana in 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')Implementazione di MedianFinder con due heap
Mantenga un max-heap per la metà inferiore e un min-heap per la metà superiore. Si assicuri sempre che il max-heap abbia la stessa dimensione del min-heap oppure un elemento in più. Quando aggiunge un numero, lo inserisca nel max-heap, quindi bilanci spostando la radice del max-heap nel min-heap se supera il minimo del min-heap, e riequilibri le dimensioni se necessario.
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.0Analizzare i passaggi di MedianFinder
Comprendere perché l'invariante dei due heap viene mantenuto è fondamentale per spiegare la soluzione durante un colloquio tecnico. Analizziamo passo per passo l'inserimento di [5, 15, 1, 3]. Dopo ogni inserimento, bilanci in modo che il max-heap inferiore contenga la metà più piccola. L'invariante garantisce sempre max(lo) <= min(hi), rendendo la mediana immediatamente accessibile in cima a uno o a entrambi gli heap.
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 di una finestra scorrevole
La Sliding Window Median (LeetCode #480) è una variante più complessa: trovare la mediana di ogni finestra di dimensione k mentre scorre lungo l'array. L'approccio con due heap viene esteso con un set di cancellazione lazy per gestire gli elementi che escono dalla finestra. Quando un elemento lascia la finestra, lo contrassegni nel set delle cancellazioni; quando raggiunge la cima di uno dei due heap, lo elimini.
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]Unione k-way: il problema
Merge K Sorted Lists (LeetCode #23) è un problema fondamentale, con applicazioni nell'ordinamento esterno, nelle operazioni di merge dei database e nei sistemi distribuiti. Dati k elenchi concatenati ordinati, per un totale di n nodi, li unisca in un unico elenco ordinato. L'approccio ingenuo (unire due elenchi alla volta) ha complessità O(kn), oppure O(n log k) con divide-and-conquer. L'approccio con heap elabora ogni nodo esattamente una volta, con un lavoro O(log k) per nodo: O(n log k) in totale.
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')Unione k-way con un min-heap
Inizializzi l'heap con il primo nodo di ogni elenco. A ogni passaggio, rimuova il minimo, lo aggiunga al risultato e inserisca il nodo successivo dello stesso elenco, se presente. L'heap contiene sempre al massimo k elementi, uno per la testa di ogni elenco attivo. Poiché elaboriamo un totale di n nodi con operazioni sull'heap da O(log k) ciascuna, il tempo totale è O(n log k) e lo spazio è O(k) per l'heap.
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]Intervallo più piccolo che copre k elenchi
Smallest Range (LeetCode #632) trova l'intervallo più piccolo [lo, hi] tale che almeno un elemento di ciascuno dei k elenchi ordinati si trovi nell'intervallo. Utilizzi un min-heap inizializzato con il primo elemento di ogni elenco e tenga traccia del massimo corrente. Riduca l'intervallo facendo avanzare sempre l'elenco con il minimo corrente. Si fermi quando uno degli elenchi è esaurito.
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-esimo elemento più piccolo in una matrice
Kth Smallest Element in a Sorted Matrix (LeetCode #378): una matrice n×n in cui ogni riga e ogni colonna sono ordinate. Trovi il k-esimo elemento più piccolo. Consideri ogni riga come un elenco ordinato e utilizzi un'unione k-way con un heap. In alternativa, esegua una ricerca binaria sull'intervallo dei valori. L'approccio con heap ha complessità O(k log n), quindi è efficiente quando k è piccolo; la ricerca binaria ha complessità O(n log(max-min)) e gestisce meglio valori di k elevati.
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)) # 13Due heap per le statistiche incrementali
Lo schema dei due heap si generalizza oltre la mediana. Può utilizzarlo per mantenere un quantile incrementale (ad esempio il 25º percentile): dimensioni l'heap inferiore in modo che contenga p*n elementi e quello superiore in modo che ne contenga (1-p)*n. Ogni volta che aggiunge un elemento, riequilibri come in precedenza. Questo schema compare nei problemi di statistica in streaming, quando sono necessari contemporaneamente inserimenti efficienti e query sui quantili.
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)Trovare i k punti più vicini all'origine
K Closest Points to Origin (LeetCode #973) usa un max-heap di dimensione k. Inserisca la distanza al quadrato di ogni punto, per evitare di calcolare la radice quadrata. Quando l'heap supera k elementi, rimuova quello più lontano. I k punti rimanenti sono i k più vicini. La complessità è O(n log k). Un'alternativa consiste nell'usare quickselect per ottenere O(n) in media, ma la soluzione con heap è più semplice da implementare correttamente e da spiegare durante un colloquio.
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}')Due heap: analisi di tempo e spazio
L'approccio con due heap per la mediana raggiunge O(log n) per addNum e O(1) per findMedian. Lo spazio è O(n) per memorizzare tutti gli elementi. L'unione k-way richiede O(n log k) di tempo e O(k) di spazio per l'heap. Si tratta di complessità quasi ottimali: è possibile dimostrare un limite inferiore Omega(n log k) basato sui confronti per l'unione k-way, mostrando che la soluzione con heap è asintoticamente ottimale. Durante i colloqui tecnici, esponga sempre con chiarezza queste complessità.
# 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)')Verifica rapida
Verifichi la Sua comprensione dei concetti di Data Structures & Algorithms — Coding Interview Prep trattati in questa lezione.
Riepilogo della lezione
In questa lezione ha imparato a usare MedianFinder con due heap, ottenendo inserimenti in O(log n) e il recupero della mediana in O(1); l'unione k-way con un min-heap in O(n log k) di tempo e O(k) di spazio; e alcune estensioni, tra cui la mediana di una finestra scorrevole, l'intervallo più piccolo e i k punti più vicini. Nella prossima lezione esploreremo le rappresentazioni dei grafi e la configurazione delle visite.
Domande Frequenti
La lezione «Mediana da un flusso di dati e merge k-way» è gratuita?
Sì — il testo completo di «Mediana da un flusso di dati e merge k-way» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso Coding Interview Prep, passa a CoddyKit PRO. Il corso Coding Interview Prep include 4 lezioni in totale.
Cosa imparerò in «Mediana da un flusso di dati e merge k-way»?
Mantenga due heap (un max-heap per la metà inferiore e un min-heap per quella superiore) per aggiornare la mediana in O(log n) e fonda k liste ordinate usando un heap Eserciti Coding Interview Prep con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare Coding Interview Prep?
Non è richiesta alcuna esperienza precedente. Coding Interview Prep su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.
Quanto tempo richiede la lezione «Mediana da un flusso di dati e merge k-way»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione Coding Interview Prep?
Sì. Ogni lezione Coding Interview Prep include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Proprietà degli heap e rappresentazione tramite array
- Heapify, push e pop da zero
- heapq di Python e trucchi per il max-heap
- Mediana da un flusso di dati e merge k-way