Kubeflow 2026: Kubernetes에서 완전한 ML 파이프라인 실행 — 훈련부터 프로덕션 배포 가이드

# 상위 수준 컴포넌트 뷰
kubectl get pods -n kubeflow
# 예상 출력 표시:
# - ml-pipeline (KFP API 서버)
# - katib-controller, katib-db-manager
# - kserve-controller-manager
# - training-operator
# - centraldashboard
# - kubeflow-user-example-com 네임스페이스의 notebooks

# manifests 저장소 클론
export KUBEFLOW_VERSION=v1.10.0
git clone https://github.com/kubeflow/manifests.git
cd manifests

# 릴리스 태그 체크아웃
git checkout ${KUBEFLOW_VERSION}

# 단일 kustomize build로 모든 컴포넌트 설치
while ! kustomize build example | kubectl apply -f -; do
  echo "Retrying to apply resources..."
  sleep 10
done

# 핵심 컴포넌트가 실행 중인지 확인
kubectl get pods -n kubeflow --watch
# 모든 Pod이 Running 또는 Completed로 표시될 때까지 대기
# 3노드 클러스터에서 일반적으로 5-10분 소요

# Central Dashboard에 접근하기 위한 포트 포워딩
kubectl port-forward svc/istio-ingressgateway -n istio-system 8080:80

# http://localhost:8080에서 접근
# 기본 자격증명: user@example.com / 12341234

# Kubeflow Helm 저장소 추가 (커뮤니티 유지관리)
helm repo add kubeflow https://kubeflow.github.io/manifests/
helm repo update

# 최소 프로파일로 설치
helm install kubeflow kubeflow/kubeflow \
  --namespace kubeflow \
  --create-namespace \
  --set pipeline.objectStore.minio.persistence.enabled=true

# doctl 설치 및 인증
doctl kubernetes cluster create kubeflow-ml \
  --region nyc3 \
  --node-pool "name=cpu-pool;size=s-4vcpu-8gb;n-node=3" \
  --node-pool "name=gpu-pool;size=gpu-h100-1vcpu-8gb;n-node=2"

# 그런 다음 옵션 A와 같이 Kubeflow manifests 적용

# Kubeflow가 생성한 모든 네임스페이스 확인
kubectl get namespaces | grep kubeflow
# kubeflow          Active
# kubeflow-user-example-com  Active

# pipeline.py — KFP SDK v2를 사용한 완전한 ML 파이프라인
import kfp
from kfp import dsl
from kfp.dsl import component, Input, Output, Dataset, Model, Metrics

@component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas", "scikit-learn"]
)
def download_data(output_dataset: Output[Dataset]):
    """Download and preprocess the dataset."""
    import pandas as pd
    from sklearn.datasets import load_iris
    from sklearn.model_selection import train_test_split

    iris = load_iris(as_frame=True)
    df = iris.frame
    train, test = train_test_split(df, test_size=0.2, random_state=42)
    train.to_csv(f"{output_dataset.path}.csv", index=False)

@component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas", "scikit-learn", "joblib"]
)
def train_model(
    input_dataset: Input[Dataset],
    output_model: Output[Model],
    n_estimators: int = 100
):
    """Train a Random Forest classifier."""
    import pandas as pd
    import joblib
    from sklearn.ensemble import RandomForestClassifier

    df = pd.read_csv(f"{input_dataset.path}.csv")
    X = df.drop("target", axis=1)
    y = df["target"]

    clf = RandomForestClassifier(
        n_estimators=n_estimators,
        random_state=42
    )
    clf.fit(X, y)
    joblib.dump(clf, f"{output_model.path}.joblib")

@component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas", "scikit-learn", "joblib"]
)
def evaluate_model(
    input_model: Input[Model],
    input_dataset: Input[Dataset],
    metrics: Output[Metrics]
) -> str:
    """Evaluate the trained model and log metrics."""
    import pandas as pd
    import joblib
    from sklearn.metrics import accuracy_score, f1_score

    df = pd.read_csv(f"{input_dataset.path}.csv")
    X = df.drop("target", axis=1)
    y = df["target"]

    clf = joblib.load(f"{input_model.path}.joblib")
    predictions = clf.predict(X)

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

    metrics.log_metric("accuracy", accuracy)
    metrics.log_metric("f1_score", f1)

    return f"Model accuracy: {accuracy:.4f}, F1: {f1:.4f}"

@dsl.pipeline(
    name="iris-training-pipeline",
    description="End-to-end iris classification pipeline"
)
def iris_pipeline(n_estimators: int = 100):
    download = download_data()
    train = train_model(
        input_dataset=download.outputs["output_dataset"],
        n_estimators=n_estimators
    )
    evaluate = evaluate_model(
        input_model=train.outputs["output_model"],
        input_dataset=download.outputs["output_dataset"]
    )

# 파이프라인 컴파일
if __name__ == "__main__":
    kfp.compiler.Compiler().compile(
        iris_pipeline,
        "iris_pipeline.yaml"
    )

# 파이프라인 컴파일 및 업로드
python pipeline.py

# KFP SDK를 통해 업로드
kfp pipeline create \
  --pipeline-name iris-classifier \
  --description "Iris classification training pipeline" \
  --engine argo \
  iris_pipeline.yaml

# CLI에서 파이프라인 실행
kfp run create \
  --experiment-name default \
  --pipeline-id <PIPELINE_ID> \
  --display-name "iris-run-$(date +%s)"

# pytorch-job.yaml — 분산 PyTorch 훈련
apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: cifar10-distributed
  namespace: kubeflow-user-example-com
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      restartPolicy: OnFailure
      template:
        spec:
          containers:
          - name: pytorch
            image: my-registry/cifar10-training:v1.2
            command: ["python", "-m", "torch.distributed.launch",
                      "--nproc_per_node=1", "train.py"]
            resources:
              limits:
                nvidia.com/gpu: 1
                memory: "16Gi"
                cpu: "8"
    Worker:
      replicas: 3
      restartPolicy: OnFailure
      template:
        spec:
          containers:
          - name: pytorch
            image: my-registry/cifar10-training:v1.2
            command: ["python", "-m", "torch.distributed.launch",
                      "--nproc_per_node=1", "train.py"]
            resources:
              limits:
                nvidia.com/gpu: 1
                memory: "16Gi"
                cpu: "8"

# 훈련 작업 제출
kubectl apply -f pytorch-job.yaml

# 훈련 진행 상황 모니터링
kubectl get pytorchjobs -n kubeflow-user-example-com -w
kubectl logs -f cifar10-distributed-master-0 \
  -n kubeflow-user-example-com

# 클러스터 전체 GPU 활용도 확인
kubectl top nodes
nvidia-smi  # 모든 GPU Pod 낶부에서 실행

# inference-service.yaml — 훈련된 모델 배포
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: iris-classifier
  namespace: kubeflow-user-example-com
  annotations:
    serving.kserve.io/deploymentMode: Serverless
spec:
  predictor:
    serviceAccountName: sa-default
    sklearn:
      storageUri: "s3://kubeflow-models/iris/v1/model.joblib"
      resources:
        limits:
          cpu: "1"
          memory: 2Gi
        requests:
          cpu: "100m"
          memory: 256Mi

# InferenceService 적용
kubectl apply -f inference-service.yaml

# 모델이 준비될 때까지 대기 (제로에서 스케일)
kubectl get inferenceservices -n kubeflow-user-example-com -w

# 예상: iris-classifier   True    100   http://iris-classifier...   Ready

# 배포된 모델 테스트
curl -X POST http://iris-classifier.kubeflow-user-example-com.example.com/v1/models/iris-classifier:predict \
  -H "Content-Type: application/json" \
  -d '{"instances": [[5.1, 3.5, 1.4, 0.2]]}'

# 응답: {"predictions": [0]}

# canary-rollout.yaml — v2의 점진적 롤아웃
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: iris-classifier
  namespace: kubeflow-user-example-com
spec:
  predictor:
    canaryTrafficPercent: 20
    sklearn:
      storageUri: "s3://kubeflow-models/iris/v2/model.joblib"

# katib-experiment.yaml — Random Forest 하이퍼파라미터 최적화
apiVersion: kubeflow.org/v1beta1
kind: Experiment
metadata:
  namespace: kubeflow-user-example-com
  name: iris-hp-tuning
spec:
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: accuracy
  algorithm:
    algorithmName: bayesianoptimization
  parallelTrialCount: 3
  maxTrialCount: 12
  maxFailedTrialCount: 3
  parameters:
    - name: n_estimators
      parameterType: int
      feasibleSpace:
        min: "50"
        max: "500"
    - name: max_depth
      parameterType: int
      feasibleSpace:
        min: "3"
        max: "20"
    - name: min_samples_split
      parameterType: double
      feasibleSpace:
        min: "0.01"
        max: "0.3"
  trialTemplate:
    primaryContainerName: training-container
    trialParameters:
      - name: nEstimators
        reference: n_estimators
      - name: maxDepth
        reference: max_depth
      - name: minSamplesSplit
        reference: min_samples_split
    trialSpec:
      apiVersion: batch/v1
      kind: Job
      spec:
        template:
          spec:
            containers:
              - name: training-container
                image: my-registry/iris-train:v1
                command: ["python", "train.py"]
                resources:
                  limits:
                    memory: "4Gi"
                    cpu: "2"
            restartPolicy: Never

# 실험 시작
kubectl apply -f katib-experiment.yaml

# trial 모니터링
kubectl get trials -n kubeflow-user-example-com
# 12개의 trial과 해당 목표 메트릭 값 표시

# 최적의 trial 보기
kubectl get experiment iris-hp-tuning \
  -n kubeflow-user-example-com \
  -o jsonpath='{.status.currentOptimalTrial}'

# gpu-quota.yaml — 네임스페이스별 GPU 제한 강제
apiVersion: v1
kind: ResourceQuota
metadata:
  name: gpu-quota
  namespace: data-science-team
spec:
  hard:
    requests.nvidia.com/gpu: 8
    limits.nvidia.com/gpu: 16

# 쿼터 적용
kubectl apply -f gpu-quota.yaml

# 네임스페이스별 GPU 할당 확인
kubectl describe resourcequota gpu-quota -n data-science-team

# dataset-pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: training-datasets
  namespace: kubeflow-user-example-com
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 500Gi
  storageClassName: nfs-client  # 또는 AWS에서 efs-sc

# Kubeflow UI를 통해 노트북 서버에 마운트
# 또는 파이프라인 컴포넌트에서 참조:
# dsl.VolumeOp(name="create-dataset-volume",
#              resource_name="training-datasets",
#              size="500Gi",
#              modes=dsl.VOLUME_MODE_RWM)

# 리소스 제한이 있는 사용자 프로필 생성
kubectl apply -f - <<EOF
apiVersion: kubeflow.org/v1
kind: Profile
metadata:
  name: team-ml-platform
spec:
  owner:
    kind: User
    name: ml-engineer@company.com
  resourceQuotaSpec:
    hard:
      cpu: "64"
      memory: 256Gi
      nvidia.com/gpu: "8"
      pods: "50"
EOF

# MySQL 메타데이터 데이터베이스 백업 (KFP 실험/실행)
kubectl exec -it ml-pipeline-mysql-0 -n kubeflow -- \
  mysqldump -u root -p$mysqlpassword mlpipeline \
  > kubeflow-metadata-backup.sql

# MinIO 아티팩트 저장소 백업
mc mirror myminio/kubeflow-pipelines/ \
  s3-backup/kubeflow-pipelines-backup/

# Kubeflow는 여러 컴포넌트에서 Prometheus 메트릭을 노출한다
kubectl apply -f \
  https://raw.githubusercontent.com/kubeflow/manifests/v1.10.0/contrib/prometheus/kustomization.yaml

# 알림을 설정해야 할 핵심 메트릭:
# - kubeflow_pipelines_run_count (총 파이프라인 실행 수)
# - kubeflow_pipelines_run_latency_seconds (파이프라인 실행 시간)
# - nvidia_gpu_utilization_gpu (Pod별 GPU 활용도)
# - container_memory_working_set_bytes (OOM 감지)