[파이썬 100강] 64강. heapq로 우선순위 큐와 Top-N 처리하기

Last updated on 
[파이썬 100강] 64강. heapq로 우선순위 큐와 Top-N 처리하기

작업 대기열, 추천 점수, 실시간 알림 우선순위 같은 문제는 “정렬 한 번”으로 끝나지 않습니다. 데이터가 계속 들어오고, 그때그때 가장 중요한 것부터 꺼내야 하죠. 이때 표준 라이브러리 heapq를 쓰면 리스트 기반으로 가볍게 우선순위 큐를 만들 수 있습니다. 서론은 여기까지 하고 바로 예제로 들어가겠습니다.

핵심 개념

  • heapq는 최소 힙(min-heap) 을 제공한다. 즉, 가장 작은 값이 항상 루트(인덱스 0)에 온다.
  • 삽입(heappush)과 꺼내기(heappop)는 평균적으로 O(log n)이라, 매번 전체 정렬(O(n log n))하는 것보다 효율적이다.
  • 파이썬 기본 힙은 최소 힙이므로 “최대 우선순위 먼저”가 필요하면 음수 점수 트릭 또는 튜플 키를 사용한다.
  • nlargest, nsmallest는 Top-N 조회를 간단히 만들 때 매우 실용적이다.
  • 힙 원소를 튜플 (priority, seq, payload) 형태로 관리하면 동점 처리와 안정적인 순서를 동시에 잡을 수 있다.

heapq를 처음 배우면 “그냥 sort 쓰면 안 되나?”라는 의문이 생깁니다. 당연히 정적인 데이터 한 번 정렬할 때는 sorted()가 더 읽기 쉽습니다. 하지만 큐에 항목이 계속 추가되고 중간중간 최우선 항목을 계속 꺼내야 하는 상황이라면 이야기가 달라집니다. 매번 정렬하면 비용이 누적되고, 지연(latency)이 쉽게 튀기 때문입니다. 운영 배치나 API 워커에서 “실시간에 가까운 우선순위 처리”가 필요할 때 heapq는 작고 빠른 기본기입니다.

기본 사용

예제 1) 가장 기본 패턴: push/pop

>>> import heapq
>>> q = []
>>> heapq.heappush(q, 30)
>>> heapq.heappush(q, 10)
>>> heapq.heappush(q, 20)
>>> q
[10, 30, 20]
>>> heapq.heappop(q)
10
>>> heapq.heappop(q)
20
>>> heapq.heappop(q)
30

해설:

  • 힙 내부 리스트 모양은 “완전 정렬된 리스트”가 아닙니다. 최솟값만 빠르게 꺼낼 수 있게 유지된 구조입니다.
  • 그래서 q를 그대로 출력했을 때 [10, 30, 20]처럼 보여도 정상입니다.
  • 실무에서는 “전체 순서”보다 “다음에 꺼낼 최우선 하나”가 중요할 때 힙을 씁니다.

예제 2) 최대 우선순위를 먼저 꺼내고 싶을 때

>>> import heapq
>>> scores = []
>>> heapq.heappush(scores, (-95, "alice"))
>>> heapq.heappush(scores, (-87, "bob"))
>>> heapq.heappush(scores, (-99, "carol"))
>>> best = heapq.heappop(scores)
>>> -best[0], best[1]
(99, 'carol')

해설:

  • 기본이 최소 힙이므로, 큰 점수를 먼저 꺼내려면 점수에 음수를 붙입니다.
  • 튜플 첫 원소(우선순위)를 기준으로 비교되므로 (우선순위, 데이터) 패턴이 매우 자주 쓰입니다.
  • 랭킹, 추천, 모니터링 경보 점수 등 “높을수록 먼저 처리”하는 케이스에 바로 적용 가능합니다.

예제 3) 실전형 미니 케이스: 작업 큐

>>> import heapq
>>> from itertools import count
>>> seq = count()
>>> jobs = []
>>> def enqueue(priority, name):
...     heapq.heappush(jobs, (priority, next(seq), name))
...
>>> enqueue(1, "장애 알림 전송")
>>> enqueue(3, "주간 리포트 생성")
>>> enqueue(1, "핫픽스 배포")
>>> heapq.heappop(jobs)
(1, 0, '장애 알림 전송')
>>> heapq.heappop(jobs)
(1, 2, '핫픽스 배포')
>>> heapq.heappop(jobs)
(3, 1, '주간 리포트 생성')

해설:

  • 우선순위가 같은 작업이 있을 때 입력 순서까지 보장하려고 seq(증가 번호)를 넣었습니다.
  • 실무 큐에서 동점 정렬 규칙이 없으면 “가끔 순서가 바뀌는 것처럼 보이는” 문제로 디버깅 시간을 크게 잡아먹습니다.
  • (priority, seq, payload)는 단순하지만 매우 강력한 운영 패턴입니다.

예제 4) Top-N 빠르게 뽑기

>>> import heapq
>>> products = [
...     {"name": "A", "sales": 120},
...     {"name": "B", "sales": 98},
...     {"name": "C", "sales": 175},
...     {"name": "D", "sales": 143},
... ]
>>> top2 = heapq.nlargest(2, products, key=lambda x: x["sales"])
>>> [(p["name"], p["sales"]) for p in top2]
[('C', 175), ('D', 143)]

해설:

  • 전체를 다 정렬하지 않고도 필요한 N개만 빠르게 뽑을 수 있습니다.
  • 대시보드의 “상위 10개”, 알림의 “하위 5개 실패율” 같은 요구에서 즉시 활용됩니다.
  • 코드 가독성도 좋아서 팀 협업에서 유지보수 부담이 낮습니다.

자주 하는 실수

실수 1) 리스트에 append만 하고 힙 연산을 기대함

>>> import heapq
>>> q = []
>>> q.append(30)
>>> q.append(10)
>>> q.append(20)
>>> heapq.heappop(q)
30

원인:

  • append는 힙 속성을 보장하지 않습니다. 그냥 리스트 뒤에 붙일 뿐입니다.

해결:

>>> import heapq
>>> q = []
>>> heapq.heappush(q, 30)
>>> heapq.heappush(q, 10)
>>> heapq.heappush(q, 20)
>>> heapq.heappop(q)
10
  • 기존 리스트를 힙으로 바꿀 때는 heapq.heapify(q)를 반드시 호출하세요.
  • 큐 자료구조 규칙을 코드 리뷰 체크리스트에 넣으면 재발을 줄일 수 있습니다.

실수 2) 튜플 원소에 비교 불가능 객체를 넣어 TypeError 발생

>>> import heapq
>>> q = []
>>> heapq.heappush(q, (1, {"job": "a"}))
>>> heapq.heappush(q, (1, {"job": "b"}))
Traceback (most recent call last):
...
TypeError: '<' not supported between instances of 'dict' and 'dict'

원인:

  • 우선순위(첫 원소)가 같으면 다음 원소끼리 비교하는데, dict는 대소 비교가 안 됩니다.

해결:

>>> import heapq
>>> from itertools import count
>>> seq = count()
>>> q = []
>>> heapq.heappush(q, (1, next(seq), {"job": "a"}))
>>> heapq.heappush(q, (1, next(seq), {"job": "b"}))
>>> heapq.heappop(q)[2]
{'job': 'a'}
  • 동점 가능성이 있으면 비교 가능한 tie-breaker(증가 정수)를 항상 넣으세요.
  • “우선순위 튜플 스키마”를 팀 공통 규약으로 문서화하면 안전합니다.

실수 3) 빈 힙에서 heappop 호출

  • 증상: 런타임에서 IndexError: index out of range 발생
  • 원인: 소비자(consumer)가 생산자(producer)보다 빨리 동작해 큐가 비었는데 보호 로직이 없음
  • 해결: if q: 확인 후 pop 하거나, 함수로 감싸 None 반환 정책을 둔다
>>> import heapq
>>> q = []
>>> def safe_pop(heap):
...     return None if not heap else heapq.heappop(heap)
...
>>> safe_pop(q)
>>> heapq.heappush(q, 7)
>>> safe_pop(q)
7

실무 패턴

  • 입력 검증 규칙

    • 우선순위는 정수/실수 한 타입으로 통일(문자열 금지)
    • 우선순위 방향(작을수록 먼저 vs 클수록 먼저)을 함수 이름에 명시 (push_min_priority, push_max_score)

  • 로그/예외 처리 규칙

    • 큐 push/pop 시점에 큐 길이(len(heap))를 함께 기록해 병목 구간을 추적
    • 빈 큐 pop은 장애로 보지 말고 정상 상태(대기)로 분류해 경보 노이즈 감소

  • 재사용 함수/구조화 팁

    • enqueue/dequeue 래퍼 함수를 만들어 heapq 직접 접근을 줄이면 규칙이 깨지지 않음
    • payload를 dict로 두고 필수 키(id, created_at, type)를 고정하면 후속 파이프라인이 단순해짐

  • 성능/메모리 체크 포인트

    • Top-N이 매우 작고 전체 데이터가 매우 클 때 heapq.nlargest가 유리
    • 하지만 N이 전체 크기에 가까우면 sorted가 더 단순하고 충분히 빠를 수 있음
    • 오래 쌓이는 큐는 주기적으로 드레인/아카이브 정책을 넣어 메모리 폭주를 방지

  • 운영 관점 체크리스트

    • 동점 처리 규칙이 문서화되어 있는가?
    • 우선순위 역전(숫자 정의 실수) 테스트 케이스가 있는가?
    • 장애 시 큐 재처리(재시도) 정책이 있는가?

오늘의 결론

한 줄 요약: heapq는 “계속 들어오는 일감 중 지금 가장 중요한 것”을 빠르게 고르는 데 최적화된 실전 도구다.

기억할 것:

  • 힙은 정렬 리스트가 아니라 “최우선 원소 빠른 접근” 자료구조다.
  • 동점 처리용 seq를 넣으면 운영 중 이상한 순서 문제를 크게 줄일 수 있다.
  • Top-N은 nlargest/nsmallest로 간결하게 구현하고, 필요 이상 복잡한 정렬 로직을 피한다.

연습문제

  1. 우선순위(낮을수록 먼저)와 작업명을 입력받아 큐에 넣고, 하나씩 꺼내 처리 순서를 출력하는 enqueue/dequeue_all 함수를 작성해 보세요.
  2. 사용자 점수 리스트에서 상위 3명을 heapq.nlargest로 뽑되, 같은 점수면 이름 오름차순으로 정렬되도록 키 함수를 설계해 보세요.
  3. (priority, seq, payload) 구조를 사용하는 큐에서 priority가 잘못 문자열로 들어오면 예외를 발생시키는 검증 코드를 추가해 보세요.

이전 강의 정답

  1. 주문 처리 시간 리스트의 평균·중앙값·모집단 표준편차 계산
>>> import statistics as stats
>>> values = [3.2, 3.5, 3.1, 3.4, 12.0, 3.3]
>>> round(stats.mean(values), 3), stats.median(values), round(stats.pstdev(values), 3)
(4.417, 3.35, 3.223)

해설: 평균은 12.0의 영향을 크게 받아 4.417까지 올라가지만 중앙값은 3.35로 평소 구간을 더 잘 보여줍니다. 산포(표준편차)가 커진 것도 이상치 존재 신호입니다.

  1. mode()와 multimode() 비교
>>> import statistics as stats
>>> scores = [70, 75, 75, 80, 80, 90]
>>> stats.mode(scores)
75
>>> stats.multimode(scores)
[75, 80]

해설: 동률 최빈값이 가능한 데이터이므로 리포트에는 "최빈값: 75, 80"처럼 복수 표기를 허용하는 포맷이 안전합니다.

  1. safe_summary() 함수 작성
>>> import statistics as stats
>>> def safe_summary(values):
...     if not values:
...         return {"mean": None, "median": None, "pstdev": None}
...     if len(values) == 1:
...         return {"mean": values[0], "median": values[0], "pstdev": 0.0}
...     return {
...         "mean": stats.mean(values),
...         "median": stats.median(values),
...         "pstdev": stats.pstdev(values),
...     }
...
>>> safe_summary([])
{'mean': None, 'median': None, 'pstdev': None}
>>> safe_summary([10])
{'mean': 10, 'median': 10, 'pstdev': 0.0}
>>> safe_summary([10, 20, 30])
{'mean': 20, 'median': 20, 'pstdev': 8.16496580927726}

실습 환경/재현 정보

  • 실행 환경: conda env python100 (Python 3.11.14)
  • 가정한 OS: macOS/Linux 공통
  • 사용 라이브러리: 표준 라이브러리 heapq, itertools
  • 검증 방식: REPL(pycon) 예제 기준 수동 점검 + 발행 전 dry-run