Coding Interview Prep · 강의

데이터 스트림의 중앙값과 K-way 병합

Q: “데이터 스트림의 중앙값과 K-way 병합” 강의는 무료인가요?

네 — “데이터 스트림의 중앙값과 K-way 병합” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 Coding Interview Prep 강의 전체를 잠금 해제할 수 있습니다. Coding Interview Prep 강의에는 총 4개의 강의가 포함되어 있습니다.

Q: Coding Interview Prep을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 Coding Interview Prep은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.

Q: “데이터 스트림의 중앙값과 K-way 병합” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

Q: 이 Coding Interview Prep 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 Coding Interview Prep 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

두 힙(작은 절반의 최대 힙과 큰 절반의 최소 힙)을 유지해 O(log n)에 중앙값을 갱신하고, 힙을 사용해 k개의 정렬 리스트를 병합합니다.

데이터 스트림의 중앙값과 K-way 병합은(는) CoddyKit의 무료 Coding Interview Prep 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 Coding Interview Prep 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. Coding Interview Prep 강의에는 총 4개의 강의가 포함되어 있습니다.

데이터 스트림에서 중앙값 구하기 문제

데이터 스트림에서 중앙값 찾기(LeetCode #295)는 두 가지 연산을 효율적으로 지원하는 문제입니다. 숫자를 추가하는 addNum(num)과 현재 중앙값을 반환하는 findMedian()입니다. 원소 수가 짝수인 리스트의 중앙값은 가운데에 있는 두 값의 평균입니다. 단순한 정렬 리스트를 사용하면 삽입에 O(n), 중앙값 조회에 O(1)이 걸립니다. 최적의 해법은 두 힙을 사용하여 삽입을 O(log n), 중앙값 조회를 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')

두 힙을 사용하는 MedianFinder 구현

하위 절반에는 최대 힙을, 상위 절반에는 최소 힙을 유지합니다. 최대 힙의 크기가 최소 힙과 같거나 원소 하나만큼 더 많도록 항상 보장해야 합니다. 숫자를 추가할 때는 최대 힙에 삽입한 다음, 최대 힙의 최댓값이 최소 힙의 최솟값보다 크면 최대 힙의 최댓값을 최소 힙으로 옮겨 균형을 맞추고, 필요하면 두 힙의 크기도 다시 조정합니다.

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.0

MedianFinder 단계 추적

면접에서 해법을 설명하려면 두 힙 불변식이 유지되는 이유를 이해하는 것이 중요합니다. [5, 15, 1, 3]을 추가하는 과정을 단계별로 추적해 보겠습니다. 삽입할 때마다 하위 최대 힙이 더 작은 절반을 보유하도록 균형을 맞춥니다. 이 불변식이 항상 max(lo) <= min(hi)를 보장하므로, 한 힙 또는 두 힙의 최상단에서 중앙값을 간단히 확인할 수 있습니다.

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})')

슬라이딩 윈도우 중앙값

슬라이딩 윈도우 중앙값(LeetCode #480)은 배열을 따라 크기 k인 윈도우가 이동할 때 각 윈도우의 중앙값을 구하는 더 어려운 변형 문제입니다. 윈도우 밖으로 빠져나가는 원소를 처리하려면 두 힙 방식에 지연 삭제 집합을 추가합니다. 원소가 윈도우를 벗어나면 삭제 집합에 표시하고, 해당 원소가 어느 힙의 최상단에 도달하면 버립니다.

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]

K방향 병합: 문제

정렬된 K개 리스트 병합(LeetCode #23)은 외부 정렬, 데이터베이스 병합, 분산 시스템에 활용되는 기본 문제입니다. 총 n개의 노드를 포함하는 k개의 정렬된 연결 리스트가 주어질 때, 이를 하나의 정렬된 리스트로 병합합니다. 단순한 방식인 두 리스트씩 순차적으로 병합하면 O(kn)이고, 분할 정복을 사용하면 O(n log k)입니다. 힙 방식은 각 노드를 정확히 한 번 처리하며 노드마다 O(log k)의 작업이 필요하므로 전체 시간 복잡도는 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')

최소 힙을 사용하는 K방향 병합

각 리스트의 첫 노드로 힙을 초기화합니다. 각 단계에서 최솟값을 꺼내 결과에 추가하고, 해당 리스트의 다음 노드가 있으면 힙에 삽입합니다. 활성 리스트마다 하나의 헤드가 들어가므로 힙에는 항상 최대 k개의 원소만 있습니다. 총 n개의 노드를 각각 O(log k)의 힙 연산으로 처리하므로 전체 시간 복잡도는 O(n log k)이고, 힙에 필요한 공간 복잡도는 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]

K개 리스트를 포함하는 최소 범위

최소 범위(LeetCode #632)는 k개의 정렬된 리스트 각각에서 하나 이상의 원소가 범위 안에 포함되도록 하는 가장 작은 범위 [lo, hi]를 찾습니다. 각 리스트의 첫 번째 원소로 초기화한 최소 힙을 사용하고 현재 최댓값을 추적합니다. 현재 최솟값이 있는 리스트를 항상 다음 원소로 진행하여 범위를 좁힙니다. 어느 한 리스트가 소진되면 중지합니다.

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번째로 작은 원소

정렬된 행렬에서 K번째로 작은 원소(LeetCode #378)는 각 행과 열이 정렬된 n×n 행렬에서 k번째로 작은 원소를 찾는 문제입니다. 각 행을 정렬된 리스트로 간주하고 힙을 사용하는 K방향 병합을 적용합니다. 또는 값의 범위에 대해 이진 탐색을 수행할 수도 있습니다. 힙 방식은 O(k log n)으로 k가 작을 때 효율적이고, 이진 탐색은 O(n log(max-min))으로 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))  # 13

실시간 통계를 위한 두 힙

두 힙 패턴은 중앙값을 넘어 일반화할 수 있습니다. 이를 사용하여 실시간 분위수(예: 25번째 백분위수)를 유지할 수 있습니다. 하위 힙의 크기는 p*n개의 원소를 보유하도록 하고, 상위 힙의 크기는 (1-p)*n개의 원소를 보유하도록 설정합니다. 원소를 추가할 때마다 이전과 같은 방식으로 균형을 맞춥니다. 이 패턴은 삽입과 분위수 조회를 동시에 효율적으로 처리해야 하는 스트리밍 통계 문제에서 활용됩니다.

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개 점 찾기

원점에서 가장 가까운 K개 점(LeetCode #973)은 크기 k인 최대 힙을 사용합니다. 제곱근 계산을 피하기 위해 각 점의 제곱 거리를 삽입합니다. 힙의 크기가 k를 초과하면 가장 먼 점을 꺼냅니다. 남은 k개의 점이 가장 가까운 k개의 점입니다. 시간 복잡도는 O(n log k)입니다. 퀵셀렉트를 사용하면 평균 O(n)에 처리할 수도 있지만, 면접에서 올바르게 구현하고 설명하기에는 힙 해법이 더 간단합니다.

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}')

두 힙: 시간 및 공간 분석

중앙값을 구하는 두 힙 방식은 addNum마다 O(log n), findMedian마다 O(1)의 시간 복잡도를 달성합니다. 모든 원소를 저장하므로 공간 복잡도는 O(n)입니다. K방향 병합은 시간 복잡도 O(n log k)이고 힙에 필요한 공간 복잡도는 O(k)입니다. 이는 거의 최적인 결과입니다. K방향 병합에 대해 비교 기반 하한 Omega(n log k)를 증명할 수 있으므로 힙 해법이 점근적으로 최적임을 알 수 있습니다. 면접에서는 이러한 복잡도를 항상 명확하게 말하세요.

# 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)')

빠른 확인

이 단원에서 다룬 자료 구조 및 알고리즘 — 코딩 면접 대비 개념에 대한 이해도를 테스트해 보세요.

학습 내용 요약

이 단원에서 배운 내용: O(log n) 삽입과 O(1) 중앙값 조회를 달성하는 두 힙 MedianFinder, O(n log k) 시간과 O(k) 공간을 사용하는 최소 힙 기반 K방향 병합, 그리고 슬라이딩 윈도우 중앙값, 최소 범위, 가장 가까운 k개 점을 포함한 확장 내용입니다. 다음으로 그래프 표현과 순회 설정을 살펴봅니다.

자주 묻는 질문