[파이썬 100강] 93강. graphlib.TopologicalSorter로 작업 의존성 순서화하기

Last updated on 
[파이썬 100강] 93강. graphlib.TopologicalSorter로 작업 의존성 순서화하기

작업 순서가 얽히기 시작하면 for문만으로는 금방 한계가 옵니다. 배치 파이프라인, 데이터 변환 단계, 빌드/배포, 문서 생성 같은 자동화는 거의 항상 “이 작업이 끝나야 다음 작업을 시작할 수 있다”는 의존성을 갖기 때문입니다. 이번 강의에서는 **표준 라이브러리 graphlib.TopologicalSorter**로 의존성 그래프를 안전하게 정렬하고, 병렬 실행 가능한 단위까지 뽑아내는 실무 패턴을 정리합니다.

핵심 개념

  • **위상 정렬(topological sort)**은 “선행 작업 → 후행 작업” 제약을 깨지 않는 실행 순서를 만드는 알고리즘입니다.
  • TopologicalSorter는 노드별 선행 노드 집합을 입력받아, 순차 실행(static_order) 또는 단계별 준비 작업(get_ready)을 제공합니다.
  • 순환 의존(cycle)이 있으면 정상 실행 순서를 만들 수 없으므로, 반드시 감지하고 빠르게 실패해야 합니다.

핵심은 단순히 “정렬된 리스트 하나 받기”가 아닙니다. 실무에서 중요한 건 세 가지입니다. 첫째, 입력 형식을 팀이 합의해서 관리 가능한 구조로 유지하는 것. 둘째, 순환 의존이 생겼을 때 조용히 넘어가지 않고 배포 전에 바로 막는 것. 셋째, 병렬화 가능한 단계(동시에 실행해도 되는 작업 묶음)를 활용해 전체 소요 시간을 줄이는 것입니다. TopologicalSorter는 이 세 가지를 표준 라이브러리 수준에서 깔끔하게 해결해 주기 때문에, 별도 외부 패키지 없이도 충분히 운영 가능한 의존성 실행기를 만들 수 있습니다.

기본 사용

예제 1) static_order()로 전체 실행 순서 얻기

>>> from graphlib import TopologicalSorter
>>> deps = {
...     "fetch_raw": set(),
...     "clean_data": {"fetch_raw"},
...     "build_features": {"clean_data"},
...     "train_model": {"build_features"},
...     "publish_report": {"train_model"},
... }
>>> ts = TopologicalSorter(deps)
>>> list(ts.static_order())
['fetch_raw', 'clean_data', 'build_features', 'train_model', 'publish_report']

해설:

  • 딕셔너리 키는 “현재 작업”, 값은 “현재 작업이 의존하는 선행 작업 집합”입니다.
  • static_order()는 한 번에 전체 순서를 반환하므로, 순차 실행 배치에 가장 단순하게 적용할 수 있습니다.

예제 2) 같은 선행조건을 가진 작업은 같은 단계에서 실행 가능

>>> deps = {
...     "extract_users": set(),
...     "extract_orders": set(),
...     "transform_users": {"extract_users"},
...     "transform_orders": {"extract_orders"},
...     "merge_daily": {"transform_users", "transform_orders"},
... }
>>> list(TopologicalSorter(deps).static_order())
['extract_users', 'extract_orders', 'transform_users', 'transform_orders', 'merge_daily']

해설:

  • 앞의 두 작업은 서로 의존성이 없어서 어떤 순서로 나와도 유효합니다.
  • 즉, 정렬 결과는 “유일한 단 하나의 순서”가 아닐 수 있습니다. 테스트에서는 특정 고정 순서만 강제하기보다 제약 만족 여부를 검증하는 쪽이 안전합니다.

예제 3) prepare/get_ready/done으로 병렬 실행 단위 처리

>>> from graphlib import TopologicalSorter
>>> deps = {
...     "A": set(),
...     "B": set(),
...     "C": {"A"},
...     "D": {"A", "B"},
...     "E": {"C", "D"},
... }
>>> ts = TopologicalSorter(deps)
>>> ts.prepare()
>>> first = tuple(sorted(ts.get_ready()))
>>> first
('A', 'B')
>>> ts.done("A")
>>> tuple(sorted(ts.get_ready()))
('C',)
>>> ts.done("B")
>>> tuple(sorted(ts.get_ready()))
('D',)
>>> ts.done("C")
>>> ts.done("D")
>>> tuple(ts.get_ready())
('E',)
>>> ts.done("E")
>>> ts.is_active()
False

해설:

  • get_ready()는 “지금 당장 실행 가능한 작업들”을 반환합니다.
  • 워커 풀(스레드/프로세스/큐)로 분산 실행할 때 done() 호출 타이밍만 정확히 맞추면 의존성 제약을 자동으로 유지할 수 있습니다.

예제 4) 실전형: 단계별 실행 배치 만들기

>>> from graphlib import TopologicalSorter
>>> def build_stages(graph):
...     ts = TopologicalSorter(graph)
...     ts.prepare()
...     stages = []
...     while ts.is_active():
...         ready = sorted(ts.get_ready())
...         if not ready:
...             break
...         stages.append(ready)
...         for job in ready:
...             ts.done(job)
...     return stages
...
>>> graph = {
...     "download": set(),
...     "validate": {"download"},
...     "index": {"validate"},
...     "backup": {"validate"},
...     "notify": {"index", "backup"},
... }
>>> build_stages(graph)
[['download'], ['validate'], ['backup', 'index'], ['notify']]

해설:

  • 이 패턴을 쓰면 “동시 실행 가능한 그룹”을 명시적으로 만들 수 있어 운영 스케줄러와 연결하기 쉽습니다.
  • 특히 CI 파이프라인이나 ETL 배치에서 병렬 구간을 찾는 데 매우 유용합니다.

자주 하는 실수

실수 1) 의존 방향을 반대로 작성

>>> from graphlib import TopologicalSorter
>>> wrong = {
...     "fetch": {"clean"},   # 잘못: fetch가 clean을 의존한다고 적음
...     "clean": {"train"},
...     "train": set(),
... }
>>> list(TopologicalSorter(wrong).static_order())
['train', 'clean', 'fetch']

원인:

  • 값은 “후행 작업 목록”이 아니라 선행 작업 목록인데, 방향을 거꾸로 이해했습니다.

해결:

>>> correct = {
...     "fetch": set(),
...     "clean": {"fetch"},
...     "train": {"clean"},
... }
>>> list(TopologicalSorter(correct).static_order())
['fetch', 'clean', 'train']

실무 팁:

  • 그래프 작성 직후 “A는 B 이후에 실행돼야 한다” 문장을 2~3개 골라 실제 결과와 대조하면 방향 오류를 초기에 잡을 수 있습니다.

실수 2) 순환 의존을 늦게 발견

>>> from graphlib import TopologicalSorter, CycleError
>>> cyclic = {
...     "build": {"test"},
...     "test": {"deploy"},
...     "deploy": {"build"},
... }
>>> try:
...     list(TopologicalSorter(cyclic).static_order())
... except CycleError as e:
...     print(type(e).__name__)
...
CycleError

원인:

  • 기능을 빠르게 붙이면서 의존 규칙 리뷰를 생략하면, 팀 단위 변경에서 순환이 쉽게 생깁니다.

해결:

>>> def validate_graph_or_raise(graph):
...     # 배포 전 검증 단계에서 순환 여부를 즉시 확인
...     list(TopologicalSorter(graph).static_order())
...     return True
...
>>> validate_graph_or_raise({"A": set(), "B": {"A"}})
True

실무 팁:

  • CI 단계에 “의존성 그래프 정렬 가능 여부 검사”를 고정하면 순환이 운영에 들어오기 전에 차단할 수 있습니다.

실수 3) get_ready()만 호출하고 done()을 누락

  • 증상: 루프가 멈추거나 기대한 다음 작업이 절대 READY 상태가 되지 않습니다.
  • 원인: 실행 완료 신호(done)를 안 보냈기 때문에 의존 해제가 일어나지 않습니다.
  • 해결: 워커 성공 콜백에서 done(task)를 호출하는 규칙을 강제하고, 실패 시 재시도/중단 정책을 분리합니다.

실무에서는 이 실수가 특히 치명적입니다. 코드상으로는 “대기”처럼 보여서 장애 인지가 늦어지기 때문입니다. 그래서 운영 로그에 최소한 ready_count, completed_count, pending_count를 주기적으로 남겨야 합니다. 그래야 “실행기가 정지한 건지, 실제로 대기 중인지”를 빠르게 구분할 수 있습니다.

실무 패턴

  • 입력 검증 규칙
    • 모든 노드가 문자열인지, 의존 목록이 집합/리스트인지 먼저 정규화합니다.
    • 정의되지 않은 노드를 의존성에 참조했는지 검사해 오타를 조기에 잡습니다.
  • 로그/예외 처리 규칙
    • 실행 시작 시 전체 노드 수, 엣지 수, 첫 READY 묶음을 기록합니다.
    • CycleError 발생 시 순환 후보 노드를 로그에 남기고 즉시 실패 처리합니다.
  • 재사용 함수/구조화 팁
    • build_stages(graph) 같은 순수 함수를 분리해 테스트를 쉽게 만듭니다.
    • 실제 실행(스레드풀/큐)은 “단계 계산 함수”와 분리하면 교체 비용이 낮아집니다.
  • 성능/메모리 체크 포인트
    • 노드가 수천 개 이상이면 한 번에 전체 작업 객체를 메모리에 올리기보다 ID 중심으로 얇게 유지합니다.
    • 병렬도를 키울수록 외부 I/O 병목(DB, API, 파일락)이 튀어나오므로, 단계별 실행 시간 메트릭을 꼭 수집합니다.

한 가지 더 중요한 점이 있습니다. 위상 정렬은 “순서 문제”를 풀어주지만 “실패 복구 정책”까지 자동으로 해결해 주지는 않습니다. 예를 들어 10개 작업 중 7번째에서 실패했을 때, 1~6을 재실행할지, 실패 노드부터 재개할지, 후속 노드를 스킵할지 정책이 필요합니다. 따라서 TopologicalSorter는 실행 엔진의 중심 부품이지만, 운영 품질은 결국 **재시도 전략, 실패 전파 규칙, 관측 가능성(로그/메트릭)**까지 합쳐졌을 때 완성됩니다.

오늘의 결론

한 줄 요약: TopologicalSorter는 의존성 순서를 안전하게 강제하고, 병렬 실행 가능한 단위를 계산해 자동화 파이프라인의 안정성과 속도를 함께 올려준다.

기억할 것:

  • 값은 “선행 작업 집합”이다. 방향을 헷갈리면 결과가 완전히 뒤틀린다.
  • CycleError 검증을 배포 전에 자동화해야 운영 장애를 줄일 수 있다.
  • 병렬 실행에서는 get_ready()와 done()의 짝을 반드시 지켜야 한다.

연습문제

  1. ingest -> normalize -> aggregate -> export 파이프라인 그래프를 만들고 static_order() 결과를 확인하세요.
  2. A -> B -> C -> A 형태의 순환 그래프를 만들어 CycleError를 재현하고, 예외 메시지를 출력해 보세요.
  3. prepare/get_ready/done을 사용해 단계별 실행 리스트를 반환하는 build_stages(graph) 함수를 직접 구현해 보세요.

이전 강의 정답

  1. UTC 시간을 서울 시간대로 변환하기
>>> from datetime import datetime, timezone
>>> from zoneinfo import ZoneInfo
>>> dt_utc = datetime(2026, 2, 17, 3, 0, tzinfo=timezone.utc)
>>> dt_kr = dt_utc.astimezone(ZoneInfo("Asia/Seoul"))
>>> dt_kr.isoformat()
'2026-02-17T12:00:00+09:00'
  1. naive datetime을 안전하게 timezone-aware로 만들기
>>> from datetime import datetime
>>> from zoneinfo import ZoneInfo
>>> naive = datetime(2026, 6, 1, 9, 30)
>>> aware = naive.replace(tzinfo=ZoneInfo("Asia/Seoul"))
>>> aware.tzinfo.key
'Asia/Seoul'
  1. 여러 타임존 동시 출력 도우미 함수
>>> from datetime import datetime, timezone
>>> from zoneinfo import ZoneInfo
>>> def world_clock(base_utc):
...     zones = ["Asia/Seoul", "Europe/London", "America/New_York"]
...     return {z: base_utc.astimezone(ZoneInfo(z)).strftime("%Y-%m-%d %H:%M") for z in zones}
...
>>> base = datetime(2026, 2, 17, 0, 0, tzinfo=timezone.utc)
>>> world_clock(base)["Asia/Seoul"]
'2026-02-17 09:00'

실습 환경/재현 정보

  • 실행 환경: conda env python100 (Python 3.11.14)
  • 가정한 OS: macOS/Linux 공통
  • 사용 모듈: graphlib (Python 3.9+ 표준 라이브러리)
  • 재현 순서:
    1. REPL에서 static_order() 예제를 실행해 기본 동작 확인
    2. 순환 그래프를 넣어 CycleError 발생 확인
    3. prepare/get_ready/done 흐름으로 단계 실행 로직 확인
  • 검증 체크:
    • 의존 방향(선행 집합) 정의가 일관적인지
    • 완료 신호(done) 누락 없이 호출되는지
    • 오류 발생 시 즉시 실패하고 원인 로그가 남는지