heapq — 힙 큐 알고리즘

heapq — 힙 큐 알고리즘

heapq 모듈은 우선순위 큐 알고리즘으로도 알려진 힙 큐 알고리즘의 구현을 제공해요.

**최소 힙(min-heap)**은 모든 부모 노드의 값이 그 어떤 자식보다 작거나 같은 이진 트리예요. 이 조건을 **힙 불변식(heap invariant)**이라고 부릅니다.

최소 힙의 경우, 이 구현은 비교되는 요소가 존재하는 모든 k에 대해 heap[k] <= heap[2*k+1]이고 heap[k] <= heap[2*k+2]인 리스트를 사용해요. 요소는 0부터 셉니다. 최소 힙의 흥미로운 속성은 가장 작은 요소가 항상 루트, 즉 heap[0]이라는 점입니다.

**최대 힙(max-heap)**은 반대 불변식을 만족해요: 모든 부모 노드의 값이 그 어떤 자식보다 큽니다. 이들은 비교되는 요소가 존재하는 모든 k에 대해 maxheap[2*k+1] <= maxheap[k]이고 maxheap[2*k+2] <= maxheap[k]인 리스트로 구현됩니다. 루트 maxheap[0]에는 가장 큰 요소가 담깁니다; heap.sort(reverse=True)는 최대 힙 불변식을 유지해요.

heapq API는 교과서의 힙 알고리즘과 두 면에서 다릅니다: (a) 0 기반 인덱싱을 씁니다. 이는 노드의 인덱스와 그 자식들의 인덱스 사이의 관계를 약간 덜 분명하게 만들지만, Python이 0 기반 인덱싱을 쓰므로 더 적합해요. (b) 교과서는 제자리 정렬에 적합해서 최대 힙을 자주 강조하지만, 우리 구현은 Python list에 더 잘 대응하므로 최소 힙을 선호해요.

이 두 가지 면 덕분에 힙을 놀라움 없이 보통 Python 리스트로 볼 수 있어요: heap[0]이 가장 작은 항목이고, heap.sort()가 힙 불변식을 유지합니다!

list.sort()처럼, 이 구현은 최소 힙과 최대 힙 모두에 비교에 < 연산자만 사용해요.

아래 API와 이 문서에서, 수식어 없는 용어 heap은 일반적으로 최소 힙을 가리킵니다. 최대 힙용 API는 _max 접미사로 이름 붙여져 있어요.

힙을 만들려면 []로 초기화된 리스트를 쓰거나, 기존 리스트를 heapify() 또는 heapify_max() 함수로 각각 최소 힙 또는 최대 힙으로 변환하세요.

출처: Python 표준 라이브러리

최소 힙용 함수

heapify()

• heapq.heapify(x)

리스트 x를 제자리에서 선형 시간에 최소 힙으로 변환해요.

heappush()

• heapq.heappush(heap, item)

값 item을 heap에 밀어 넣고, 최소 힙 불변식을 유지해요.

heappop()

• heapq.heappop(heap)

heap에서 가장 작은 항목을 꺼내(pop) 반환하며, 최소 힙 불변식을 유지해요. 힙이 비어 있으면 IndexError가 발생합니다. 꺼내지 않고 가장 작은 항목에 접근하려면 heap[0]을 쓰세요.

heappushpop()

• heapq.heappushpop(heap, item)

item을 힙에 밀어 넣은 다음 heap에서 가장 작은 항목을 꺼내 반환해요. 이 결합 동작은 heappush() 다음에 별도의 heappop()을 호출하는 것보다 더 효율적으로 실행됩니다.

heapreplace()

• heapq.heapreplace(heap, item)

heap에서 가장 작은 항목을 꺼내 반환하고, 새 item도 밀어 넣어요. 힙 크기는 변하지 않습니다. 힙이 비어 있으면 IndexError가 발생해요.

이 한 단계 연산은 heappop() 다음에 heappush()를 하는 것보다 효율적이고, 고정 크기 힙을 사용할 때 더 적절할 수 있어요. 이 꺼내기/밀어넣기 조합은 항상 힙에서 요소를 반환하고 그것을 item으로 교체합니다.

반환되는 값은 추가한 item보다 클 수 있어요. 그게 원하지 않으면 heappushpop()을 쓰는 걸 고려해 보세요. 그것의 밀어넣기/꺼내기 조합은 두 값 중 더 작은 것을 반환하고 더 큰 값을 힙에 남겨 둡니다.

최대 힙용 함수

• heapq.heapify_max(x) — 리스트 x를 제자리에서 선형 시간에 최대 힙으로 변환. 버전 3.14에서 추가됨.
• heapq.heappush_max(heap, item) — 값 item을 최대 힙 heap에 밀어 넣고 최대 힙 불변식 유지. 버전 3.14에서 추가됨.
• heapq.heappop_max(heap) — 최대 힙 heap에서 가장 큰 항목을 꺼내 반환하며 최대 힙 불변식 유지. 최대 힙이 비어 있으면 IndexError 발생. 꺼내지 않고 가장 큰 항목에 접근하려면 maxheap[0] 사용. 버전 3.14에서 추가됨.
• heapq.heappushpop_max(heap, item) — item을 최대 힙 heap에 밀어 넣은 다음 heap에서 가장 큰 항목을 꺼내 반환. 이 결합 동작은 heappush_max() 다음에 별도의 heappop_max() 호출보다 효율적. 버전 3.14에서 추가됨.
• heapq.heapreplace_max(heap, item) — 최대 힙 heap에서 가장 큰 항목을 꺼내 반환하고 새 item도 밀어 넣음. 최대 힙 크기는 변하지 않음. 최대 힙이 비어 있으면 IndexError 발생. 반환되는 값은 추가한 item보다 작을 수 있음. 자세한 사용 노트는 유사 함수 heapreplace() 참고. 버전 3.14에서 추가됨.

일반 목적 함수

이 모듈은 힙에 기반한 세 가지 일반 목적 함수도 제공해요.

merge()

• heapq.merge(*iterables, key=None, reverse=False)

여러 개의 정렬된 입력을 하나의 정렬된 출력으로 병합해요(예: 여러 로그 파일의 타임스탬프 항목 병합). 정렬된 값들에 대한 이터레이터를 반환합니다.

sorted(itertools.chain(*iterables))와 비슷하지만 이터러블을 반환하고, 데이터를 한 번에 모두 메모리로 끌어오지 않으며, 각 입력 스트림이 이미 정렬돼 있다고(작은 것에서 큰 것으로) 가정해요.

키워드 인자로 지정해야 하는 두 개의 선택적 인자가 있어요. key는 각 입력 요소에서 비교 키를 추출하는 데 쓰는 한 인자 키 함수를 지정해요. 기본값은 None(요소를 직접 비교)입니다. reverse는 불리언 값이에요. True로 설정하면 각 비교가 뒤집힌 것처럼 입력 요소를 병합합니다. sorted(itertools.chain(*iterables), reverse=True)와 비슷한 동작을 얻으려면 모든 이터러블이 가장 큰 것에서 가장 작은 것으로 정렬돼야 해요.

버전 3.5에서 변경: 선택적 key와 reverse 매개변수 추가.

nlargest()

• heapq.nlargest(n, iterable, key=None)

iterable이 정의하는 데이터셋에서 n개 가장 큰 요소가 담긴 리스트를 반환해요. key가 제공되면 iterable의 각 요소에서 비교 키를 추출하는 한 인자 함수를 지정해요(예: key=str.lower). 다음와 동일합니다: sorted(iterable, key=key, reverse=True)[:n].

nsmallest()

• heapq.nsmallest(n, iterable, key=None)

iterable이 정의하는 데이터셋에서 n개 가장 작은 요소가 담긴 리스트를 반환해요. key가 제공되면 iterable의 각 요소에서 비교 키를 추출하는 한 인자 함수를 지정해요(예: key=str.lower). 다음와 동일합니다: sorted(iterable, key=key)[:n].

마지막 두 함수는 더 작은 n 값에 대해 가장 잘 동작해요. 더 큰 값에는 sorted() 함수를 쓰는 것이 더 효율적입니다. 또 n==1일 때는 내장 min()과 max() 함수를 쓰는 것이 더 효율적이에요. 이 함수들을 반복 사용해야 하면 이터러블을 실제 힙으로 바꾸는 걸 고려해 보세요.

기본 예제

모든 값을 힙에 밀어 넣은 다음 가장 작은 값을 하나씩 꺼내면 힙 정렬(heapsort)을 구현할 수 있어요:

>>> def heapsort(iterable):
...     h = []
...     for value in iterable:
...         heappush(h, value)
...     return [heappop(h) for i in range(len(h))]
...
>>> heapsort([1, 3, 5, 7, 9, 2, 4, 6, 8, 0])
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

이는 sorted(iterable)과 비슷하지만, sorted()와 달리 이 구현은 안정적이지 않아요.

힙 요소는 튜플일 수 있어요. 이는 추적 중인 기본 레코드와 함께 비교 값(작업 우선순위 같은)을 할당하는 데 유용합니다:

>>> h = []
>>> heappush(h, (5, 'write code'))
>>> heappush(h, (7, 'release product'))
>>> heappush(h, (1, 'write spec'))
>>> heappush(h, (3, 'create tests'))
>>> heappop(h)
(1, 'write spec')

다른 응용

중앙값(median)은 숫자 집합의 중심 경향 측정치예요. 이상치에 의해 왜곡된 분포에서는 중앙값이 평균(산술 평균)보다 더 안정적인 추정치를 제공합니다. 실행 중앙값(running median)은 새 데이터가 도착함에 따라 계속 갱신되는 온라인 알고리즘입니다.

실행 중앙값은 두 힙을 균형시키면 효율적으로 구현할 수 있는데, 중간점 이하 값용 최대 힙과 중간점 초과 값용 최소 힙을 쓰는 거예요. 두 힙의 크기가 같으면 새 중앙값은 두 힙 꼭대기의 평균이고, 그렇지 않으면 중앙값은 더 큰 힙의 꼭대기에 있습니다:

def running_median(iterable):
    "Yields the cumulative median of values seen so far."

    lo = []  # max-heap
    hi = []  # min-heap (same size as or one smaller than lo)

    for x in iterable:
        if len(lo) == len(hi):
            heappush_max(lo, heappushpop(hi, x))
            yield lo[0]
        else:
            heappush(hi, heappushpop_max(lo, x))
            yield (lo[0] + hi[0]) / 2

예를 들어:

>>> list(running_median([5.0, 9.0, 4.0, 12.0, 8.0, 9.0]))
[5.0, 7.0, 5.0, 7.0, 8.0, 8.5]

우선순위 큐 구현 노트

우선순위 큐는 힙의 흔한 용도로, 여러 가지 구현 과제를 제시해요:

• 정렬 안정성: 동일한 우선순위를 가진 두 작업을 원래 추가된 순서대로 반환하려면 어떻게 하나?
• 우선순위가 같고 작업에 기본 비교 순서가 없으면 (priority, task) 쌍의 튜플 비교가 깨진다.
• 작업의 우선순위가 바뀌면 힙의 새 위치로 어떻게 옮기나?
• 아니면 대기 중인 작업을 삭제해야 하면, 그것을 찾아 큐에서 제거하려면 어떻게 하나?

처음 두 과제에 대한 해결책은 우선순위, 항목 번호, 작업을 포함하는 3요소 리스트로 항목을 저장하는 것입니다. 항목 번호는 타이브레이커 역할을 해서, 같은 우선순위를 가진 두 작업이 추가된 순서대로 반환됩니다. 그리고 두 항목 번호가 같을 수 없으므로, 튜플 비교는 두 작업을 직접 비교하려 시도하지 않습니다.

비교 불가한 작업 문제에 대한 또 다른 해결책은 작업 항목을 무시하고 우선순위 필드만 비교하는 래퍼 클래스를 만드는 것입니다:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

남은 과제들은 대기 중인 작업을 찾고, 그 우선순위를 변경하거나 완전히 제거하는 것과 관련돼 있어요. 작업을 찾는 것은 큐의 항목을 가리키는 사전으로 할 수 있습니다.

항목 제거나 우선순위 변경은 힙 구조 불변식을 깨뜨리기 때문에 더 어렵습니다. 그래서 가능한 해결책은 항목을 제거된 것으로 표시하고 수정된 우선순위로 새 항목을 추가하는 것입니다:

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
counter = itertools.count()     # unique sequence count

def add_task(task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED. Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

이론

힙은 0부터 요소를 세어 모든 k에 대해 a[k] <= a[2*k+1]이고 a[k] <= a[2*k+2]인 배열이에요. 비교를 위해 존재하지 않는 요소는 무한대로 간주됩니다. 힙의 흥미로운 속성은 a[0]이 항상 가장 작은 요소라는 점입니다.

위의 이상한 불변식은 토너먼트를 위한 효율적인 메모리 표현을 의도한 것입니다. 아래 숫자는 k이지 a[k]가 아니에요:

     0

    1 2

  3 4 5 6

 7 8 9 10 11 12 13 14

15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30

위 트리에서 각 셀 k는 2*k+1과 2*k+2를 덮고 있습니다. 스포츠에서 보는 보통 이진 토너먼트에서는 각 셀이 그것이 덮는 두 셀의 승자이고, 트리를 따라 승자를 추적하면 그가 상대했던 모든 상대를 볼 수 있어요. 그러나 이런 토너먼트의 많은 컴퓨터 응용에서는 승자의 역사를 추적할 필요가 없습니다. 메모리를 더 효율적으로 쓰려면, 승자가 승격될 때 더 낮은 레벨의 다른 무언가로 그것을 대체하려 하며, 규칙은 셀과 그것이 덮는 두 셀이 서로 다른 세 항목을 담지만 맨 위 셀이 두 덮힌 셀을 "이긴다"는 것이 됩니다.

이 힙 불변식이 항상 보호되면 인덱스 0은 분명히 전체 승자입니다. 그것을 제거하고 "다음" 승자를 찾는 가장 단순한 알고리즘적 방법은 어떤 패자(위 그림에서 셀 30이라 하자)를 0 위치로 옮긴 다음 이 새 0을 값들을 교환하며 트리 아래로 침투시켜 불변식이 다시 세워질 때까지 하는 것입니다. 이는 분명히 트리의 전체 항목 수에 대해 로그적(logarithmic)이에요. 모든 항목을 반복하면 O(n log n) 정렬을 얻습니다.

이 정렬의 멋진 특징은 정렬이 진행되는 동안 새 항목을 효율적으로 삽입할 수 있다는 점입니다. 단, 삽입된 항목이 마지막으로 추출한 0번째 요소보다 "더 낫지" 않아야 해요. 이는 트리가 모든 도착 이벤트를 보유하고 "승리" 조건이 가장 작은 예약 시간을 의미하는 시뮬레이션 맥락에서 특히 유용합니다. 이벤트가 실행될 다른 이벤트를 예약하면 그것들은 미래로 예약되므로 쉽게 힙에 들어갈 수 있어요. 그래서 힙은 스케줄러를 구현하는 좋은 구조입니다(제 MIDI 시퀀서에 이걸 썼어요 :-).

스케줄러 구현용 다양한 구조가 광범위하게 연구됐는데, 힙은 속도가 상당히 빠르고, 속도가 거의 일정하며, 최악의 경우가 평균 경우와 크게 다르지 않아서 여기에 좋아요. 그러나 전반적으로 더 효율적인 다른 표현들도 있는데, 최악의 경우가 끔찍할 수 있죠.

힙은 대규모 디스크 정렬에서도 아주 유용합니다. 큰 정렬은 "런(run)"을 만들어내고(사전 정렬된 시퀀스로, 크기가 보통 CPU 메모리 양과 관련), 이어서 이 런들을 병합하는 단계를 거치는데, 그 병합은 종종 아주 교묘하게 조직됩니다. 초기 정렬이 가능한 가장 긴 런을 만들어내는 것이 아주 중요해요. 토너먼트는 그것을 달성하는 좋은 방법입니다. 토너먼트를 보유할 수 있는 사용 가능한 모든 메모리를 써서, 현재 런에 맞는 항목들을 교체·침투시키면, 무작위 입력에 대해 메모리의 두 배 크기 런을 만들고, 퍼지하게(fuzzily) 정렬된 입력에는 훨씬 더 좋습니다.

더욱이 0번째 항목을 디스크에 출력하고 현재 토너먼트에 맞지 않을 수 있는 입력을 얻으면(값이 마지막 출력 값보다 "이기므로"), 그것은 힙에 맞지 않아 힙 크기가 줄어듭니다. 해방된 메모리는 즉시 두 번째 힙을 점진적으로 만드는 데 교묘하게 재사용될 수 있는데, 그 힙은 첫 힙이 녹는 것과 정확히 같은 속도로 커집니다. 첫 힙이 완전히 사라지면 힙을 바꾸고 새 런을 시작합니다. 교묘하고 꽤 효과적이에요!

한마디로 힙은 알아 둘 가치가 있는 유용한 메모리 구조입니다. 저는 몇몇 응용에서 씁니다, 그리고 'heap' 모듈을 곁에 두는 게 좋다고 생각해요. :-)

더 알아보기

• Python 표준 라이브러리의 bisect 모듈 — 배열 이분(bisection) 알고리즘
• Python 공식 문서: heapq