Tutorial Lengkap Kubeflow: MLOps di Kubernetes

Kubeflow adalah platform open-source untuk deploy, mengelola, dan scaling workflow machine learning di Kubernetes. Platform ini menyediakan solusi MLOps lengkap dengan pipelines, model serving, notebooks, dan experiment tracking.

Mengapa Kubeflow?

• Kubernetes native: Manfaatkan skalabilitas dan reliabilitas K8s
• End-to-end MLOps: Dari eksperimen hingga production
• Portable: Jalankan di cluster Kubernetes manapun
• Composable: Gunakan hanya komponen yang diperlukan
• Open source: Komunitas aktif dan ekosistem luas

• Orkestrasi ML pipeline
• Distributed training
• Model serving skala besar
• Experiment tracking
• Feature engineering

Instalasi

1. Prerequisites

# Install kubectl
curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
chmod +x kubectl && sudo mv kubectl /usr/local/bin/

Install kustomize
curl -s "https://raw.githubusercontent.com/kubernetes-sigs/kustomize/master/hack/installkustomize.sh" | bash

sudo mv kustomize /usr/local/bin/

# Clone manifests
git clone https://github.com/kubeflow/manifests.git
cd manifests

Install dengan kustomize
while ! kustomize build example | kubectl apply -f -; do
    echo "Mencoba ulang..."
    sleep 10
done

Cek instalasi
kubectl get pods -n kubeflow

# Port forward
kubectl port-forward svc/istio-ingressgateway -n istio-system 8080:80

Akses di http://localhost:8080
Kredensial default: user@example.com / 12341234

from kfp import dsl
from kfp import compiler

@dsl.component
def preprocessdata(datapath: str) -> str:

    import pandas as pd
    df = pd.readcsv(datapath)

    df = df.dropna()
    outputpath = "/tmp/preprocessed.csv"
    df.tocsv(outputpath, index=False)
    return outputpath


@dsl.component
def trainmodel(datapath: str, epochs: int) -> str:

    import pickle
    from sklearn.ensemble import RandomForestClassifier
    import pandas as pd

    df = pd.readcsv(datapath)

    X = df.drop("target", axis=1)
    y = df["target"]

    model = RandomForestClassifier(nestimators=100)
    model.fit(X, y)

    modelpath = "/tmp/model.pkl"

    with open(modelpath, "wb") as f:
        pickle.dump(model, f)

    return modelpath


@dsl.component
def evaluatemodel(modelpath: str, testdata: str) -> float:
    import pickle
    import pandas as pd
    from sklearn.metrics import accuracyscore


    with open(modelpath, "rb") as f:
        model = pickle.load(f)

    df = pd.readcsv(testdata)
    X = df.drop("target", axis=1)
    y = df["target"]

    predictions = model.predict(X)
    accuracy = accuracyscore(y, predictions)


    return accuracy

@dsl.pipeline(name="ML Training Pipeline")
def mlpipeline(datapath: str, epochs: int = 10):

    preprocesstask = preprocessdata(datapath=datapath)

    traintask = trainmodel(

        datapath=preprocesstask.output,

        epochs=epochs
    )
    evaluatetask = evaluatemodel(

        modelpath=traintask.output,

        testdata=preprocesstask.output

    )

Compile pipeline
compiler.Compiler().compile(mlpipeline, "pipeline.yaml")

2. Jalankan Pipeline

from kfp.client import Client

Koneksi ke Kubeflow
client = Client(host="http://localhost:8080/pipeline")

Buat experiment
experiment = client.createexperiment("my-experiment")


Jalankan pipeline
run = client.runpipeline(
    experimentid=experiment.id,

    jobname="training-run-1",
    pipelinepackagepath="pipeline.yaml",
    params={"datapath": "gs://bucket/data.csv", "epochs": 20}

)

print(f"Run ID: {run.id}")

from kfp import dsl

@dsl.component
def checkaccuracy(accuracy: float) -> bool:
    return accuracy > 0.9

@dsl.component
def deploymodel(modelpath: str) -> str:
    # Logika deploy model
    return f"Model di-deploy dari {modelpath}"


@dsl.component
def retrainmodel(datapath: str) -> str:

    # Logika retrain
    return "Model dilatih ulang"

@dsl.pipeline(name="Conditional Pipeline")
def conditionalpipeline(datapath: str):

    preprocesstask = preprocessdata(datapath=datapath)

    traintask = trainmodel(datapath=preprocesstask.output, epochs=10)

    evaluatetask = evaluatemodel(

        modelpath=traintask.output,

        testdata=preprocesstask.output

    )

    with dsl.If(evaluatetask.output > 0.9):
        deploymodel(modelpath=traintask.output)


    with dsl.Else():
        retrainmodel(datapath=datapath)

4. Pipeline dengan Loops

from kfp import dsl
from typing import List

@dsl.component
def trainwithparams(learningrate: float, batchsize: int) -> float:
    # Logika training
    return 0.85 + learningrate  0.1


@dsl.pipeline(name="Hyperparameter Search")
def hyperparameterpipeline():
    learningrates = [0.01, 0.1, 0.5]


    with dsl.ParallelFor(learningrates) as lr:
        trainwithparams(learningrate=lr, batchsize=32)

from kfp import dsl

@dsl.containercomponent

def customtraining():
    return dsl.ContainerSpec(
        image="python:3.9",
        command=["python", "-c"],
        args=["""
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
print('Training selesai')
        """]
    )

@dsl.pipeline(name="Container Pipeline")
def containerpipeline():

    customtraining()

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

@dsl.component(
    baseimage="python:3.9",

    packagestoinstall=["pandas", "scikit-learn"]

)
def trainwithartifacts(

    trainingdata: Input[Dataset],
    model: Output[Model],
    metrics: Output[Metrics]
):
    import pandas as pd
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.metrics import accuracyscore

    import pickle

    # Load data
    df = pd.readcsv(trainingdata.path)

    X = df.drop("target", axis=1)
    y = df["target"]

    # Train
    clf = RandomForestClassifier()
    clf.fit(X, y)

    # Simpan model
    with open(model.path, "wb") as f:
        pickle.dump(clf, f)

    # Log metrics
    accuracy = accuracyscore(y, clf.predict(X))
    metrics.logmetric("accuracy", accuracy)

    metrics.logmetric("nsamples", len(df))

from kfp import dsl

@dsl.component(
    baseimage="pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime",
    packagestoinstall=["transformers"]
)
def gputraining(modelname: str, epochs: int) -> str:
    import torch
    from transformers import AutoModelForSequenceClassification, Trainer

    device = torch.device("cuda" if torch.cuda.isavailable() else "cpu")

    print(f"Menggunakan device: {device}")

    # Logika training di sini
    return "trainingselesai"

@dsl.pipeline(name="GPU Pipeline")
def gpupipeline():

    gputraining(modelname="bert-base-uncased", epochs=3).setgpulimit(1)

# inference-service.yaml
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: sklearn-model
  namespace: kubeflow
spec:
  predictor:
    sklearn:
      storageUri: "gs://bucket/models/sklearn"
      resources:
        requests:
          cpu: 100m
          memory: 256Mi
        limits:
          cpu: 1
          memory: 1Gi

kubectl apply -f inference-service.yaml

# predictor.py
import kserve
from typing import Dict
import pickle
import numpy as np

class CustomPredictor(kserve.Model):
    def init(self, name: str):
        super().init(name)
        self.name = name
        self.model = None

    def load(self):
        with open("/mnt/models/model.pkl", "rb") as f:
            self.model = pickle.load(f)
        self.ready = True

    def predict(self, payload: Dict, headers: Dict = None) -> Dict:
        instances = payload["instances"]
        inputs = np.array(instances)
        predictions = self.model.predict(inputs)
        return {"predictions": predictions.tolist()}

if name == "main":
    model = CustomPredictor("custom-model")
    model.load()
    kserve.ModelServer().start([model])

# transformer.py
import kserve
from typing import Dict
import numpy as np

class CustomTransformer(kserve.Model):
    def init(self, name: str, predictorhost: str):
        super().init(name)
        self.predictorhost = predictorhost

    def preprocess(self, inputs: Dict, headers: Dict = None) -> Dict:
        # Preprocess input
        data = inputs["instances"]
        processed = [[float(x) for x in row] for row in data]
        return {"instances": processed}

    def postprocess(self, outputs: Dict, headers: Dict = None) -> Dict:
        # Postprocess output
        predictions = outputs["predictions"]
        labels = ["kelas0", "kelas1"]
        return {
            "predictions": [
                {"label": labels[p], "confidence": 0.9}
                for p in predictions
            ]
        }

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: model-canary
spec:
  predictor:
    canaryTrafficPercent: 20
    sklearn:
      storageUri: "gs://bucket/models/v2"
  # Default predictor (80% traffic)
  default:
    sklearn:
      storageUri: "gs://bucket/models/v1"

# experiment.yaml
apiVersion: kubeflow.org/v1beta1
kind: Experiment
metadata:
  name: hyperparameter-tuning
  namespace: kubeflow
spec:
  objective:
    type: maximize
    goal: 0.99
    objectiveMetricName: accuracy
  algorithm:
    algorithmName: bayesianoptimization
  parallelTrialCount: 3
  maxTrialCount: 12
  maxFailedTrialCount: 3
  parameters:
name: learningrate

      parameterType: double

      feasibleSpace:
        min: "0.001"
        max: "0.1"
name: batchsize
      parameterType: int

      feasibleSpace:
        min: "16"
        max: "128"
  trialTemplate:
    primaryContainerName: training
    trialParameters:
name: learningRate
        reference: learningrate

name: batchSize
        reference: batchsize

    trialSpec:
      apiVersion: batch/v1
      kind: Job
      spec:
        template:
          spec:
            containers:
name: training
                image: training-image:latest

                command:
python
train.py
--learning-rate=${trialParameters.learningRate}
--batch-size=${trialParameters.batchSize}
            restartPolicy: Never

kubectl apply -f experiment.yaml

# train.py
import argparse
import logging

def train(learningrate, batchsize):
    # Logika training
    accuracy = 0.85 + learningrate  0.1
    return accuracy

if name == "main":
    parser = argparse.ArgumentParser()
    parser.addargument("--learning-rate", type=float, default=0.01)
    parser.addargument("--batch-size", type=int, default=32)

    args = parser.parseargs()

    accuracy = train(args.learningrate, args.batchsize)

    # Log metric untuk Katib
    logging.basicConfig(level=logging.INFO)
    logging.info(f"accuracy={accuracy}")

Kubeflow Notebooks

1. Buat Notebook Server

# notebook.yaml
apiVersion: kubeflow.org/v1
kind: Notebook
metadata:
  name: my-notebook
  namespace: kubeflow-user
spec:
  template:
    spec:
      containers:
name: notebook
          image: kubeflownotebookswg/jupyter-pytorch-full:v1.7.0

          resources:
            requests:
              cpu: "1"
              memory: 2Gi
            limits:
              cpu: "2"
              memory: 4Gi
          volumeMounts:
name: workspace
              mountPath: /home/jovyan

      volumes:
name: workspace
          persistentVolumeClaim:

            claimName: notebook-pvc

2. GPU Notebook

apiVersion: kubeflow.org/v1
kind: Notebook
metadata:
  name: gpu-notebook
spec:
  template:
    spec:
      containers:
name: notebook
          image: kubeflownotebookswg/jupyter-pytorch-cuda-full:v1.7.0

          resources:
            limits:
              nvidia.com/gpu: 1

Training Operators

1. PyTorchJob

apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: distributed-training
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      template:
        spec:
          containers:
name: pytorch
              image: pytorch-training:latest

              command:
python
-m
torch.distributed.launch
train.py
              resources:

                limits:
                  nvidia.com/gpu: 1
    Worker:
      replicas: 2
      template:
        spec:
          containers:
name: pytorch
              image: pytorch-training:latest

              resources:
                limits:
                  nvidia.com/gpu: 1

2. TFJob

apiVersion: kubeflow.org/v1
kind: TFJob
metadata:
  name: tf-distributed
spec:
  tfReplicaSpecs:
    Chief:
      replicas: 1
      template:
        spec:
          containers:
name: tensorflow
              image: tensorflow/tensorflow:2.10.0-gpu

              command: ["python", "train.py"]
    Worker:
      replicas: 2
      template:
        spec:
          containers:
name: tensorflow
              image: tensorflow/tensorflow:2.10.0-gpu

Best Practices

1. Best Practices Pipeline

from kfp import dsl

Gunakan caching untuk operasi mahal
@dsl.component
def expensivepreprocessing(data: str) -> str:

    # Hasil ini akan di-cache
    return processeddata

@dsl.pipeline(name="Production Pipeline")
def productionpipeline():

    # Enable caching
    preprocess = expensivepreprocessing(data="input")
    preprocess.setcachingoptions(True)

    # Set retry policy
    train = trainmodel(data=preprocess.output)

    train.setretry(numretries=3, backoffduration="60s")

    # Set resource limits
    train.setcpulimit("2")
    train.setmemorylimit("4Gi")

2. Pattern Model Registry

from kfp import dsl
from kfp.dsl import Model, Output

@dsl.component
def registermodel(

    model: Input[Model],
    modelname: str,
    version: str
) -> str:
    import mlflow

    mlflow.settrackinguri("http://mlflow:5000")

    with mlflow.startrun():

        mlflow.sklearn.logmodel(
            model.path,
            modelname,

            registeredmodelname=f"{modelname}-{version}"
        )

    return f"Model terdaftar: {model_name}-{version}"

Kesimpulan

Kubeflow adalah essential untuk MLOps di Kubernetes dengan:

Key takeaways:

• Gunakan pipelines untuk workflow reproducible
• Manfaatkan KServe untuk production serving
• Enable Katib untuk optimasi hyperparameter
• Gunakan training operators untuk distributed training
• Implementasikan resource management yang proper