Kubeflow Pipelines v2 ML 워크플로우 자동화와 운영 가이드

# KFP SDK v2 설치
pip install kfp==2.7.0

# Kubernetes 확장 라이브러리 (GPU, Volume 등 K8s 특화 기능)
pip install kfp-kubernetes==1.2.0

# Kubeflow Pipelines 백엔드 배포 (Kubernetes 클러스터에)
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/cluster-scoped-resources?ref=2.2.0"
kubectl wait --for condition=established --timeout=60s crd/applications.app.k8s.io
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/env/platform-agnostic-pns?ref=2.2.0"

# 포트 포워딩으로 UI 접근
kubectl port-forward -n kubeflow svc/ml-pipeline-ui 8080:80

from kfp import dsl
from kfp.dsl import Input, Output, Dataset, Model, Metrics

@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.2.0", "scikit-learn==1.4.0"]
)
def preprocess_data(
    raw_data_path: str,
    test_size: float,
    train_dataset: Output[Dataset],
    test_dataset: Output[Dataset],
    data_stats: Output[Metrics]
):
    """데이터를 로드하고 학습/테스트로 분할한다."""
    import pandas as pd
    from sklearn.model_selection import train_test_split

    df = pd.read_csv(raw_data_path)

    # 결측치 처리
    df = df.dropna(subset=["target"])
    df = df.fillna(df.median(numeric_only=True))

    train_df, test_df = train_test_split(
        df, test_size=test_size, random_state=42, stratify=df["target"]
    )

    train_df.to_csv(train_dataset.path, index=False)
    test_df.to_csv(test_dataset.path, index=False)

    # 메트릭 기록
    data_stats.log_metric("total_rows", len(df))
    data_stats.log_metric("train_rows", len(train_df))
    data_stats.log_metric("test_rows", len(test_df))
    data_stats.log_metric("feature_count", len(df.columns) - 1)


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.2.0", "scikit-learn==1.4.0", "joblib==1.3.0"]
)
def train_model(
    train_dataset: Input[Dataset],
    n_estimators: int,
    max_depth: int,
    trained_model: Output[Model],
    training_metrics: Output[Metrics]
):
    """Random Forest 모델을 학습한다."""
    import pandas as pd
    import joblib
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.metrics import accuracy_score, f1_score

    train_df = pd.read_csv(train_dataset.path)
    X_train = train_df.drop(columns=["target"])
    y_train = train_df["target"]

    model = RandomForestClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        random_state=42,
        n_jobs=-1
    )
    model.fit(X_train, y_train)

    # 학습 정확도 기록
    train_pred = model.predict(X_train)
    training_metrics.log_metric("train_accuracy", accuracy_score(y_train, train_pred))
    training_metrics.log_metric("train_f1", f1_score(y_train, train_pred, average="weighted"))

    # 모델 저장
    joblib.dump(model, trained_model.path)
    trained_model.metadata["framework"] = "sklearn"
    trained_model.metadata["model_type"] = "RandomForestClassifier"


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.2.0", "scikit-learn==1.4.0", "joblib==1.3.0"]
)
def evaluate_model(
    test_dataset: Input[Dataset],
    trained_model: Input[Model],
    eval_metrics: Output[Metrics],
    accuracy_threshold: float = 0.85
) -> bool:
    """테스트 데이터로 모델을 평가하고 임계값을 넘는지 확인한다."""
    import pandas as pd
    import joblib
    from sklearn.metrics import accuracy_score, f1_score, classification_report

    test_df = pd.read_csv(test_dataset.path)
    X_test = test_df.drop(columns=["target"])
    y_test = test_df["target"]

    model = joblib.load(trained_model.path)
    predictions = model.predict(X_test)

    accuracy = accuracy_score(y_test, predictions)
    f1 = f1_score(y_test, predictions, average="weighted")

    eval_metrics.log_metric("test_accuracy", accuracy)
    eval_metrics.log_metric("test_f1", f1)
    eval_metrics.log_metric("passed_threshold", accuracy >= accuracy_threshold)

    return accuracy >= accuracy_threshold

@dsl.pipeline(
    name="ml-training-pipeline",
    description="데이터 전처리 → 학습 → 평가 → 조건부 배포 파이프라인"
)
def ml_training_pipeline(
    raw_data_path: str = "gs://my-bucket/data/raw.csv",
    test_size: float = 0.2,
    n_estimators: int = 100,
    max_depth: int = 10,
    accuracy_threshold: float = 0.85
):
    # 1단계: 데이터 전처리
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=test_size
    )
    preprocess_task.set_display_name("Data Preprocessing")

    # 2단계: 모델 학습
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        n_estimators=n_estimators,
        max_depth=max_depth
    )
    train_task.set_display_name("Model Training")
    train_task.set_cpu_limit("4")
    train_task.set_memory_limit("8Gi")

    # 3단계: 모델 평가
    eval_task = evaluate_model(
        test_dataset=preprocess_task.outputs["test_dataset"],
        trained_model=train_task.outputs["trained_model"],
        accuracy_threshold=accuracy_threshold
    )
    eval_task.set_display_name("Model Evaluation")

    # 4단계: 조건부 배포 (정확도 임계값 통과 시)
    with dsl.If(eval_task.output == True):
        deploy_task = deploy_model(
            model=train_task.outputs["trained_model"],
            model_name="fraud-detector",
            serving_endpoint="https://serving.example.com"
        )
        deploy_task.set_display_name("Model Deployment")


# 파이프라인 컴파일
from kfp import compiler
compiler.Compiler().compile(
    pipeline_func=ml_training_pipeline,
    package_path="ml_training_pipeline.yaml"
)

@dsl.container_component
def run_spark_job(
    input_data: Input[Dataset],
    output_data: Output[Dataset],
    spark_config: str
):
    """Spark 작업을 컨테이너로 실행한다."""
    return dsl.ContainerSpec(
        image="my-registry/spark-processor:3.5",
        command=["spark-submit"],
        args=[
            "--master", "k8s://https://kubernetes.default.svc",
            "--conf", spark_config,
            "--input", input_data.path,
            "--output", output_data.path,
            "/app/etl_job.py"
        ]
    )


# Importer 활용: 외부 모델을 파이프라인에 도입
@dsl.pipeline(name="model-comparison-pipeline")
def comparison_pipeline():
    existing_model = dsl.importer(
        artifact_uri="gs://models/production/v2.1/model.pkl",
        artifact_class=Model,
        reimport=False,
        metadata={"version": "2.1", "framework": "sklearn"}
    )

    # 기존 모델과 새 모델을 비교
    compare_task = compare_models(
        baseline_model=existing_model.output,
        candidate_model=train_task.outputs["trained_model"]
    )

@dsl.pipeline(name="caching-example")
def caching_pipeline(data_path: str, retrain: bool = False):
    # 데이터 전처리는 캐싱 활성화 (동일 데이터면 재사용)
    preprocess_task = preprocess_data(raw_data_path=data_path)
    preprocess_task.set_caching_options(True)

    # 학습은 retrain 플래그에 따라 캐싱 제어
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        n_estimators=200,
        max_depth=15
    )
    if retrain:
        train_task.set_caching_options(False)  # 강제 재학습

    # 외부 API 호출 컴포넌트는 캐싱 비활성화
    deploy_task = deploy_model(model=train_task.outputs["trained_model"])
    deploy_task.set_caching_options(False)  # 항상 새로 실행

# .github/workflows/kfp-deploy.yaml
name: KFP Pipeline CI/CD

on:
  push:
    branches: [main]
    paths:
      - 'pipelines/**'
      - 'components/**'

env:
  KFP_HOST: ${{ secrets.KFP_HOST }}
  KFP_NAMESPACE: kubeflow

jobs:
  validate-and-compile:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install kfp==2.7.0 kfp-kubernetes==1.2.0
          pip install pytest

      - name: Lint pipeline code
        run: |
          pip install ruff
          ruff check pipelines/ components/

      - name: Run unit tests
        run: pytest tests/unit/ -v

      - name: Compile pipeline
        run: |
          python -c "
          from pipelines.training_pipeline import ml_training_pipeline
          from kfp import compiler
          compiler.Compiler().compile(
              pipeline_func=ml_training_pipeline,
              package_path='compiled_pipeline.yaml'
          )
          print('Pipeline compiled successfully')
          "

      - name: Upload compiled pipeline
        uses: actions/upload-artifact@v4
        with:
          name: compiled-pipeline
          path: compiled_pipeline.yaml

  deploy-pipeline:
    needs: validate-and-compile
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4

      - name: Download compiled pipeline
        uses: actions/download-artifact@v4
        with:
          name: compiled-pipeline

      - name: Deploy to KFP
        run: |
          pip install kfp==2.7.0
          python -c "
          from kfp.client import Client

          client = Client(host='${KFP_HOST}')

          # 파이프라인 업로드 또는 업데이트
          try:
              pipeline = client.upload_pipeline(
                  pipeline_package_path='compiled_pipeline.yaml',
                  pipeline_name='ml-training-pipeline',
                  description='Automated ML training pipeline'
              )
              print(f'Pipeline uploaded: {pipeline.pipeline_id}')
          except Exception:
              # 이미 존재하면 새 버전으로 업로드
              pipeline = client.upload_pipeline_version(
                  pipeline_package_path='compiled_pipeline.yaml',
                  pipeline_name='ml-training-pipeline',
                  pipeline_version_name='v${GITHUB_SHA::7}'
              )
              print(f'Pipeline version uploaded: {pipeline.pipeline_version_id}')
          "

from kfp.client import Client

def trigger_pipeline_run(
    host: str,
    pipeline_name: str,
    experiment_name: str,
    params: dict
) -> str:
    """프로그래밍 방식으로 파이프라인 실행을 트리거한다."""
    client = Client(host=host)

    # 실험 생성 또는 조회
    experiment = client.create_experiment(
        name=experiment_name,
        namespace="kubeflow"
    )

    # 파이프라인 조회
    pipelines = client.list_pipelines(filter=f'name="{pipeline_name}"')
    if not pipelines.pipelines:
        raise ValueError(f"Pipeline '{pipeline_name}' not found")

    pipeline_id = pipelines.pipelines[0].pipeline_id

    # 실행 생성
    run = client.run_pipeline(
        experiment_id=experiment.experiment_id,
        job_name=f"{pipeline_name}-{params.get('run_tag', 'manual')}",
        pipeline_id=pipeline_id,
        params=params
    )

    print(f"Run created: {run.run_id}")
    print(f"Monitor at: {host}/#/runs/details/{run.run_id}")
    return run.run_id


# 반복 스케줄 실행 설정
def create_recurring_run(client: Client, pipeline_id: str, experiment_id: str):
    """매일 자정에 파이프라인을 실행하는 스케줄을 생성한다."""
    recurring_run = client.create_recurring_run(
        experiment_id=experiment_id,
        job_name="daily-training",
        pipeline_id=pipeline_id,
        params={"raw_data_path": "gs://data/daily/latest.csv"},
        cron_expression="0 0 * * *",
        max_concurrency=1,
        enabled=True
    )
    return recurring_run

from kfp import dsl
from kfp import kubernetes

@dsl.pipeline(name="gpu-training-pipeline")
def gpu_training_pipeline():
    train_task = train_deep_learning_model(
        dataset_path="gs://data/training",
        epochs=50,
        batch_size=64
    )

    # GPU 리소스 할당
    kubernetes.add_node_selector(
        train_task,
        label_key="accelerator",
        label_value="nvidia-a100"
    )
    kubernetes.add_toleration(
        train_task,
        key="nvidia.com/gpu",
        operator="Exists",
        effect="NoSchedule"
    )
    train_task.set_accelerator_type("nvidia.com/gpu")
    train_task.set_accelerator_limit(2)
    train_task.set_cpu_limit("16")
    train_task.set_memory_limit("64Gi")

    # Secret 마운트 (모델 레지스트리 인증 정보)
    kubernetes.use_secret_as_env(
        train_task,
        secret_name="model-registry-credentials",
        secret_key_to_env={
            "username": "REGISTRY_USERNAME",
            "password": "REGISTRY_PASSWORD"
        }
    )

    # PVC 마운트 (공유 데이터 볼륨)
    kubernetes.mount_pvc(
        train_task,
        pvc_name="shared-data-pvc",
        mount_path="/mnt/shared-data"
    )

# 파이프라인 실행의 전체 로그 확인
kubectl logs -n kubeflow -l pipeline/runid=<run-id> --all-containers

# 특정 컴포넌트의 로그 확인
kubectl logs -n kubeflow <pod-name> -c main

# Argo Workflow 상태 확인
kubectl get workflows -n kubeflow
kubectl describe workflow <workflow-name> -n kubeflow

# ML Metadata 직접 조회 (디버깅용)
kubectl port-forward -n kubeflow svc/metadata-grpc-service 8080:8080