Mediana de um Fluxo de Dados e Intercalação em K Vias
Mantenha dois heaps — um max-heap da metade menor e um min-heap da metade maior — para atualizações da mediana em O(log n) e intercale k listas ordenadas usando um heap.
Mediana de um Fluxo de Dados e Intercalação em K Vias é uma aula grátis de Coding Interview Prep no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de Coding Interview Prep, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de Coding Interview Prep inclui 4 aulas no total.
Problema da mediana de um fluxo de dados
Encontrar a mediana de um fluxo de dados (LeetCode #295) pede que você dê suporte eficientemente a duas operações: addNum(num) para adicionar um número e findMedian() para retornar a mediana atual. A mediana de uma lista de tamanho par é a média dos dois valores centrais. Uma lista ordenada por força bruta permite inserção em O(n) e mediana em O(1). A solução ideal usa dois montes para obter inserção em O(log n) e mediana em 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')Implementação de MedianFinder com dois montes
Mantenha um monte máximo para a metade inferior e um monte mínimo para a metade superior. Garanta sempre que o monte máximo tenha o mesmo tamanho ou um elemento a mais que o monte mínimo. Ao adicionar um número: faça push no monte máximo e, em seguida, equilibre movendo o elemento superior do monte máximo para o monte mínimo se ele exceder o mínimo do monte mínimo; depois, reequilibre os tamanhos se necessário.
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.0Percorra as etapas de MedianFinder
Compreender por que a invariante dos dois montes é mantida é fundamental para explicar a solução em uma entrevista. Vamos acompanhar passo a passo a adição de [5, 15, 1, 3]. Após cada inserção: equilibre para que o monte máximo inferior contenha a metade menor. A invariante garante que max(lo) <= min(hi) seja sempre válida, tornando a mediana facilmente acessível no topo de um dos montes ou de ambos.
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 da janela deslizante
A mediana da janela deslizante (LeetCode #480) é uma variante mais difícil: encontrar a mediana de cada janela de tamanho k enquanto ela percorre o vetor. A abordagem com dois montes é ampliada com um conjunto de exclusão tardia para lidar com os elementos que saem da janela. Quando um elemento deixa a janela, marque-o no conjunto de exclusão; quando ele chegar ao topo de qualquer monte, descarte-o.
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]Mesclagem de K vias: o problema
Mesclar K listas ordenadas (LeetCode #23) é um problema fundamental, com aplicações em ordenação externa, mesclagens de bancos de dados e sistemas distribuídos. Dadas k listas encadeadas ordenadas, totalizando n nós, mescle-as em uma única lista ordenada. A abordagem ingênua (mesclar duas por vez) tem complexidade O(kn), ou O(n log k) com divisão e conquista. A abordagem com monte processa cada nó exatamente uma vez, com O(log k) de trabalho por nó: O(n log k) no 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')Mesclagem de K vias com um monte mínimo
Inicialize o monte com o primeiro nó de cada lista. A cada etapa, faça pop do mínimo, adicione-o ao resultado e faça push do próximo nó dessa lista, se houver. O monte sempre tem no máximo k elementos — um início por lista ativa. Como processamos n nós no total, com O(log k) operações de monte para cada um, o tempo total é O(n log k) e o espaço é O(k) para o monte.
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]Menor intervalo que cobre K listas
Menor intervalo (LeetCode #632) encontra o menor intervalo [lo, hi] tal que pelo menos um elemento de cada uma das k listas ordenadas esteja dentro dele. Use um monte mínimo inicializado com o primeiro elemento de cada lista e acompanhe o máximo atual. Encurte o intervalo avançando sempre na lista que contém o mínimo atual. Pare quando qualquer lista se esgotar.
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 menor em uma matriz
K-ésimo menor elemento em uma matriz ordenada (LeetCode #378): uma matriz n×n em que cada linha e coluna está ordenada. Encontre o k-ésimo menor elemento. Trate cada linha como uma lista ordenada e use a mesclagem de K vias com um monte. Como alternativa, faça uma busca binária no intervalo de valores. A abordagem com monte é O(k log n), eficiente quando k é pequeno; a busca binária é O(n log(max-min)), funcionando melhor para 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)) # 13Dois montes para estatísticas contínuas
O padrão de dois montes se generaliza além da mediana. Você pode usá-lo para manter um quantil em fluxo (por exemplo, o percentil 25): dimensione o monte inferior para conter p*n elementos e o monte superior para conter (1-p)*n elementos. Cada vez que um elemento for adicionado, reequilibre como antes. Esse padrão aparece em problemas de estatística em fluxo nos quais você precisa realizar inserções eficientes e consultas de quantis simultaneamente.
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)Encontre os K pontos mais próximos da origem
K pontos mais próximos da origem (LeetCode #973) usa um monte máximo de tamanho k. Faça push da distância ao quadrado de cada ponto, para evitar calcular a raiz quadrada. Quando o monte ultrapassar k, faça pop do ponto mais distante. Os k pontos restantes são os k mais próximos. A complexidade é O(n log k). Uma alternativa usa seleção rápida, com complexidade média O(n), mas a solução com monte é mais simples de implementar corretamente e explicar durante uma 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álise de tempo e espaço dos dois montes
A abordagem com dois montes para a mediana alcança O(log n) por addNum e O(1) por findMedian. O espaço é O(n) para armazenar todos os elementos. A mesclagem de K vias tem complexidade de tempo O(n log k) e espaço O(k) para o monte. Esses valores são quase ideais: é possível provar um limite inferior baseado em comparações de Ω(n log k) para a mesclagem de K vias, mostrando que a solução com monte é assintoticamente ideal. Sempre declare essas complexidades com clareza nas 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)')Verificação rápida
Teste sua compreensão dos conceitos de Estruturas de Dados & Algoritmos — Preparação para Entrevistas de Programação desta lição.
Recapitulação da lição
Nesta lição, você aprendeu: o MedianFinder com dois montes, alcançando inserção em O(log n) e mediana em O(1); a mesclagem de K vias com um monte mínimo, em O(n log k) de tempo e O(k) de espaço; e extensões que incluem mediana de janela deslizante, menor intervalo e pontos mais próximos entre K pontos. Em seguida, exploraremos representações de grafos e a preparação para sua travessia.
Perguntas Frequentes
A aula “Mediana de um Fluxo de Dados e Intercalação em K Vias” é grátis?
Sim — o texto completo de “Mediana de um Fluxo de Dados e Intercalação em K Vias” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de Coding Interview Prep, atualize para CoddyKit PRO. O curso de Coding Interview Prep inclui 4 aulas no total.
O que vou aprender em “Mediana de um Fluxo de Dados e Intercalação em K Vias”?
Mantenha dois heaps — um max-heap da metade menor e um min-heap da metade maior — para atualizações da mediana em O(log n) e intercale k listas ordenadas usando um heap. Você pratica Coding Interview Prep com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.
Preciso ter experiência prévia para começar Coding Interview Prep?
Nenhuma experiência prévia é necessária. Coding Interview Prep no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.
Quanto tempo leva a aula “Mediana de um Fluxo de Dados e Intercalação em K Vias”?
A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.
Posso escrever e executar código nesta aula de Coding Interview Prep?
Sim. Cada aula de Coding Interview Prep inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.
Todas as aulas deste curso
- Propriedade de Heap e Representação em Array
- Heapify, Push e Pop do Zero
- heapq do Python e Técnicas de Max-Heap
- Mediana de um Fluxo de Dados e Intercalação em K Vias