Skip to content

Python heapq: 우선순위 큐와 힙 연산을 간단하게

Soren Atelier
Name
Soren Atelier

Updated on

우선순위에 따라 항목을 처리해야 합니다 -- 작업 스케줄러의 태스크, 시뮬레이션의 이벤트, 그래프에서의 최단 경로 등. 정렬된 리스트가 확실한 선택처럼 보이지만, 정렬된 리스트에 삽입하는 것은 O(n)이므로 수백만 개의 항목을 처리할 때 병목이 됩니다. 요소를 추가할 때마다 전체 리스트를 다시 정렬하는 것은 O(n log n)으로 더 나쁩니다.

비용은 빠르게 누적됩니다. 삽입할 때마다 큐를 다시 정렬하는 작업 스케줄러는 대부분의 시간을 관리 업무에 낭비합니다. 매 단계마다 전체 리스트를 스캔하여 최소 거리를 찾는 경로 탐색 알고리즘은 선형 시간 연산을 이차 시간으로 바꿔버립니다.

Python의 heapq 모듈은 O(log n) 삽입과 O(log n) 최소 요소 추출을 제공하는 힙 기반 우선순위 큐를 제공합니다. 또한 전체 컬렉션을 정렬하지 않고 이터러블에서 N개의 가장 큰 또는 가장 작은 요소를 찾는 최적화된 함수도 포함합니다. 이 가이드에서는 heapq의 모든 함수를 실용적인 예제와 함께 다룹니다.

📚

    힙이란?

    힙은 각 부모 노드가 자식 노드보다 작거나 같은 이진 트리입니다(최소 힙). Python의 heapq는 일반 리스트를 사용하여 최소 힙을 구현하며, 가장 작은 요소는 항상 인덱스 0에 있습니다.

    주요 연산과 시간 복잡도:

    연산함수시간 복잡도
    항목 추가heapq.heappush(heap, item)O(log n)
    최소값 꺼내기heapq.heappop(heap)O(log n)
    추가 후 꺼내기heapq.heappushpop(heap, item)O(log n)
    꺼낸 후 추가heapq.heapreplace(heap, item)O(log n)
    리스트에서 힙 생성heapq.heapify(list)O(n)
    N개의 최소 항목heapq.nsmallest(n, iterable)O(n log k)
    N개의 최대 항목heapq.nlargest(n, iterable)O(n log k)

    기본 힙 연산

    힙 생성 및 사용

    import heapq
 
# Start with an empty heap (just a regular list)
heap = []
 
# Push items
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 2)
 
print(heap)        # [1, 2, 3, 5] (heap order, not sorted)
print(heap[0])     # 1 (smallest element, always at index 0)
 
# Pop smallest item
smallest = heapq.heappop(heap)
print(smallest)    # 1
print(heap)        # [2, 5, 3]

    리스트를 힙으로 변환

    import heapq
 
data = [5, 3, 8, 1, 2, 7, 4, 6]
heapq.heapify(data)  # In-place, O(n)
print(data)  # [1, 2, 4, 3, 5, 7, 8, 6] (heap order)
 
# Pop elements in sorted order
sorted_data = []
while data:
    sorted_data.append(heapq.heappop(data))
print(sorted_data)  # [1, 2, 3, 4, 5, 6, 7, 8]

    heappushpop과 heapreplace

    이 결합된 연산은 별도의 push/pop 호출보다 효율적입니다:

    import heapq
 
heap = [1, 3, 5, 7]
heapq.heapify(heap)
 
# heappushpop: push first, then pop smallest
result = heapq.heappushpop(heap, 2)
print(result)  # 1 (pushed 2, popped 1)
print(heap)    # [2, 3, 5, 7]
 
# heapreplace: pop smallest first, then push
result = heapq.heapreplace(heap, 10)
print(result)  # 2 (popped 2, pushed 10)
print(heap)    # [3, 7, 5, 10]

    nlargest와 nsmallest

    이 함수들은 전체 컬렉션을 정렬하지 않고 N개의 가장 큰 또는 가장 작은 요소를 찾습니다.

    import heapq
 
scores = [85, 92, 78, 95, 88, 72, 91, 83, 97, 69]
 
# Top 3 scores
top_3 = heapq.nlargest(3, scores)
print(top_3)  # [97, 95, 92]
 
# Bottom 3 scores
bottom_3 = heapq.nsmallest(3, scores)
print(bottom_3)  # [69, 72, 78]

    키 함수 사용

    둘 다 사용자 정의 비교를 위한 key 매개변수를 받습니다:

    import heapq
 
students = [
    {'name': 'Alice', 'gpa': 3.9},
    {'name': 'Bob', 'gpa': 3.5},
    {'name': 'Charlie', 'gpa': 3.8},
    {'name': 'Diana', 'gpa': 3.95},
    {'name': 'Eve', 'gpa': 3.2},
]
 
top_students = heapq.nlargest(3, students, key=lambda s: s['gpa'])
for s in top_students:
    print(f"{s['name']}: {s['gpa']}")
# Diana: 3.95
# Alice: 3.9
# Charlie: 3.8

    nlargest/nsmallest vs sorted() 사용 시기

    시나리오최적 접근법이유
    N이 1min() / max()O(n), 가장 간단
    N이 전체에 비해 작음nlargest / nsmallestO(n log k), 메모리 적게 사용
    N이 전체 길이에 가까움sorted(iterable)[:n]O(n log n)이지만 실제로 빠름
    모든 항목 정렬 필요sorted()전체 정렬이 적절

    최대 힙 구현

    Python의 heapq는 최소 힙만 제공합니다. 최대 힙을 만들려면 값을 부정합니다:

    import heapq
 
# Max-heap using negation
max_heap = []
for val in [5, 1, 3, 8, 2]:
    heapq.heappush(max_heap, -val)
 
# Pop largest items
while max_heap:
    print(-heapq.heappop(max_heap), end=' ')
# 8 5 3 2 1

    사용자 정의 객체의 우선순위 큐

    직접 비교할 수 없는 객체에는 튜플 (priority, counter, item)을 사용합니다:

    import heapq
from dataclasses import dataclass
 
@dataclass
class Task:
    name: str
    description: str
 
# Use (priority, tie_breaker, task) tuples
task_queue = []
counter = 0
 
def add_task(priority, task):
    global counter
    heapq.heappush(task_queue, (priority, counter, task))
    counter += 1
 
def get_next_task():
    if task_queue:
        priority, _, task = heapq.heappop(task_queue)
        return task
    return None
 
add_task(3, Task("Low", "Clean up logs"))
add_task(1, Task("Critical", "Fix production bug"))
add_task(2, Task("Medium", "Update docs"))
add_task(1, Task("Critical", "Patch security hole"))
 
while task_queue:
    task = get_next_task()
    print(f"{task.name}: {task.description}")
 
# Critical: Fix production bug
# Critical: Patch security hole
# Medium: Update docs
# Low: Clean up logs

    카운터는 타이브레이커 역할을 하여 동일한 우선순위의 항목이 삽입 순서(FIFO)로 반환되도록 합니다.

    실용적인 사용 사례

    정렬된 이터러블 병합

    import heapq
 
list1 = [1, 4, 7, 10]
list2 = [2, 5, 8, 11]
list3 = [3, 6, 9, 12]
 
merged = list(heapq.merge(list1, list2, list3))
print(merged)  # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

    heapq.merge()는 메모리 효율적입니다 -- 모든 리스트를 한 번에 메모리에 로드하지 않습니다.

    실행 중 중앙값

    import heapq
 
class RunningMedian:
    """Track the median of a stream of numbers."""
    def __init__(self):
        self.low = []   # max-heap (negated values)
        self.high = []  # min-heap
 
    def add(self, num):
        heapq.heappush(self.low, -num)
        heapq.heappush(self.high, -heapq.heappop(self.low))
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))
 
    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2
 
rm = RunningMedian()
for num in [2, 1, 5, 7, 2, 0, 5]:
    rm.add(num)
    print(f"Added {num}, median = {rm.median()}")

    다익스트라 최단 경로

    import heapq
 
def dijkstra(graph, start):
    """Find shortest paths from start to all nodes."""
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]
 
    while pq:
        dist, node = heapq.heappop(pq)
        if dist > distances[node]:
            continue
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(pq, (new_dist, neighbor))
 
    return distances
 
graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': [],
}
 
print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

    Jupyter에서 힙 성능 분석

    힙 기반 알고리즘을 대안과 비교하거나 우선순위 큐 성능을 프로파일링할 때, 인터랙티브 실험이 프로세스를 가속합니다. RunCell (opens in a new tab)은 Jupyter용 AI 에이전트로, 힙 연산 벤치마킹, 성능 특성 시각화, 데이터 구조 선택 최적화를 인터랙티브하게 도와줍니다.

    FAQ

    Python에서 heapq란?

    heapq는 일반 Python 리스트를 사용하여 최소 힙을 구현하는 표준 라이브러리 모듈입니다. 힙 속성(가장 작은 요소가 항상 인덱스 0)을 유지하면서 O(log n) 시간에 항목을 삽입하고 꺼내는 함수를 제공합니다. Python에서 우선순위 큐를 구현하는 표준 방법입니다.

    Python에서 최대 힙을 만드는 방법은?

    Python의 heapq는 최소 힙만 지원합니다. 최대 힙을 시뮬레이션하려면 삽입 시 값을 부정하고 꺼낼 때 다시 부정합니다: 삽입에 heapq.heappush(heap, -value), 최대값 가져오기에 -heapq.heappop(heap)을 사용합니다. 복잡한 객체의 경우 튜플에서 우선순위를 부정합니다: (-priority, counter, item).

    heapq와 sorted()는 언제 사용해야 하나요?

    heapq.nlargest(n, data) 또는 heapq.nsmallest(n, data)는 Top N 항목만 필요하고 N이 전체 데이터 크기에 비해 작을 때 사용합니다. 모든 항목을 순서대로 필요하거나 N이 전체 크기에 가까울 때는 sorted()를 사용합니다. N이 1일 때는 min()/max()를 사용합니다.

    heapq는 스레드 안전한가요?

    아니요. 개별 heappush와 heappop 호출은 원자적이지 않습니다. 여러 스레드가 같은 힙에 접근하는 경우 외부 동기화가 필요합니다. 스레드 안전한 우선순위 큐에는 적절한 잠금으로 heapq를 래핑하는 queue.PriorityQueue를 사용하세요.

    heapq 연산의 시간 복잡도는?

    heappush와 heappop은 모두 O(log n)입니다. heapify는 리스트를 O(n)에 힙으로 변환합니다. nlargest(k, data)와 nsmallest(k, data)는 O(n log k)로 실행됩니다(n은 데이터 크기, k는 요청된 항목 수).

    결론

    Python의 heapq 모듈은 우선순위 큐와 Top-N 선택을 위한 표준 도구입니다. 힙 기반 접근 방식은 O(log n) push 및 pop 연산을 제공하여 정렬된 리스트를 유지하는 것보다 극적으로 빠릅니다. 우선순위 큐 패턴에는 heappush/heappop을, Top-N 선택에는 nlargest/nsmallest를, 정렬된 시퀀스 결합에는 merge를, 기존 리스트 변환에는 heapify를 사용하세요.

    📚