Prefect

from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta
import httpx
import pandas as pd

# Task 정의 - 재시도, 캐싱 설정
@task(
    retries=3,
    retry_delay_seconds=60,
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=1)
)
def extract_data(url: str) -> dict:
    """API에서 데이터 추출"""
    logger = get_run_logger()
    logger.info(f"Fetching data from {url}")

    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

@task
def transform_data(raw_data: dict) -> pd.DataFrame:
    """데이터 변환"""
    logger = get_run_logger()
    df = pd.DataFrame(raw_data['results'])

    # 데이터 정제
    df['created_at'] = pd.to_datetime(df['created_at'])
    df['value'] = df['value'].fillna(0)

    logger.info(f"Transformed {len(df)} rows")
    return df

@task(timeout_seconds=300)
def load_data(df: pd.DataFrame, destination: str) -> int:
    """데이터 적재"""
    logger = get_run_logger()

    # 실제로는 DB, S3 등에 저장
    df.to_parquet(destination)
    logger.info(f"Saved {len(df)} rows to {destination}")

    return len(df)

# Flow 정의
@flow(
    name="ETL Pipeline",
    description="Daily data extraction and loading",
    version="1.0.0"
)
def etl_pipeline(
    api_url: str = "https://api.example.com/data",
    output_path: str = "data/output.parquet"
):
    """메인 ETL 워크플로우"""
    logger = get_run_logger()
    logger.info("Starting ETL pipeline")

    # Task 실행 - 의존성 자동 추론
    raw_data = extract_data(api_url)
    transformed_df = transform_data(raw_data)
    rows_loaded = load_data(transformed_df, output_path)

    logger.info(f"Pipeline completed: {rows_loaded} rows processed")
    return rows_loaded

# 로컬 실행
if __name__ == "__main__":
    etl_pipeline()

# 스케줄링 설정 (배포 시)
# from prefect.deployments import Deployment
# from prefect.server.schemas.schedules import CronSchedule
#
# deployment = Deployment.build_from_flow(
#     flow=etl_pipeline,
#     name="daily-etl",
#     schedule=CronSchedule(cron="0 2 * * *"),  # 매일 새벽 2시
#     work_queue_name="default"
# )
# deployment.apply()

# 병렬 처리 (여러 소스 동시 추출)
@flow
def parallel_etl():
    sources = ["https://api1.example.com", "https://api2.example.com"]

    # 병렬 실행
    futures = [extract_data.submit(url) for url in sources]
    results = [f.result() for f in futures]

    return results

# CLI 실행
# prefect deployment run "ETL Pipeline/daily-etl"
# prefect server start  # UI 서버 시작