Prefect - 오케스트레이션

오케스트레이션이란

오케스트레이션(Orchestration)은 여러 작업을 순서대로, 조건에 맞게, 안정적으로 실행하는 것을 말한다.

오케스트라에서 지휘자가 각 파트(바이올린, 첼로, 플루트)를 박자에 맞춰 조율하는 것과 같다. 각 연주자(작업)는 자기 역할만 하고, 지휘자(오케스트레이터)가 전체 흐름을 책임진다.

데이터 파이프라인으로 예를 들면, 아래 같은 상황이 생긴다.

1. 외부 API에서 데이터 수집
2. 수집한 데이터 정규화
3. 중복 제거 후 DB 저장
4. 저장 완료되면 슬랙 알림

이걸 그냥 순서대로 실행하는 Python 스크립트를 짜면 되지 않냐고 할 수 있다. 근데 실제로는 문제가 생긴다.

• 2번에서 실패하면 1번부터 다시 해야 하나?
• 어느 단계에서 실패한 건지 로그 뒤져야 알 수 있음
• 매일 자동 실행하려면 cron으로 따로 설정해야 함
• 팀원이 "어제 파이프라인 돌았어?" 물어보면 확인할 방법이 없음

오케스트레이션 툴은 이런 문제를 해결한다. 각 단계를 추적하고, 실패하면 재시도하고, 실행 이력을 UI로 보여준다.

---

Docker랑 뭐가 다른가

둘 다 "배포/실행"과 관련된 도구라서 헷갈리기 쉬운데, 역할이 완전히 다르다.

Docker는 실행 환경을 격리한다. "이 코드를 어디서든 동일하게 실행할 수 있는 환경"을 만드는 게 목적이다. Python 버전, 패키지, OS 설정을 컨테이너로 묶어서 어떤 서버에서도 똑같이 동작하게 한다.

Prefect는 작업의 흐름을 관리한다. "이 작업들을 언제, 어떤 순서로, 실패하면 어떻게 처리할지"를 관리하는 게 목적이다. 코드가 어디서 실행되는지는 상관없다.

비유하면 이렇다.

역할 비유

Docker	실행 환경 격리	주방 (어디서든 동일한 조리 환경)
Prefect	작업 흐름 관리	주방장 (레시피 순서대로 요리 지시)

실제로 둘은 같이 쓰인다. Docker로 실행 환경을 컨테이너로 만들고, Prefect로 그 컨테이너를 스케줄에 맞춰 실행하고 결과를 추적하는 식이다.

---

Prefect를 알아본 이유

오케스트레이션 툴 하면 Apache Airflow가 먼저 떠오른다. 근데 Airflow는 DAG(방향 비순환 그래프)라는 독특한 방식으로 파이프라인을 정의해야 하고, 로컬 셋업도 복잡하다.

Prefect는 다르다. 기존 Python 함수에 데코레이터만 붙이면 파이프라인이 된다. Prefect Cloud 없이도 로컬 서버(prefect server start)만으로 UI와 실행 이력을 볼 수 있다.

---

핵심 개념 — @flow와 @task

Prefect의 구조는 단순하다.

• @task: 파이프라인의 단위 작업. 재시도, 캐싱, 실패 추적의 단위가 된다.
• @flow: task들을 묶는 오케스트레이터. 전체 실행을 관리하고 UI에서 하나의 실행 단위로 보인다.

예시로 쓸 파이프라인 구조는 이렇다.

main_flow()
  └── source_flow(source_name)    # 소스별 수집 흐름
        ├── collect_all()         # 전체 페이지 수집
        ├── process_items()       # 정규화 → 검증 → 중복 제거
        └── save_items()          # DB 저장 + 이력 기록

---

설치

pip install prefect
# Poetry 쓴다면
poetry add prefect

---

@task 정의

# pipeline/tasks/collect.py
from prefect import task, get_run_logger


@task(name="collect_all", retries=2, retry_delay_seconds=30)
def collect_all(source_url: str) -> list[dict]:
    """외부 API에서 전체 페이지 데이터를 수집한다."""
    logger = get_run_logger()

    all_items = []
    page = 1

    while True:
        response = fetch_page(source_url, page)
        if not response:
            break
        all_items.extend(response)
        logger.info(f"페이지 {page} 수집 완료 ({len(response)}건)")
        page += 1

    logger.info(f"수집 완료: 총 {len(all_items)}건")
    return all_items

@task 데코레이터의 주요 파라미터:

• retries=2: 실패 시 최대 2회 재시도
• retry_delay_seconds=30: 재시도 간격 30초
• name: Prefect UI에서 보이는 task 이름

get_run_logger()는 Prefect가 제공하는 로거다. 일반 logging과 달리 Prefect UI에서 실행별 로그를 바로 볼 수 있다.

---

@task 체이닝

# pipeline/tasks/process.py
from prefect import task, get_run_logger


@task(name="process_items")
def process_items(raw_items: list[dict]) -> list[dict]:
    """raw → 정규화 → 검증 → 중복 제거."""
    logger = get_run_logger()

    processed = []
    for item in raw_items:
        try:
            normalized = normalize(item)
            if not validate(normalized):
                logger.warning(f"검증 실패 스킵: {item.get('title', '')[:30]}")
                continue
            processed.append(normalized)
        except Exception as e:
            logger.error(f"처리 실패: {e}")
            continue  # 한 건 실패가 전체 배치에 영향 없도록

    deduped = deduplicate(processed)
    logger.info(f"처리 완료: {len(raw_items)}건 → {len(deduped)}건 (중복 제거 후)")
    return deduped

각 단계에서 예외를 잡아 아이템 단위로 스킵하는 게 중요하다. 한 건의 파싱 실패로 전체 배치가 죽으면 안 되기 때문이다.

---

@flow 정의

# pipeline/flows/source_flow.py
from prefect import flow, get_run_logger
from pipeline.tasks.collect import collect_all
from pipeline.tasks.process import process_items
from pipeline.tasks.save import save_items


@flow(name="source_flow", log_prints=True)
def source_flow(source_name: str, source_url: str) -> None:
    """단일 소스의 수집 → 처리 → 저장 파이프라인."""
    logger = get_run_logger()
    logger.info(f"[{source_name}] 파이프라인 시작")

    raw_items = collect_all(source_url)
    processed = process_items(raw_items)
    save_items(processed, source_name)

    logger.info(f"[{source_name}] 완료")

# pipeline/flows/main_flow.py
from prefect import flow, get_run_logger

SOURCES = [
    {"name": "source_a", "url": "https://api.source-a.com/posts"},
    {"name": "source_b", "url": "https://api.source-b.com/posts"},
    {"name": "source_c", "url": "https://api.source-c.com/posts"},
]


@flow(name="main_flow")
def main_flow() -> None:
    """전체 소스를 순차 실행한다."""
    logger = get_run_logger()

    for source in SOURCES:
        try:
            source_flow(source["name"], source["url"])
        except Exception as e:
            logger.error(f"[{source['name']}] 실패, 다음 소스로: {e}")
            continue  # 한 소스 실패해도 나머지 계속 진행

@flow 안에서 다른 @flow를 호출하면 subflow가 된다. Prefect UI에서 main_flow 실행 안에 소스별 source_flow 실행이 계층 구조로 표시된다. 어느 소스에서 실패했는지 바로 파악된다.

---

로컬 실행

# main.py
from pipeline.flows.main_flow import main_flow

if __name__ == "__main__":
    main_flow()

python main.py

Prefect는 별도 서버 없이도 로컬에서 바로 실행된다. UI가 필요하면:

# Prefect 로컬 서버 실행 (별도 터미널)
prefect server start

# 실행 → http://localhost:4200 에서 UI 확인
python main.py

UI에서 각 flow/task의 실행 상태, 소요 시간, 로그를 한 화면에서 볼 수 있다.

---

@task vs 일반 함수 — 언제 분리할까

모든 함수를 @task로 만들 필요는 없다. task는 실행 추적, 재시도, 캐싱의 단위다. 그 기능이 필요 없는 순수한 로직은 그냥 일반 함수로 두는 게 낫다.

@task로 분리할 것                                                                                           일반 함수로 둘 것

외부 API 호출 (실패 가능성 있음)	정규화 로직 (순수 함수)
DB 저장 (트랜잭션 단위)	데이터 변환
시간이 오래 걸리는 작업	검증 로직

정규화, 검증, 변환은 @task로 안 만들고 일반 Python 함수로 둔다. process_items task 안에서 호출되는 내부 로직이라고 보면 된다.

---

에러 핸들링 전략

파이프라인에서 에러 핸들링의 핵심은 어느 수준에서 실패를 허용할지 결정하는 것이다.

• 아이템 단위 실패: process_items 안에서 try/except로 스킵. 한 건 실패가 배치 전체에 영향 없음.
• 소스 단위 실패: source_flow 수준에서 실패하면 retries=2로 재시도 후 포기. main_flow는 다음 소스로 넘어감.
• 전체 파이프라인 실패: Prefect UI에서 FAILED로 표시. 알림도 설정 가능.

계층을 명확히 나눠두면 UI에서 어디서 실패했는지 한눈에 보인다.

---

정리

Prefect 핵심은 Python 함수를 최소한의 변경으로 파이프라인으로 만드는 것이다.

• 오케스트레이션 — 작업의 순서, 실패 처리, 재시도, 이력 추적을 한 곳에서 관리하는 것
• Docker와의 차이 — Docker는 실행 환경 격리, Prefect는 작업 흐름 관리. 역할이 다르고 같이 쓰인다.
• @task: 재시도와 실패 추적의 단위
• @flow: task들의 오케스트레이터. subflow로 계층 구조를 만들 수 있다.
• get_run_logger(): Prefect UI와 연동되는 로거
• 모든 함수를 task로 만들 필요는 없다. 순수 함수는 그냥 함수로 두면 된다.