MLflow 與 Kubeflow

– 14 分鐘閱讀

在機器學習專案開發過程中，有效管理模型訓練實驗、比較不同引數設定的效能，以及追蹤模型中繼資料是成功的關鍵因素。MLflow作為一個強大的開放原始碼平台，能夠與Kubeflow共同工作，為機器學習工程師提供全方位的模型生命週期管理解決方案。

MLflow的程式化中繼資料存取

類別似於Kubeflow Metadata，MLflow提供了豐富的API讓開發者能夠程式化地存取和管理實驗中繼資料。以下是MLflow主要API的使用範例：

# 取得特定實驗的所有執行記錄
df_runs = mlflow.search_runs(experiment_ids="0")
print("已完成的執行數量: ", len(df_runs))

# 根據均方根誤差(RMSE)指標對執行結果進行排序
df_runs.sort_values(["metrics.rmse"], ascending=True, inplace=True)
df_runs.head()

這段程式碼展示了MLflow的核心功能之一：程式化查詢實驗結果。search_runs()函式允許我們檢索特定實驗ID的所有執行記錄，回傳的DataFrame包含了每次執行的引數、指標和標籤等完整訊息。當我們需要分析多次實驗結果時，可以使用Pandas的強大功能對資料進行排序和過濾，例如上面的程式碼按RMSE指標升序排列，幫助我們快速識別表現最佳的模型設定。

MLflow的程式化查詢能力非常靈活，除了基本的實驗ID查詢外，還支援複雜的條件過濾、時間範圍篩選和自定義查詢邏輯。這使得自動化分析和報告生成變得簡單高效。

值得注意的是，雖然程式化查詢功能強大，但MLflow的網頁介面提供了更直觀的方式來評估和比較實驗結果，這也是我們接下來要探討的重點。

MLflow UI：直觀的實驗管理平台

MLflow的追蹤UI是一個強大的視覺化工具，允許使用者直觀地檢視、搜尋和比較不同的實驗執行，還可以下載執行產物或中繼資料供其他工具分析。由於MLflow不是Kubeflow的原生元件，其存取不是透過Kubeflow UI提供的。根據設定的虛擬服務，MLflow UI通常可以透過<Kubeflow Istio入口閘道器URL>/mlflow進行存取。

MLflow首頁面功能

MLflow的首頁面提供了強大的過濾和搜尋功能。例如，如果我們只想檢視KNN模型的結果，可以使用搜尋條件tags.model="knn"。也可以使用更複雜的過濾器，如tags.model="knn" and metrics.duration_prediction < 0.002，這將回傳預測時間少於0.002秒的KNN模型結果。

實驗詳情與比較檢視

點選單個執行記錄可以檢視其詳細訊息，包括使用的引數、記錄的指標和產生的輸出檔案。這對於深入瞭解特定模型設定的表現至關重要。

MLflow的一大優勢是能夠直觀地比較多個實驗執行。透過選擇多個執行並點選比較按鈕，我們可以並排檢視不同設定的引數和效能指標。這種視覺化比較使得識別最佳模型設定變得簡單明瞭。

特別是對於指標比較，MLflow提供了時間序列檢視，讓我們能夠觀察不同模型在訓練過程中各項指標的變化趨勢。這對於診斷模型收斂問題或比較學習速率非常有用。

為何MLflow UI優於Kubeflow中繼資料工具？

雖然Kubeflow自帶中繼資料管理功能，但MLflow UI提供了更豐富的視覺化選項和更靈活的比較功能。在實際專案中，玄貓發現MLflow的直觀介面能顯著提高團隊分析實驗結果的效率，尤其是在需要比較大量不同引數設定時。

此外，MLflow的搜尋功能支援複雜的查詢邏輯，讓我們能夠精確定位符合特定條件的實驗，而不必手動篩選大量結果。這在超引數最佳化過程中特別有價值。

從資料準備到模型訓練：機器學習工作流程全貌

在機器學習專案中，資料準備是第一步，而模型訓練則是核心環節。訓練過程的目標是建立一個能夠準確預測未見資料的函式（即"模型"）。

機器學習訓練的本質

模型訓練本質上類別似於人類學習新技能的過程：我們觀察、練習、糾正錯誤，並逐步改進。在機器學習中，我們從一個可能表現不佳的初始模型開始，然後透過一系列訓練步驟來改進它。

每個訓練步驟中，我們都會：

  1. 將訓練資料輸入模型
  2. 比較模型的預測結果與真實結果
  3. 評估模型表現
  4. 調整模型引數以提高準確性

一個優秀的模型能夠在不過度擬合特定輸入的情況下做出準確預測。這種平衡是機器學習中的核心挑戰之一。

使用TensorFlow構建推薦系統

TensorFlow是由Google開發的開放原始碼機器學習框架，目前是機器學習應用中最流行的函式庫之一，特別是在實作深度學習方面。TensorFlow在各種硬體上都有出色的計算任務支援，包括CPU、GPU和TPU。選擇TensorFlow進行本教程的原因是其高階API非常使用者友好，抽象了許多複雜細節。

深度學習簡介

近年來，深度學習——一類別利用人工神經網路從輸入資料中逐步提取高階特徵的演算法——變得越來越流行。深度學習能夠利用神經網路中的隱藏層學習輸入的高度抽象模型。

深度學習演算法在許多日常應用中都能找到，如影像識別和自然語言處理。神經網路的多個隱藏層使這些演算法能夠從資料中發現越來越抽象的細節。例如，影像分類別神經網路的初始層可能只發現物體邊緣，而更深層次可能學習更複雜的特徵並分類別影像中的物體。

推薦系統實作計劃

在這個例項中，我們將實作一個產品推薦系統，具體包括兩個主要步驟：

  1. 使用TensorFlow訓練產品推薦模型
  2. 使用Kubeflow包裝訓練程式碼並佈署到生產叢集

TensorFlow的高階Keras API使實作模型變得相對容易。事實上，模型的核心部分可以用不到50行Python程式碼實作。

Keras：深度學習的高階API

Keras是TensorFlow用於深度學習模型的高階API，具有使用者友好的介面和高度可擴充套件性。作為額外的優勢，Keras直接提供了許多常見的神經網路實作，所以你可以立即啟動並執行模型。

協同過濾推薦模型

我們的推薦系統根據一個簡單的假設——如果兩個人（Alice和Bob）對一組產品有相似的意見，那麼他們在其他產品上也更可能有相似的想法。換句話說，Alice更可能與Bob有相同的偏好，而不是隨機選擇的第三人。因此，我們可以僅使用者的購買歷史來構建推薦模型。

這就是協同過濾的理念——我們從許多使用者那裡收集偏好訊息（因此稱為"協同"），並使用這些資料進行選擇性預測（因此稱為"過濾"）。

推薦系統實作需求

要構建這個推薦模型，我們需要幾個關鍵元件：

  1. 使用者購買歷史：作為模型的輸入資料
  2. 資料儲存：使用MinIO作為儲存系統，確保模型在不同平台上工作
  3. 訓練模型：根據GitHub上的Keras模型實作

我們將首先使用Kubeflow的筆記本伺服器試驗這個模型，然後使用Kubeflow的TF工作 API將訓練作業佈署到叢集。

環境準備

要開始實驗，需要準備以下環境：

  1. 下載必要的筆記本和範例程式碼
  2. 確保有一個執行中的Kubeflow叢集，包括MinIO服務
  3. 安裝MinIO客戶端（“mc”）
  4. 準備使用者購買歷史資料以便進行訓練

這些準備工作完成後，我們就可以開始實際的模型訓練和佈署過程了。

機器學習工作流程整合：Kubeflow與MLflow的協同優勢

在本文中，我們探討了Kubeflow中繼資料元件如何支援儲存和檢視機器學習中繼資料，同時也討論了這種實作的侷限性，包括僅支援Python和較弱的UI介面。我們還介紹瞭如何使用MLflow補充Kubeflow，以及在這種情況下可以實作的額外功能。

將MLflow與Kubeflow結合使用，能夠顯著增強機器學習工作流程的可管理性和視覺化能力。這種整合為資料科學家和機器學習工程師提供了一個強大的工具集，使他們能夠更有效地跟蹤、比較和最佳化模型訓練過程。

隨著我們進一步探索使用TensorFlow和Scikit-learn訓練機器學習模型，這些工具的價值將變得更加明顯。無論是構建推薦系統還是其他型別的機器學習應用，良好的中繼資料管理和實驗跟蹤都是確保專案成功的關鍵因素。

在機器學習工程實踐中，玄貓發現合理規劃中繼資料管理策略能夠大幅提高迭代效率，減少重複工作，並使團隊協作更加順暢。隨著模型複雜度和實驗數量的增加，像MLflow這樣的專業工具的重要性只會越來越突出。

Kubeflow 與 TensorFlow 推薦系統實作

推薦系統已成為現代應用不可或缺的一部分，從電商到內容平台，都需要精準的推薦功能來提升使用者經驗。在開發推薦系統時，我們面臨的挑戰不僅是演算法選擇，還包括如何有效地管理資料、訓練模型以及佈署服務。Kubeflow 作為一個專為機器學習設計的雲原生平台，提供了完整的解決方案。

環境準備與資料存取設定

在開始建立推薦系統前，我們需要先設定好 MinIO 服務，用於儲存訓練資料和模型。MinIO 是一個高效能的開放原始碼物件儲存服務，完全相容 Amazon S3 API。

# 將 MinIO 服務轉發到本地連線埠
kubectl port-forward -n kubeflow svc/minio-service 9000:9000 &
# 設定 MinIO 主機
mc config host add minio http://localhost:9000 minio minio123
# 建立儲存桶
mc mb minio/data
# 複製資料到 MinIO
mc cp go/src/github.com/medium/items-recommender/data/recommend_1.csv \
minio/data/recommender/users.csv
mc cp go/src/github.com/medium/items-recommender/data/trx_data.csv \
minio/data/recommender/transactions.csv

上述指令集完成了幾個關鍵步驟：首先使用 kubectl port-forward 將 Kubeflow 名稱空間中的 MinIO 服務轉發到本地的 9000 連線埠，這樣我們就能透過 localhost 存取 MinIO。接著使用 MinIO 客戶端 (mc) 設定了連線引數，包括服務位址和登入憑證。然後建立了一個名為 data 的儲存桶，最後將推薦系統所需的使用者資料和交易資料上載到 MinIO 中。這些步驟建立了資料存取的基礎設施，讓我們的訓練程式能夠讀取這些資料。

建立並啟動 Notebook 環境

在 Kubeflow 中，Jupyter Notebook 是進行模型開發和實驗的主要工具。以下是建立新 Notebook 的步驟：

  1. 在 Kubeflow 儀錶板中導航到「Notebook Servers」面板
  2. 點選「New Server」並按照指示進行設定
  3. 選擇 tensorFlow-1.15.2-notebook-cpu:1.0 映像檔
  4. 待伺服器啟動後，點選右上角的「Upload」按鈕上載 Recommender_Kubeflow.ipynb 檔案
  5. 點選該檔案開始新的工作階段

Notebook 的前幾個部分包括匯入必要的程式函式庫和從 MinIO 讀取訓練資料，然後進行資料標準化，為訓練做好準備。這個過程稱為特徵準備，是機器學習流程中的關鍵步驟。

TensorFlow 模型訓練實作

準備好 Notebook 環境和資料後，我們可以開始建立 TensorFlow 工作階段：

# 建立 TF 工作階段並在 Keras 中設定
sess = tf.Session()
K.set_session(sess)
K.set_learning_phase(1)

這段程式碼初始化了 TensorFlow 工作階段，並將其設定為 Keras 的後端。tf.Session() 建立了一個 TensorFlow 計算圖的執行環境，K.set_session(sess) 告訴 Keras 使用這個工作階段，而 K.set_learning_phase(1) 將 Keras 設定為訓練模式（1 表示訓練，0 表示推論）。這些步驟是使用 TensorFlow 1.x 版本進行模型訓練的標準設定。

接下來，我們定義協同過濾模型類別：

class DeepCollaborativeFiltering(Model):
    def __init__(self, n_customers, n_products, n_factors, p_dropout = 0.2):
        x1 = Input(shape = (1,), name="user")
        P = Embedding(n_customers, n_factors, input_length = 1)(x1)
        P = Reshape((n_factors,))(P)
        
        x2 = Input(shape = (1,), name="product")
        Q = Embedding(n_products, n_factors, input_length = 1)(x2)
        Q = Reshape((n_factors,))(Q)
        
        x = concatenate([P, Q], axis=1)
        x = Dropout(p_dropout)(x)
        x = Dense(n_factors)(x)
        x = Activation('relu')(x)
        x = Dropout(p_dropout)(x)
        output = Dense(1)(x)
        
        super(DeepCollaborativeFiltering, self).__init__([x1, x2], output)
    
    def rate(self, customer_idxs, product_idxs):
        if (type(customer_idxs) == int and type(product_idxs) == int):
            return self.predict([np.array(customer_idxs).reshape((1,)),\
                                np.array(product_idxs).reshape((1,))])
        if (type(customer_idxs) == str and type(product_idxs) == str):
            return self.predict( \
                [np.array(customerMapping[customer_idxs]).reshape((1,)),\
                np.array(productMapping[product_idxs]).reshape((1,))])
        return self.predict([
            np.array([customerMapping[customer_idx] \
                     for customer_idx in customer_idxs]),
            np.array([productMapping[product_idx] \
                     for product_idx in product_idxs])
        ])

這個 DeepCollaborativeFiltering 類別實作了一個深度協同過濾模型。協同過濾是推薦系統中常用的技術，根據使用者與商品之間的互動資料來預測使用者對未接觸商品的偏好。

在這個實作中：

  1. 模型接收兩個輸入：使用者 ID 和商品 ID
  2. 每個 ID 透過嵌入層 (Embedding) 轉換為低維度向量，這些向量捕捉了使用者和商品的潛在特徵
  3. 使用者和商品的嵌入向量連線在一起，然後透過幾個全連線層 (Dense)
  4. Dropout 層用於防止過擬合
  5. 最終輸出是一個單一數值，代表預測的評分

rate 方法提供了一個便捷的介面，用於預測使用者對商品的評分，支援不同型別的輸入（整數索引或字串識別符號）。

現在我們可以建立模型例項：

model = DeepCollaborativeFiltering(n_customers, n_products, n_factors)
model.summary()

設定訓練引數與執行訓練

訓練前，我們需要設定一些超引數：

bs = 64
val_per = 0.25
epochs = 3

這裡設定的超引數控制著訓練過程：

  • bs = 64：批次大小為 64，即每次更新模型引數時使用 64 個樣本
  • val_per = 0.25：驗證集比例為 25%，即將 25% 的資料作為驗證集
  • epochs = 3：訓練 3 個完整週期

超引數的選擇對模型效能有顯著影響，但這裡我們先使用一些預設值。在實際應用中，我們可能需要透過超引數調整來最佳化模型效能。

接下來執行模型訓練：

model.compile(optimizer = 'adam', loss = mean_squared_logarithmic_error)
model.fit(x = [customer_idxs, product_idxs], y = ratings,
          batch_size = bs, epochs = epochs, validation_split = val_per)
print('Done training!')

這段程式碼完成了模型訓練：

  1. model.compile() 設定了最佳化器 (Adam) 和損失函式 (均方對數誤差)
  2. model.fit() 開始訓練過程，輸入為使用者索引和商品索引，輸出為評分
  3. 資料被分成訓練集和驗證集，模型進行指定次數的訓練週期

訓練完成後，我們應該能看到類別似以下的輸出結果：

Train on 100188 samples, validate on 33397 samples
Epoch 1/3
100188/100188 [==============================] - 21s 212us/step - loss: 0.0105 - val_loss: 0.0186
Epoch 2/3
100188/100188 [==============================] - 20s 203us/step - loss: 0.0092 - val_loss: 0.0188
Epoch 3/3
100188/100188 [==============================] - 21s 212us/step - loss: 0.0078 - val_loss: 0.0192
Done training!

從輸出中可以看到，訓練損失逐漸降低，這表明模型在訓練集上的表現越來越好。然而，驗證損失略有上升，這可能表明模型開始過擬合。在實際應用中，我們可能需要調整模型結構或使用正則化技術來改善這一點。

模型匯出與儲存

訓練完成後，我們需要匯出模型以便後續使用。首先設定匯出目標：

directorystream = minioClient.get_object('data', 'recommender/directory.txt')
directory = ""
for d in directorystream.stream(32*1024):
    directory += d.decode('utf-8')
arg_version = "1"
export_path = 's3://models/' + directory + '/' + arg_version + '/'
print ('Exporting trained model to', export_path)

這段程式碼從 MinIO 讀取一個目錄路徑，然後構建完整的模型匯出路徑。它使用 S3 協定格式 (s3://)，因為 MinIO 與 Amazon S3 API 相容。版本號設為 “1”，表示這是模型的第一個版本。

然後執行模型匯出：

# 輸入/輸出
tensor_info_users = tf.saved_model.utils.build_tensor_info(model.input[0])
tensor_info_products = tf.saved_model.utils.build_tensor_info(model.input[1])
tensor_info_pred = tf.saved_model.utils.build_tensor_info(model.output)
print ("tensor_info_users", tensor_info_users.name)
print ("tensor_info_products", tensor_info_products.name)
print ("tensor_info_pred", tensor_info_pred.name)

# 簽名
prediction_signature = (tf.saved_model.signature_def_utils.build_signature_def(
    inputs={"users": tensor_info_users, "products": tensor_info_products},
    outputs={"predictions": tensor_info_pred},
    method_name=tf.saved_model.signature_constants.PREDICT_METHOD_NAME))

# 匯出
legacy_init_op = tf.group(tf.tables_initializer(), name='legacy_init_op')
builder = tf.saved_model.builder.SavedModelBuilder(export_path)
builder.add_meta_graph_and_variables(
    sess, [tf.saved_model.tag_constants.SERVING],
    signature_def_map={
        tf.saved_model.signature_constants.DEFAULT_SERVING_SIGNATURE_DEF_KEY:
        prediction_signature,
    },
    legacy_init_op=legacy_init_op)
builder.save()

這段程式碼完成了模型匯出的複雜過程：

  1. 首先，為模型的輸入（使用者和商品索引）和輸出（預測評分）建立張量訊息
  2. 然後，定義預測簽名，指定輸入和輸出的名稱和型別
  3. 最後，使用 SavedModelBuilder 將模型和元圖（包含計算圖結構和變數）儲存到指定路徑

這種匯出格式使模型能夠被 TensorFlow Serving 等服務工具使用，便於佈署到生產環境。

佈署 TensorFlow 訓練任務

雖然 Jupyter Notebook 是原型開發和實驗的絕佳工具，但在實際生產環境中，我們通常需要更穩健的訓練方式。當我們需要使用更多資料、專用硬體或持續訓練時，Notebook 方式可能就不夠用了。

為了滿足這些需求，我們的訓練程式碼必須能夠輕鬆封裝並佈署到各種環境中。Kubeflow 提供了 TF工作，這是一個 Kubernetes 自定義資源，專為管理 TensorFlow 訓練任務而設計。

TF工作 允許我們：

  • 在叢集中排程和管理 TensorFlow 訓練任務
  • 利用分散式訓練功能，跨多個節點訓練大型模型
  • 自動處理錯誤還原和資源管理
  • 與 Kubeflow 管道整合，實作端對端的機器學習工作流程

要將我們的 Notebook 訓練程式碼轉換為 TF工作，需要進行以下步驟：

  1. 將訓練程式碼重構為可獨立執行的 Python 指令碼
  2. 建立 Docker 映像檔，包含所有必要的依賴
  3. 定義 TF工作 YAML 設定，指定映像檔、資源需求和訓練引數
  4. 使用 kubectl 或 Kubeflow UI 提交 TF工作

推薦系統架構設計考量

在設計和實作推薦系統時，除了模型訓練外，還有幾個關鍵方面需要考慮：

資料處理管道

推薦系統的效能很大程度上取決於資料品質。設計一個健全的資料處理管道至關重要，包括：

  • 資料收集：從各種來源收集使用者行為資料
  • 資料清洗：處理缺失值、異常值和重複項
  • 特徵工程：建立有意義的特徵，如使用者活躍度、商品熱門程度等
  • 資料分割：適當的訓練/驗證/測試集分割策略

模型選擇與評估

協同過濾只是眾多推薦演算法之一。在實際應用中，我們可能需要考慮：

  • 內容基礎過濾：根據商品屬性的推薦
  • 混合方法：結合多種推薦策略
  • 上下文感知推薦：考慮時間、位置等上下文因素
  • 模型評估指標：準確率、召回率、覆寫率、多樣性等

系統架構

一個完整的推薦系統通常包括：

  • 離線訓練子系統：定期重新訓練模型
  • 近線特徵處理：更新使用者和商品的特徵表示
  • 線上推薦服務：快速回應推薦請求
  • A/B 測試框架：評估不同推薦策略的效果

擴充套件性考量

隨著使用者和商品數量的增長，推薦系統面臨的擴充套件性挑戰包括：

  • 計算效率：如何處理大規模矩陣計算
  • 儲存效率：如何高效儲存和檢索使用者-商品互動資料
  • 服務延遲：如何確保推薦請求的低延遲回應
  • 增量更新：如何在不重新訓練整個模型的情況下更新推薦結果

從實驗到生產的最佳實踐

將推薦系統從實驗階段轉移到生產環境需要考慮以下最佳實踐：

  1. 模型版本控制：使用明確的版本標識和變更日誌記錄模型演化
  2. 特徵一致性：確保訓練和推論階段使用相同的特徵處理邏輯
  3. 模型監控：實施監控系統，追蹤模型效能和資料分佈變化
  4. 回復機制：設計快速回復到先前模型版本的機制
  5. 持續評估：建立持續評估框架，比較不同模型版本的效能

使用 Kubeflow 可以極大地簡化這些實踐的實施，因為它提供了模型版本控制、模型服務和監控等內建功能。

在推薦系統開發過程中，我發現平衡線上實驗與離線評估是一個關鍵挑戰。離線指標雖然重要，但最終還是線上使用者行為才是評判推薦系統成功的真正標準。因此，建立一個強大的 A/B 測試框架和精確的線上指標追蹤系統，對於持續改進推薦演算法至關重要。

在實際工作中，我常看到工程團隊過度專注於演算法創新而忽視資料品質和系統架構的重要性。事實上，一個根據高品質資料和穩健架構的簡單模型，往往比在雜亂資料上執行的複雜演算法表現更好。因此，在投入大量資源最佳化模型之前，確保資料管道和系統架構的健全性非常重要。

在 Kubeflow 環境中開發推薦系統時，充分利用其提供的工具生態系統可以大幅提高開發效率。例如，Katib 用於超引數調整，Pipelines 用於協調端對端工作流程，KFServing 用於模型佈署等。這些工具共同構成了一個完整的機器學習平台，能夠支援從實驗到生產的全流程。

透過本文介紹的方法和技術，我們可以在 Kubeflow 中構建一個功能完善的推薦系統，從資料準備到模型訓練再到服務佈署，形成一個完整的機器學習生命週期。隨著系統的不斷迭代和最佳化，我們可以持續提升推薦的準確性和相關性，為使用者提供更好的體驗。

機器學習模型的開發是一個迭代過程，需要不斷實驗和改進。Kubeflow 提供了一個完整的平台，使這個過程更加高效和系統化。透過本文介紹的方法，你可以開始在 Kubeflow 上構建自己的推薦系統，並逐步最佳化它以滿足實際業務需求。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。