Kubeflow 2026: 在 Kubernetes 上运行完整 ML 流水线 — 从训练到生产部署指南

# 高层组件视图
kubectl get pods -n kubeflow
# 预期输出显示以下 Pod：
# - 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

# 通过 SDK 上传到 KFP
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 挂载到 Notebook 服务器
# 或在流水线组件中引用：
# 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 检测)