ETL

# Python ETL 파이프라인 예제

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ETLPipeline:
    def __init__(self, source_conn, target_conn):
        self.source_engine = create_engine(source_conn)
        self.target_engine = create_engine(target_conn)

    # 1. EXTRACT - 소스에서 데이터 추출
    def extract(self, table_name: str, incremental_col: str = None,
                last_value = None) -> pd.DataFrame:
        """증분 또는 전체 데이터 추출"""

        if incremental_col and last_value:
            # Incremental Load: 변경된 데이터만
            query = f"""
                SELECT * FROM {table_name}
                WHERE {incremental_col} > '{last_value}'
                ORDER BY {incremental_col}
            """
            logger.info(f"Incremental extract from {table_name} after {last_value}")
        else:
            # Full Load: 전체 데이터
            query = f"SELECT * FROM {table_name}"
            logger.info(f"Full extract from {table_name}")

        df = pd.read_sql(query, self.source_engine)
        logger.info(f"Extracted {len(df)} rows")
        return df

    # 2. TRANSFORM - 데이터 변환
    def transform(self, df: pd.DataFrame) -> pd.DataFrame:
        """데이터 정제 및 변환"""

        # 2-1. 데이터 정제
        df = df.drop_duplicates()
        df = df.dropna(subset=['user_id', 'event_type'])  # 필수 필드 null 제거

        # 2-2. 타입 변환
        df['created_at'] = pd.to_datetime(df['created_at'])
        df['amount'] = pd.to_numeric(df['amount'], errors='coerce').fillna(0)

        # 2-3. 파생 컬럼 생성
        df['event_date'] = df['created_at'].dt.date
        df['event_hour'] = df['created_at'].dt.hour
        df['is_weekend'] = df['created_at'].dt.dayofweek >= 5

        # 2-4. 비즈니스 로직 적용
        df['amount_category'] = pd.cut(
            df['amount'],
            bins=[0, 100, 500, float('inf')],
            labels=['small', 'medium', 'large']
        )

        # 2-5. 메타데이터 추가
        df['etl_timestamp'] = datetime.now()
        df['etl_batch_id'] = datetime.now().strftime('%Y%m%d%H%M%S')

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

    # 3. LOAD - 목적지에 적재
    def load(self, df: pd.DataFrame, table_name: str,
             if_exists: str = 'append') -> None:
        """데이터 적재 (append 또는 replace)"""

        df.to_sql(
            table_name,
            self.target_engine,
            if_exists=if_exists,
            index=False,
            method='multi',  # 배치 insert로 성능 향상
            chunksize=1000
        )
        logger.info(f"Loaded {len(df)} rows to {table_name}")

    # ETL 실행
    def run(self, source_table: str, target_table: str,
            incremental_col: str = None, last_value = None):
        """ETL 파이프라인 실행"""

        logger.info(f"Starting ETL: {source_table} -> {target_table}")

        # Extract
        raw_data = self.extract(source_table, incremental_col, last_value)

        if raw_data.empty:
            logger.info("No new data to process")
            return

        # Transform
        transformed_data = self.transform(raw_data)

        # Load
        self.load(transformed_data, target_table)

        logger.info("ETL completed successfully")
        return transformed_data[incremental_col].max() if incremental_col else None


# 사용 예시
if __name__ == "__main__":
    pipeline = ETLPipeline(
        source_conn="postgresql://source_db/production",
        target_conn="postgresql://target_db/analytics"
    )

    # 증분 ETL 실행
    last_processed = pipeline.run(
        source_table="events",
        target_table="fact_events",
        incremental_col="created_at",
        last_value="2024-01-15 00:00:00"
    )