Kubeflow 是一個架構在 Kubernetes 的 MLOps 平台，從開發到部署都擁有各種 solution，包含大家最愛用的 jupyter notebook（？）、支援各種模型訓練 framework 的 Trainer、超參數調整的 Katlib、以及整合各種推論 runtime 的 KServe，當然 Kubeflow 也積極整合各種 LLM 所需的推論架構。
在這個專案中，我自建了一個 kubernetes cluster，並且部署了一套 Kubeflow，並在其中嘗試模型開發、訓練與部署，最後也試著在 in-cluster, out-of-cluster 的環境發送模型的推論請求。
主要用到的 Kubeflow 服務包括：
- Kubeflow Trainer：定義模型，並且在 cluster 裡面做 DDP 的模型訓練
- Kubeflow KServe：部署模型，並且利用 Knative autoscaler 做 autosacling
- Kubeflow Dashboard 與 Notebooks：開發與觀測，包括實驗、監控與協作環境
Github repo: kubeflow-mlops-starter
Step 0: Build Kind Cluster
kind create cluster --name=kubeflow --config kubeflow-example-config.yaml這邊直接用 kind 建立一個 local cluster，建好之後可以搭配 k9s 檢測 cluster 狀態。
Step 1: Deploy Kubeflow Platform
Kubeflow 的 components 稍微多了一點，我這邊是用 Kubeflow Manifests v1.0 的 branch 去做部署，有一些資源耗費較大或暫時用不到的部件，像是 Katib, Kubeflow Pipelines, Spark Operator，在這個專案裡就先沒有裝了。裝好之後，應該會看到 k9s 裡面有一堆 pod。
Step 2: Model Training
Kubeflow 的架構本身就是 model agnostic 的，所以 text, image, tabular data 自然都可以被處理，這邊我們嘗試利用一個廣告相關的資料訓練 CTR prediction model。這個資料集是 TaobaoAd_x1，主要就是利用 user, item features 去預測點擊機率 （binary classification），模型方面我們採用 Google 提出的 DCN v2，這樣子的模型在現代自然是一個很小的模型，但整個預測表現還是不錯的。
在訓練的部分，我們利用 Kubeflow Trainer 進行 torch model distributed training ( Distributed Data Parallel (DDP))，順帶一提，最近 Trainer 好像剛上 v2，某些 function 可能在新版會有不同的寫法。但總之，定義好 dataset 處理、模型、forward 之後就可以硬 train 一波了。
送出 Training job 之後，根據你定義的 worker 數量，k8 會啟動相對應的 pod，並且執行你定義的 training function，至於訓練的結果看是要送去 object storage，還是哪裡都是可以，我這邊就直接放到 kubeflow notebook 開的 PVC 裡面。模型訓練的好壞暫不是這個專案的重點，但 kubeflow 是有提供 tensorboard 的介面來視覺化模型的訓練過程，也方便實時監控。
Step 3: Define Custom Model
為了要讓 model 可以在 cluster 裡面部署，我們會需要準備 model inference 的 runtime，這邊參考了 KServe 的 Deploy Custom Python Serving Runtime with InferenceService。在這個專案裡，我們主要定義 preprocess handler 如何做 data preprocessing，其他細節就麻煩參考 GitHub Repo。
至於 image 的部分，我這邊用了 BuildPacks 去打包 image，雖然是蠻方便的，但我覺得用 docker 包可能會更容易用一點 XD
pack build --builder=heroku/builder:24 ${DOCKER_USER}/dcnv2:v1Step 4: Deploy the Service in Kubernetes
接下來，就可以利用剛包好的 image 搭配 KServe 來讓模型上線，實務上就是寫一個 yaml 來部署 InferenceService：
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
name: dcnv2
namespace: kubeflow-user-example-com
annotations:
sidecar.istio.io/inject: "false"
spec:
predictor:
scaleTarget: 1
scaleMetric: concurrency
maxReplicas: 10
containers:
- name: kserve-container
image: boboru/dcnv2:v1
resources:
requests:
cpu: "100m"
memory: "512Mi"
limits:
cpu: "1"
memory: "1Gi"
env:
- name: PROTOCOL
value: v2
- name: MODEL_PATH
value: /mnt/models/model_weights.pth
- name: ENCODER_PATH
value: /mnt/models/preprocess_metadata.pkl
- name: DENSE_COLS
value: price
- name: SPARSE_COLS
value: userid,cms_segid,cms_group_id,final_gender_code,age_level,pvalue_level,shopping_level,occupation,new_user_class_level,adgroup_id,cate_id,campaign_id,customer,brand,pid,btag
- name: STORAGE_URI
value: pvc://torch-workspace這邊的 STORAGE_URI 是 InferenceService 提供的寫法，他會幫忙 mount torch-workspace 這一個 PVC 到 /mnt/models/ 這個位置，所以我們所訓練的模型權重、資料前處理所需的資料都可以這樣被提供，當然這也可以掛載到 S3, GCS 之類的地方。
而在 Autoscaling 的部分，可以定義在 predictor（如果有額外定義 transformer service 也是一樣的道理）：
predictor:
scaleTarget: 1
scaleMetric: concurrency
maxReplicas: 10這樣就可以直接依據 concurrency 的水位來決定要有幾個 replica，那這邊因為是用 serverless 的方式做部署，所以會用 Knative 的 autoscaling，如果是要用 KEDA 做 autoscaling 或者部署 LLM，可以轉用 raw deployment 的部署方式。
一旦部署完，k9 就可以看到那一個 InferenceService，而 Central Dashboard 上面的 Kserve endpoints 也可以看到：
Step 5: Test Inference Service
Out-of-cluster
假設要在 cluster 外面發出推論請求，會需要帶 token 並且因為這中間有過一層 Ingress，所以 request header 會需要帶 Host：
curl -v \
-H "Host: $SERVICE_HOSTNAME" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-d @$INPUT_PATH \
http://${INGRESS_HOST}:${INGRESS_PORT}/v1/models/$MODEL_NAME:predict
這邊打的是 v1 protocol 的 endpoint，因為 body 比較好寫 XDD，如果是要走 v2 protocal 可以參考 test_infer.py，同時 KServe 也有提供 Python Inference SDK InferenceRESTClient 和 InferenceGRPCClient，這樣就不用透過 request 來發送請求了。
In-cluster
如果是 in-cluster 的話，就不用特別帶 token 了，可以直接打 internal service：
base_url = "http://dcn-v2.kubeflow-user-example-com.svc.cluster.local/v1/models/dcnv2:predict"
response = requests.post(base_url, headers=headers, json={"inputs": v1_request})V1 Protocal
{'predictions': [[0.05610664561390877]]}V2 Protocal
{
"model_name": "dcnv2",
"model_version": None,
"id": "6f509eae-fc74-42a1-9ecb-3803ebd908ac",
"parameters": None,
"outputs": [
{
"name": "output-0",
"shape": [1, 1],
"datatype": "FP32",
"parameters": None,
"data": [0.05610664561390877],
}
],
}
Step 6: Autoscaling
最後，可以做一點 load test 來看 autoscaling 會不會正常運作。這邊我們用 Hey 來發送 request（30 秒、30 個 concurrent workers）：
hey -z 30s -c 30 -m POST -host ${SERVICE_HOSTNAME} \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" \
-D $INPUT_PATH \
http://${INGRESS_HOST}:${INGRESS_PORT}/v1/models/$MODEL_NAME:predict接著就可以看到 Deployment 裡面，從 1 個 pod 長到 10 個 pods (maxReplicas)：
Summary
我認為生產環境重要的還是 Stability & Scalability、Latency & Throughput，Kubeflow 在這邊就提供了一個很好的 end-to-end 平台，從開發到部署再到可觀測性都有一套 cloud-native 的 solution，尤其最近更是積極整合 LLM，讓人非常期待 Kubeflow 未來的發展，與和其他 ecosystem 的整合。
下一個想試的是 Ray 😎
