Mediana de un flujo de datos y combinación k-way
Mantenga dos heaps —un max-heap de la mitad menor y un min-heap de la mitad mayor— para actualizar la mediana en O(log n) y combine k listas ordenadas usando un heap.
Mediana de un flujo de datos y combinación k-way es una lección gratuita de Coding Interview Prep en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de Coding Interview Prep, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de Coding Interview Prep incluye 4 lecciones en total.
Problema de la mediana de un flujo de datos
Encontrar la mediana de un flujo de datos (LeetCode #295) le pide admitir dos operaciones de forma eficiente: addNum(num) para añadir un número y findMedian() para devolver la mediana actual. La mediana de una lista de longitud par es el promedio de sus dos valores centrales. Una lista ordenada mediante fuerza bruta ofrece una inserción O(n) y una mediana O(1). La solución óptima utiliza dos heaps para obtener una inserción O(log n) y una mediana 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')Implementación de MedianFinder con dos heaps
Mantenga un max-heap para la mitad inferior y un min-heap para la mitad superior. Asegúrese siempre de que el max-heap tenga el mismo tamaño que el min-heap o un elemento más. Al añadir un número: insértelo en el max-heap y, después, equilibre moviendo la raíz del max-heap al min-heap si la raíz supera el mínimo del min-heap; reajuste también los tamaños si es necesario.
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.0Recorrido de los pasos de MedianFinder
Comprender por qué se mantiene el invariante de los dos heaps es fundamental para explicar la solución en una entrevista. Recorramos paso a paso la inserción de [5, 15, 1, 3]. Después de cada inserción, equilibre los heaps para que el max-heap inferior contenga la mitad menor. El invariante garantiza que max(lo) <= min(hi) siempre se cumpla, lo que hace que la mediana sea fácilmente accesible en la raíz de uno o ambos heaps.
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 de una ventana deslizante
La mediana de una ventana deslizante (LeetCode #480) es una variante más difícil: encontrar la mediana de cada ventana de tamaño k mientras se desplaza por el array. El enfoque de dos heaps se amplía con un conjunto de eliminación diferida para gestionar los elementos que salen de la ventana. Cuando un elemento abandone la ventana, márquelo en el conjunto de eliminados y, cuando llegue a la raíz de cualquiera de los heaps, descártelo.
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]Fusión en k vías: el problema
Fusionar k listas ordenadas (LeetCode #23) es un problema fundamental con aplicaciones en la ordenación externa, la fusión de bases de datos y los sistemas distribuidos. Dadas k listas enlazadas ordenadas que contienen un total de n nodos, fusiónelas en una única lista ordenada. El enfoque ingenuo (fusionar de dos en dos) tiene una complejidad de O(kn), o de O(n log k) con divide y vencerás. El enfoque con heap procesa cada nodo exactamente una vez, con un trabajo de O(log k) por nodo: O(n log k) en 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')Fusión en k vías con un min-heap
Inicialice el heap con el primer nodo de cada lista. En cada paso, extraiga el mínimo, añádalo al resultado e inserte el siguiente nodo de esa lista, si existe. El heap siempre contiene como máximo k elementos: un encabezado por cada lista activa. Como procesamos un total de n nodos con operaciones de heap de O(log k) cada una, el tiempo total es O(n log k) y el espacio es O(k) para el 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]Rango más pequeño que cubre k listas
Rango más pequeño (LeetCode #632) encuentra el rango más pequeño [lo, hi] tal que al menos un elemento de cada una de las k listas ordenadas se encuentre dentro del rango. Utilice un min-heap inicializado con el primer elemento de cada lista y realice un seguimiento del máximo actual. Reduzca el rango avanzando siempre en la lista cuyo mínimo actual sea menor. Deténgase cuando se agote cualquiera de las listas.
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-ésimo elemento más pequeño en una matriz
K-ésimo elemento más pequeño en una matriz ordenada (LeetCode #378): una matriz n×n cuyas filas y columnas están ordenadas. Encuentre el k-ésimo elemento más pequeño. Trate cada fila como una lista ordenada y utilice la combinación en k vías con un heap. Como alternativa, realice una búsqueda binaria en el rango de valores. El enfoque con heap es O(k log n), lo que resulta eficiente cuando k es pequeño; la búsqueda binaria es O(n log(max-min)) y funciona mejor con valores grandes 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)) # 13Dos heaps para estadísticas acumuladas
El patrón de los dos heaps se generaliza más allá de la mediana. Puede utilizarlo para mantener un cuantil acumulado (por ejemplo, el percentil 25): dimensione el heap inferior para contener p*n elementos y el superior para contener (1-p)*n elementos. Cada vez que se añada un elemento, reajuste los heaps como antes. Este patrón aparece en problemas de estadísticas en streaming en los que se necesitan inserciones y consultas de cuantiles eficientes de forma simultánea.
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)Encontrar k puntos más cercanos al origen
Encontrar k puntos más cercanos al origen (LeetCode #973) utiliza un max-heap de tamaño k. Inserte la distancia al cuadrado de cada punto (para evitar calcular la raíz cuadrada). Cuando el heap supere k elementos, extraiga el más lejano. Los k puntos restantes son los k más cercanos. La complejidad es O(n log k). Como alternativa, puede utilizar quickselect para obtener O(n) en promedio, pero la solución con heap es más sencilla de implementar correctamente y de explicar durante una entrevista.
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}')Análisis temporal y espacial de dos heaps
El enfoque de dos heaps para la mediana logra O(log n) por cada addNum y O(1) por cada findMedian. El espacio es O(n) para almacenar todos los elementos. La combinación en k vías tiene un tiempo de O(n log k) y un espacio de O(k) para el heap. Estas complejidades son casi óptimas: puede demostrarse una cota inferior basada en comparaciones de Omega(n log k) para la combinación en k vías, lo que demuestra que la solución con heap es asintóticamente óptima. Exponga siempre estas complejidades con claridad en las entrevistas.
# 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)')Comprobación rápida
Ponga a prueba su comprensión de los conceptos de Data Structures & Algorithms — Coding Interview Prep de esta lección.
Resumen de la lección
En esta lección ha aprendido: MedianFinder con dos heaps, que logra una inserción O(log n) y una mediana O(1); la combinación en k vías con un min-heap, en un tiempo de O(n log k) y un espacio de O(k); y extensiones como la mediana de una ventana deslizante, el rango más pequeño y los k puntos más cercanos. A continuación, explorará las representaciones de grafos y la configuración de sus recorridos.
Preguntas frecuentes
¿La lección «Mediana de un flujo de datos y combinación k-way» es gratis?
Sí — el texto completo de «Mediana de un flujo de datos y combinación k-way» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de Coding Interview Prep, actualiza a CoddyKit PRO. El curso de Coding Interview Prep incluye 4 lecciones en total.
¿Qué aprenderé en «Mediana de un flujo de datos y combinación k-way»?
Mantenga dos heaps —un max-heap de la mitad menor y un min-heap de la mitad mayor— para actualizar la mediana en O(log n) y combine k listas ordenadas usando un heap. Practicas Coding Interview Prep con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar Coding Interview Prep?
No se requiere experiencia previa. Coding Interview Prep en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.
¿Cuánto tiempo toma la lección «Mediana de un flujo de datos y combinación k-way»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de Coding Interview Prep?
Sí. Cada lección de Coding Interview Prep incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Propiedad del heap y representación en array
- Heapify, push y pop desde cero
- heapq de Python y trucos para max-heap
- Mediana de un flujo de datos y combinación k-way