Prefect

상위: Data Engineering

요약

Prefect는 Python-native 워크플로우 오케스트레이터이다. "기존 Python 코드에 데코레이터만 붙이면 오케스트레이션이 된다"는 철학을 가진다. Airflow의 DAG 정의 복잡성과 Dagster의 Asset 중심 패러다임 사이에서, 가장 Python스러운 접근을 취한다. Prefect 3.0은 이벤트 드리븐, 트랜잭션 시맨틱스, 멀티모달 오케스트레이션을 지원한다.

핵심 개념

@flow / @task 데코레이터

기존 Python 함수에 데코레이터만 추가하면 된다. Airflow처럼 별도의 DAG 객체를 생성하거나, Dagster처럼 Asset을 정의할 필요가 없다.

from prefect import flow, task

@task(retries=3, retry_delay_seconds=10)
def extract():
    return pd.read_csv("orders.csv")

@task
def transform(df):
    return df.dropna()

@task
def load(df):
    df.to_parquet("cleaned_orders.parquet")

@flow(name="ETL Pipeline")
def etl_pipeline():
    raw = extract()
    cleaned = transform(raw)
    load(cleaned)

핵심: DAG를 명시적으로 정의하지 않는다. Python 함수 호출 순서가 곧 실행 순서이다. if/for 같은 일반 Python 제어 흐름을 자유롭게 사용할 수 있다.

Airflow와의 근본적 차이

동적 워크플로우

Prefect의 가장 큰 장점. Python 제어 흐름을 그대로 사용한다.

@flow
def process_files():
    files = list_files("s3://bucket/incoming/")  # 런타임에 결정

    for file in files:
        if file.endswith(".csv"):
            process_csv(file)
        elif file.endswith(".json"):
            process_json(file)

Airflow에서 같은 작업을 하려면 Dynamic Task Mapping이나 BranchOperator가 필요하다. Prefect는 그냥 Python이다.

이벤트 드리븐 (Prefect 3.0)

Prefect 3.0에서 오픈소스화된 이벤트 엔진. 스케줄 기반뿐 아니라 이벤트 기반 워크플로우를 지원한다.

from prefect import flow
from prefect.events import DeploymentEventTrigger

@flow
def process_upload(file_path: str):
    ...

# S3 파일 업로드 이벤트로 트리거
process_upload.serve(
    triggers=[
        DeploymentEventTrigger(
            expect={"s3.object.created"},
            parameters={"file_path": "{{ event.resource.key }}"}
        )
    ]
)

트랜잭션 시맨틱스 (Prefect 3.0)

태스크를 원자적 단위로 그룹화. 실패 시 롤백 가능.

from prefect import flow, task
from prefect.transactions import transaction

@task
def create_order(order):
    db.insert(order)

@task
def send_notification(order):
    email.send(order.customer_email)

@flow
def order_pipeline(order):
    with transaction():
        create_order(order)       # 실패 시 전체 롤백
        send_notification(order)

아키텍처

Prefect Server (or Prefect Cloud)
├── API Server (REST API + UI)
├── Event Engine (이벤트 처리)
└── Database (SQLite or PostgreSQL)

Execution Layer
├── Work Pool (실행 인프라 추상화)
│   ├── Process Pool (로컬)
│   ├── Docker Pool
│   ├── Kubernetes Pool
│   └── Serverless Pool (AWS Lambda 등)
└── Worker (Work Pool에서 작업 가져와 실행)

• Prefect Server: 셀프호스트 가능 (SQLite 기본). prefect server start 한 줄
• Prefect Cloud: 매니지드 서비스. RBAC, 감사 로그, 알림 등 추가 기능
• Work Pool: 실행 인프라를 추상화. 같은 코드를 로컬/Docker/K8s에서 실행

Hybrid 실행 모델

Prefect Cloud를 사용하더라도 코드와 데이터는 사용자 인프라에서 실행된다. Cloud는 메타데이터(상태, 로그, 이벤트)만 관리한다. Airflow는 코드도 스케줄러 서버에 올려야 한다.

로컬 개발

# 서버 없이 바로 실행 가능
python my_flow.py

# 서버가 필요하면
prefect server start  # 로컬 서버 시작
python my_flow.py     # 서버에 자동 연결

Airflow와 달리 서버 없이도 플로우를 실행할 수 있다. 개발 중에는 그냥 Python 스크립트처럼 실행하고, 프로덕션에서만 서버를 붙인다.

관련 개념

• Airflow: 가장 널리 쓰이는 워크플로우 오케스트레이터
• Dagster: Asset 중심 오케스트레이터
• Dagster vs Prefect vs Airflow: 3대 오케스트레이터 비교