Kubeflow

from kfp import dsl, compiler
from kfp.dsl import component, OutputPath, InputPath

# 1. 컴포넌트 정의 (각 단계를 컨테이너로)
@component(base_image="python:3.9")
def preprocess_data(input_path: str, output_path: OutputPath()):
    """데이터 전처리 컴포넌트"""
    import pandas as pd
    df = pd.read_csv(input_path)
    df_clean = df.dropna()
    df_clean.to_csv(output_path, index=False)
    print(f"Preprocessed: {len(df)} -> {len(df_clean)} rows")

@component(base_image="pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime")
def train_model(data_path: InputPath(), model_path: OutputPath(), epochs: int = 10):
    """모델 학습 컴포넌트"""
    import torch
    # 학습 로직...
    torch.save(model.state_dict(), model_path)

@component(base_image="python:3.9")
def deploy_model(model_path: InputPath(), endpoint_name: str):
    """KServe로 모델 배포"""
    # KServe InferenceService 생성 로직
    print(f"Deployed to {endpoint_name}")

# 2. 파이프라인 정의
@dsl.pipeline(name="ML Training Pipeline", description="End-to-end ML pipeline")
def ml_pipeline(data_url: str, epochs: int = 20):
    # DAG 형태로 파이프라인 구성
    preprocess_task = preprocess_data(input_path=data_url)
    train_task = train_model(
        data_path=preprocess_task.outputs['output_path'],
        epochs=epochs
    ).set_gpu_limit(1)  # GPU 할당
    deploy_task = deploy_model(
        model_path=train_task.outputs['model_path'],
        endpoint_name="prod-model"
    )

# 3. 컴파일 & 실행
compiler.Compiler().compile(ml_pipeline, 'pipeline.yaml')
# kubectl apply -f pipeline.yaml 또는 Kubeflow UI에서 업로드