Médiane d’un flux de données et fusion k-voies
Maintenez deux tas (un tas max pour la moitié inférieure et un tas min pour la moitié supérieure) afin de mettre à jour la médiane en O(log n), puis fusionnez k listes triées avec un tas.
Médiane d’un flux de données et fusion k-voies est une leçon DSA Interview Prep gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage DSA Interview Prep, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours DSA Interview Prep comprend 4 leçons au total.
Problème de la médiane d’un flux de données
Trouver la médiane d’un flux de données (LeetCode #295) demande de prendre en charge efficacement deux opérations : addNum(num) pour ajouter un nombre et findMedian() pour renvoyer la médiane actuelle. La médiane d’une liste de longueur paire est la moyenne des deux valeurs centrales. Une liste triée obtenue par force brute offre une insertion en O(n) et une médiane en O(1). La solution optimale utilise deux tas pour obtenir une insertion en O(log n) et une médiane en 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')Implémentation de MedianFinder avec deux tas
Conservez un tas max pour la moitié inférieure et un tas min pour la moitié supérieure. Assurez-vous toujours que le tas max contient autant d’éléments que le tas min, ou un élément de plus. Lors de l’ajout d’un nombre : insérez-le dans le tas max, puis rééquilibrez en déplaçant le sommet du tas max vers le tas min si ce sommet dépasse le minimum du tas min, et rééquilibrez les tailles si nécessaire.
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.0Suivre les étapes de MedianFinder
Comprendre pourquoi l’invariant des deux tas est maintenu est essentiel pour expliquer la solution en entretien. Suivons l’ajout de [5, 15, 1, 3] étape par étape. Après chaque insertion, rééquilibrez afin que le tas max inférieur contienne la moitié des plus petites valeurs. L’invariant garantit que max(lo) <= min(hi) est toujours vérifiée, ce qui rend la médiane immédiatement accessible au sommet de l’un des tas ou des deux.
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})')Médiane d’une fenêtre glissante
La médiane d’une fenêtre glissante (LeetCode #480) est une variante plus difficile : il faut trouver la médiane de chaque fenêtre de taille k lorsqu’elle se déplace dans le tableau. L’approche à deux tas est étendue avec un ensemble de suppression différée pour gérer les éléments qui sortent de la fenêtre. Lorsqu’un élément quitte la fenêtre, marquez-le dans l’ensemble des suppressions ; lorsqu’il atteint le sommet de l’un ou l’autre tas, supprimez-le.
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]Fusion à k voies : le problème
Fusionner k listes triées (LeetCode #23) est un problème fondamental, avec des applications dans le tri externe, les fusions de bases de données et les systèmes distribués. Étant donné k listes chaînées triées totalisant n nœuds, fusionnez-les en une seule liste triée. L’approche naïve, qui consiste à fusionner deux listes à la fois, est en O(kn), ou en O(n log k) avec la méthode diviser pour régner. L’approche par tas traite chaque nœud exactement une fois, avec un coût de O(log k) par nœud : O(n log k) au total.
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')Fusion à k voies avec un tas min
Initialisez le tas avec le premier nœud de chaque liste. À chaque étape, extrayez le minimum, ajoutez-le au résultat, puis insérez le nœud suivant de cette liste, s’il existe. Le tas contient toujours au plus k éléments — un élément en tête par liste active. Comme nous traitons au total n nœuds avec des opérations sur le tas en O(log k) chacune, le temps total est de O(n log k) et l’espace occupé par le tas est de 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]Plus petit intervalle couvrant k listes
Plus petit intervalle (LeetCode #632) recherche le plus petit intervalle [lo, hi] tel qu’au moins un élément de chacune des k listes triées se trouve dans cet intervalle. Utilisez un tas min initialisé avec le premier élément de chaque liste et suivez le maximum actuel. Réduisez l’intervalle en avançant toujours dans la liste contenant le minimum actuel. Arrêtez-vous lorsqu’une liste est épuisée.
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-ième plus petit élément dans une matrice
K-ième plus petit élément dans une matrice triée (LeetCode #378) : une matrice n×n dont chaque ligne et chaque colonne est triée. Trouvez le k-ième plus petit élément. Considérez chaque ligne comme une liste triée et utilisez une fusion à k voies avec un tas. Vous pouvez aussi effectuer une recherche binaire sur l’intervalle de valeurs. L’approche par tas est en O(k log n), ce qui est efficace lorsque k est petit ; la recherche binaire est en O(n log(max-min)), ce qui convient mieux aux grandes valeurs de 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)) # 13Deux tas pour les statistiques en continu
Le schéma des deux tas se généralise au-delà de la médiane. Vous pouvez l’utiliser pour maintenir un quantile évolutif (par exemple, le 25e centile) : dimensionnez le tas inférieur pour contenir p*n éléments et le tas supérieur pour contenir (1-p)*n éléments. Chaque fois qu’un élément est ajouté, rééquilibrez comme précédemment. Ce schéma apparaît dans les problèmes de statistiques en flux où vous avez simultanément besoin d’insertions efficaces et de requêtes de quantile.
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)Trouver les k points les plus proches de l’origine
Points les plus proches de l’origine (LeetCode #973) utilise un tas max de taille k. Insérez la distance au carré de chaque point, afin d’éviter de calculer la racine carrée. Lorsque le tas dépasse k éléments, extrayez le point le plus éloigné. Les k points restants sont les k plus proches. La complexité est de O(n log k). Une autre solution utilise la sélection rapide, en O(n) en moyenne, mais la solution avec un tas est plus simple à implémenter correctement et à expliquer pendant un entretien.
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}')Analyse du temps et de l’espace avec deux tas
L’approche à deux tas pour la médiane atteint O(log n) par addNum et O(1) par findMedian. L’espace occupé est de O(n) pour stocker tous les éléments. La fusion à k voies s’effectue en O(n log k) et utilise O(k) d’espace pour le tas. Ces résultats sont presque optimaux : vous pouvez démontrer une borne inférieure fondée sur les comparaisons de Ω(n log k) pour la fusion à k voies, ce qui montre que la solution avec un tas est optimale du point de vue asymptotique. Énoncez toujours clairement ces complexités en entretien.
# 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)')Vérification rapide
Testez votre compréhension des concepts de Structures de données et algorithmes — Préparation aux entretiens de programmation présentés dans cette leçon.
Récapitulatif de la leçon
Dans cette leçon, vous avez appris à utiliser MedianFinder avec deux tas, pour obtenir une insertion en O(log n) et une médiane en O(1), la fusion à k voies avec un tas min en O(n log k) et un espace de O(k), ainsi que plusieurs extensions, notamment la médiane d’une fenêtre glissante, le plus petit intervalle et les k points les plus proches. Ensuite, nous étudierons les représentations des graphes et la mise en place de leur parcours.
Questions Fréquemment Posées
La leçon « Médiane d’un flux de données et fusion k-voies » est-elle gratuite ?
Oui — le texte complet de « Médiane d’un flux de données et fusion k-voies » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours DSA Interview Prep, passe à CoddyKit PRO. Le cours DSA Interview Prep comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Médiane d’un flux de données et fusion k-voies » ?
Maintenez deux tas (un tas max pour la moitié inférieure et un tas min pour la moitié supérieure) afin de mettre à jour la médiane en O(log n), puis fusionnez k listes triées avec un tas. Tu pratiques DSA Interview Prep avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer DSA Interview Prep ?
Aucune expérience préalable n'est requise. DSA Interview Prep sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Médiane d’un flux de données et fusion k-voies » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon DSA Interview Prep ?
Oui. Chaque leçon DSA Interview Prep inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Propriété des tas et représentation par tableau
- Heapify, push et pop depuis zéro
- heapq de Python et astuces pour les tas max
- Médiane d’un flux de données et fusion k-voies