Prefect — Pythonic 대안 · Data Engineering Quest

라이브러리처럼 느껴지는 orchestration

Prefect (2026.8 기준 3.8) 가 갈고닦은 건 한 가지야: "Python 처럼 느껴져야지, platform 처럼 말고." Flow 는 decorate 된 Python 함수고, task 도 decorate 된 Python 함수야. python flow.py 로 로컬에서 돌리고, 같은 코드를 한 줄도 안 고치고 Prefect Cloud 나 self-hosted 서버에 배포할 수 있어.

Prefect 가 힘주는 것

Native async/await. Task 가 coroutine 이 될 수 있어. 동시성이 뒷문이 아니라 정문으로 들어와.
Hybrid 실행. Prefect Cloud 가 orchestrate 하고, 본인 코드는 본인 인프라에서 돌아. 시크릿이랑 데이터가 본인 네트워크를 안 떠난다는 뜻이야.
작은 프로젝트엔 격식이 거의 없어. Python 파일 하나가 그대로 배포 가능한 flow 야.
Stateful flow. Task 끼리 result 로 대화하고, retry 땐 자동으로 cache 에서 replay 돼.

Code

Prefect flow — Pythonic, async-friendly, 그대로 배포 가능·python

import pandas as pd
from prefect import flow, task, get_run_logger

@task(retries=3, retry_delay_seconds=30)
def extract(ds: str) -> pd.DataFrame:
    log = get_run_logger()
    df = pd.read_csv(f'source/orders_{ds}.csv', dtype=str)
    log.info('extracted %d rows', len(df))
    return df

@task
def transform(raw: pd.DataFrame) -> pd.DataFrame:
    return (raw
        .assign(amount_usd=lambda d: pd.to_numeric(d['amount_usd']))
        .dropna(subset=['amount_usd']))

@task
def load(clean: pd.DataFrame, ds: str) -> str:
    path = f'warehouse/orders/date={ds}/part-0.parquet'
    clean.to_parquet(path, index=False)
    return path

@flow(name='orders_pipeline')
def orders_pipeline(ds: str = '2026-04-30') -> str:
    raw = extract(ds)
    clean = transform(raw)
    return load(clean, ds)

if __name__ == '__main__':
    orders_pipeline('2026-04-30')

Exercise

새 venv 에서 pip install prefect. 위 orders flow 를 복사해 python flow.py 로 실행해 — 로컬 실행이야. 그다음 다른 터미널에서 prefect server start 를 띄우고 localhost:4200 에서 UI 를 연 채 다시 실행해 봐 — 같은 코드가 이제 UI 에 잡혀. 핵심은 이거야: 배포는 대부분 설정 하나 젖히는 일이지, rewrite 가 아니라는 것.

Prefect — Pythonic 대안

라이브러리처럼 느껴지는 orchestration

Prefect 가 힘주는 것

Code

External links

Exercise

Progress

댓글 0