Kubeflow：Katib 的神經網路架構搜尋能力

– 16 分鐘閱讀

Katib不僅能夠進行超引數調整，還支援神經網路架構搜尋(Neural Architecture Search, NAS)。目前，Katib實作了兩種主要的NAS方法：

  1. DARTS (Differentiable Architecture Search) - 一種根據梯度的方法，透過鬆弛搜尋空間使架構選擇可微分，從而能夠使用梯度下降進行最佳化。

  2. ENAS (Efficient Neural Architecture Search) - 一種根據強化學習的方法，透過引數分享提高搜尋效率，大幅減少所需的計算資源。

這些NAS功能讓Katib能夠自動尋找最佳的神經網路架構，而不僅是調整既定架構的超引數。Katib團隊正在積極開發更多NAS方法，未來會有更多選項可用。 在玄貓實際使用Katib進行NAS的經驗中，發現DARTS對於中小型模型架構搜尋效果較好，而ENAS則在搜尋大型模型架構時更為高效。如果你的計算資源有限，ENAS通常是更好的選擇，因為它的引數分享機制能大幅減少計算需求。

Argo執行器設定與權衡

Kubeflow的工作流程管理依賴於Argo，而Argo執行器的選擇對系統的安全性和可擴充套件性有重要影響。在選擇Argo執行器時，需要考慮以下幾種選項的優缺點：

Docker執行器

優點：

  • 支援所有工作流程範例
  • 可靠性高，經過充分測試
  • 高度可擴充套件
  • 利用Docker守護程式進行重要操作

缺點：

  • 安全性最低
  • 需要掛載主機的docker.sock（常被OPA拒絕）

Kubelet執行器

優點：

  • 安全性高，無法逃脫Pod的服務帳戶許可權
  • 適中的可擴充套件性
  • 日誌檢索和容器輪詢直接針對Kubelet

缺點：

  • 可能需要額外的Kubelet設定
  • 只能在卷（如emptyDir）中儲存引數/成品，而非基礎映像層

Kubernetes API執行器

優點：

  • 安全性高，無法逃脫Pod的服務帳戶許可權
  • 無需額外設定

缺點：

  • 可擴充套件性最低
  • 日誌檢索和容器輪詢對K8s API伺服器造成壓力
  • 只能在卷中儲存引數/成品

PNS執行器

優點：

  • 安全性高
  • 成品收集可以從基礎映像層完成
  • 可擴充套件：程式輪詢透過procfs完成

缺點：

  • 程式不再以pid 1執行
  • 對於完成太快的容器，成品收集可能失敗
  • 無法從掛載卷下的基礎映像層捕捉成品目錄
  • 相對不成熟

玄貓在實際佈署中發現，對於大多數標準Kubeflow使用場景，Kubelet執行器提供了安全性和可擴充套件性的最佳平衡。但如果你的環境對安全性要求特別高，可以考慮Kubernetes API執行器，儘管它的擴充套件性較低。

雲端特定工具與設定

雖然雲端特定工具可以加速開發，但也可能導致供應商鎖定。以下是在不同雲平台上使用Kubeflow的一些特定功能。

Google Cloud特定功能

由於Kubeflow起源於Google，在Google Cloud上執行時有一些額外功能：

TPU加速例項

機器學習流程的不同部分可以從不同型別的機器中獲益。在模型訓練階段，TPU加速的機器可以提供更大的效益。使用TPU資源需要明確匯入kfp.gcp模組：

import kfp.gcp as gcp

# 將TPU資源增加到容器操作
container_op.apply(gcp.use_tpu(
    tpu_cores=8,  # TPU核心數量
    tpu_resource='v3',  # TPU版本
    tf_version='2.3.0'  # TensorFlow版本
))

這段程式碼展示瞭如何在Kubeflow Pipeline中設定TPU資源。首先匯入kfp.gcp模組，然後使用apply方法和use_tpu函式為容器操作增加TPU資源。引數包括TPU核心數量、TPU版本和相容的TensorFlow版本。這種方式讓開發者能夠在需要大量計算能力的訓練任務中利用Google的TPU加速能力。

需要注意的是，TPU節點僅在特定區域可用，在使用前應檢查Google Cloud檔案以確認支援的區域。

使用Dataflow處理TFX

在Google Cloud上，可以設定Kubeflow的TFX元件使用Google的Dataflow進行分散式處理。這需要指定分散式輸出位置並設定TFX使用Dataflow執行器：

generated_output_uri = root_output_uri + kfp.dsl.EXECUTION_ID_PLACEHOLDER
beam_pipeline_args = [
    '--runner=DataflowRunner',
    '--project=' + project_id,
    '--temp_location=' + root_output_uri + '/tmp',
    '--region=' + gcp_region,
    '--disk_size_gb=50',  # 根據需要調整
]

records_example = tfx_csv_gen(
    input_uri=fetch.output,  # 必須在分散式儲存上
    beam_pipeline_args=beam_pipeline_args,
    output_examples_uri=generated_output_uri
)

這段程式碼展示瞭如何將TFX管道設定為使用Google Cloud Dataflow進行分散式處理。首先建立輸出URI，然後設定Beam管道引數，指定使用DataflowRunner、專案ID、臨時儲存位置、區域和磁碟大小。最後，在tfx_csv_gen操作中使用這些引數生成TFX範例。

使用Dataflow的主要優勢是能夠處理更大規模的資料，但需要注意的是，輸入必須儲存在分散式儲存上（如GCS），因為Dataflow工作者之間沒有分享的永續性儲存區。

雖然雲端特定加速可能帶來效益，但在選擇使用這些功能時需謹慎評估，確保這種權衡值得未來可能面臨的供應商轉換困難。

在應用程式中使用模型服務

Kubeflow提供了多種佈署訓練模型並提供REST和gRPC介面的方法，但在支援將這些模型用於自定義應用程式方面有所不足。以下是一些利用Kubeflow暴露的模型服務構建應用程式的方法。

流處理與批處理應用

利用模型推理的應用程式大致可分為兩類別：實時/流處理應用和批處理應用。

流處理應用

在流處理應用中，模型推理直接在資料產生或接收時進行。典型情況下，一次只處理一個請求，並在請求到達時立即用於推理。

大多數現代流處理應用使用Apache Kafka作為系統的資料骨幹。實作流處理應用有兩種主要選項：

  1. 流處理引擎 - 根據將計算組織成塊並利用叢集架構。這種方式支援執行平行化、容錯處理和檢查點機制，提高叢集基礎執行的可靠性。

  2. 流處理函式庫 - 提供一組簡化流處理應用構建的特定語言建構的函式庫。這些函式庫通常不支援分散式處理或叢集功能，這些功能通常留給開發者自行實作。

選擇流處理引擎還是流處理函式庫是權衡功能強大性和簡單性的結果。流處理引擎提供更多功能，但要求開發者遵循其程式設計模型和佈署方式，並且通常需要更陡峭的學習曲線。而流處理函式庫通常更容易使用，提供更多靈活性，但需要開發者自行處理分散式計算和容錯等問題。

在玄貓的實踐中，對於需要處理海量資料的大型組織，流處理引擎如Apache Flink或Spark Streaming是更好的選擇；而對於中小型應用或原型開發，流處理函式庫如Kafka Streams或Spring Cloud Stream提供了更快的開發速度和更低的上手門檻。

批處理應用

在批處理場景中，所有資料都預先可用，可以順序或平行用於推理。批處理通常適用於不需要即時結果的場景，如夜間報告生成、大規模資料分析或模型評估。

批處理應用可以利用Kubeflow的模型服務進行大規模平行推理，透過將資料分割成多個批次並同時傳送請求來提高處理速度。這種方法特別適合於處理歷史資料或定期進行的大規模分析任務。

實作模型服務客戶端

無論是流處理還是批處理應用，都需要實作客戶端程式碼來呼叫Kubeflow佈署的模型服務。以下是一個使用Python請求模型服務的簡單範例：

import requests
import json

def predict(instances, server_url):
    """向模型服務傳送預測請求"""
    headers = {"content-type": "application/json"}
    data = json.dumps({"instances": instances})
    response = requests.post(server_url, data=data, headers=headers)
    return response.json()

# 使用範例
instances = [
    [1.0, 2.0, 3.0, 4.0],
    [5.0, 6.0, 7.0, 8.0]
]
server_url = "http://my-model-service.kubeflow:8000/v1/models/my-model:predict"
predictions = predict(instances, server_url)
print(predictions)

這段程式碼展示瞭如何使用Python的requests函式庫向Kubeflow佈署的模型服務傳送預測請求。函式predict接受兩個引數：instances（要預測的資料例項列表）和server_url（模型服務的URL）。

函式首先設定適當的HTTP頭，指定內容型別為JSON。然後將例項資料轉換為JSON格式，並使用POST方法傳送到指定的服務URL。最後，函式解析並回傳JSON格式的回應。

在使用範例中，我們建立了兩個例項的列表，每個例項是一個特徵向量，然後指定模型服務的URL並呼叫predict函式。URL格式遵循TensorFlow Serving的標準格式，包含模型名稱和predict端點。

這種方法適用於任何透過REST API暴露的Kubeflow模型服務，無論是KFServing、TensorFlow Serving還是其他服務框架。

Katib作為Kubeflow中的超引數調整和自動化機器學習元件，不僅提供了強大的超引數搜尋功能，還支援神經網路架構搜尋，幫助開發者自動尋找最佳模型結構。在實際佈署中，選擇適當的Argo執行器對於確保系統安全性和可擴充套件性至關重要，不同執行器之間的權衡需要根據具體需求進行評估。

對於在雲環境中執行Kubeflow，各雲平台提供的特定工具和加速功能可以提升開發效率，但也要警惕供應商鎖定的風險。特別是在Google Cloud上，TPU加速和Dataflow分散式處理等功能可以顯著提高大規模機器學習工作流程的效能。

最後，雖然Kubeflow提供了多種模型佈署選項，但在將這些模型整合到實際應用中時，開發者需要根據應用型別（流處理或批處理）選擇適當的實作方法，並開發相應的客戶端程式碼來呼叫模型服務。透過這些方法，組織可以充分利用Kubeflow的能力，將機器學習模型有效地整合到生產環境中，從而為業務帶來實際價值。

串流處理技術的現代選擇

隨著即時資料處理需求的增長，串流處理技術已成為現代資料架構的核心元件。在評估各種技術選項時，我發現可以將它們大致分為兩類別：串流處理引擎和串流處理函式庫。

主流串流處理引擎

目前業界領先的串流處理引擎包括：

  • Apache Spark - 以其統一的批次和串流處理能力著稱
  • Apache Flink - 專為低延遲和高吞吐量的串流處理而設計
  • Apache Beam - 提供統一的程式設計模型，可在多個執行引擎上執行

流行的串流處理函式庫

相對於完整的引擎，串流處理函式庫通常更輕量與易於整合：

  • Apache Kafka Streams - 直接建立在Kafka之上的輕量級串流處理函式庫
  • Akka Streams - 根據反應式程式設計原則的串流處理函式庫

串流引擎與串流函式庫的深度比較

在我協助客戶選擇串流技術時，常發現一個關鍵差異點：企業所有權模式。這個因素往往被忽視，但實際上對技術採用有重大影響。

企業所有權模式的影響

串流處理引擎（如Flink）通常由企業級中央團隊負責管理，而串流處理函式庫（如Kafka Streams）則往往由個別開發團隊掌控。這種差異會直接影響採用難度和速度。

串流引擎的優勢與限制

串流處理引擎適合需要以下特性的應用：

  • 透過叢集平行化實作的高擴充套件性和高吞吐量
  • 事件時間語義處理能力
  • 內建的檢查點機制
  • 完善的監控和管理支援
  • 混合串流和批次處理能力

然而，使用這類別引擎的主要缺點是開發團隊必須遵循其預設的程式設計和佈署模型，靈活性相對較低。

串流函式庫的優勢與挑戰

相較之下，串流處理函式庫提供的程式設計模型允許開發者根據專案需求量身開發應用或微服務，並以簡單的獨立Java應用形式佈署。在實際專案中，我發現這種方式能大幅加速開發流程。

不過，選擇串流函式庫意味著團隊需要自行實作擴充套件性、高用性和監控解決方案。好在根據Kafka的實作可以透過利用Kafka本身的功能來支援部分這些需求。

Cloudflow：串流應用開發的新正規化

在實際專案中，我經常發現客戶需要同時使用多種引擎和函式庫來建構串流應用，這導致了額外的整合與維護複雜性。為解決這個問題，Cloudflow這類別開放原始碼專案應運而生。

Cloudflow架構與核心概念

Cloudflow允許開發者快速開發、協調和操作在Kubernetes上執行的分散式串流應用。它支援將串流應用構建為一組小型、可組合的元件，這些元件透過Kafka進行通訊，並透過根據結構描述的契約連線在一起。

在Cloudflow的核心是一個操作器（operator），負責佈署/解除佈署、管理和擴充套件管道及個別串流元件（streamlet）。該操作器還利用現有的Flink和Spark操作器來管理Flink和Spark串流元件。

Cloudflow的主要優勢

Cloudflow解決了串流應用開發中的幾個關鍵挑戰：

  1. 開發簡化 - 透過生成大量樣板程式碼，讓開發者專注於業務邏輯
  2. 建構工具 - 提供從業務邏輯到可佈署Docker映像的完整工具鏈
  3. 佈署自動化 - 提供Kubernetes工具，實作單一命令佈署分散式應用
  4. 操作管理 - 提供洞察、可觀測性和生命週期管理工具

此外，Cloudflow還支援按需擴充套件串流應用的個別元件，這是生產環境中的關鍵需求。

在串流應用中整合模型推論服務

在構建需要機器學習能力的串流應用時，模型推論服務的整合是一個常見需求。根據我的經驗，有幾種方式可以實作這一目標。

動態控制流模式

在使用Cloudflow實作串流應用時，模型服務呼叫通常由根據"動態控制流模式"的獨立串流元件實作。

在這種模式中，串流元件維護一個狀態，該狀態可以是模型伺服器的URL（當使用外部模型伺服器時）或者模型本身（當使用嵌入式模型時）。實際的資料處理是透過呼叫模型伺服器取得推論結果來完成的，可以使用REST或gRPC介面。

public class ModelServingProcessor extends StreamletLogic {
    private String modelServerUrl;
    
    @Override
    public void process(StreamletContext context) {
        // 從設定流讀取模型伺服器URL
        context.source("config-in")
               .map(record -> record.asString())
               .to(url -> this.modelServerUrl = url);
        
        // 處理輸入資料並呼叫模型服務
        context.source("data-in")
               .map(this::inferWithModel)
               .to(context.sink("result-out"));
    }
    
    private Result inferWithModel(Record record) {
        // 使用HTTP客戶端呼叫模型服務
        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(modelServerUrl + "/predict"))
                .POST(HttpRequest.BodyPublishers.ofString(record.asJson()))
                .header("Content-Type", "application/json")
                .build();
                
        try {
            HttpResponse<String> response = client.send(request, 
                HttpResponse.BodyHandlers.ofString());
            return new Result(response.body());
        } catch (Exception e) {
            log.error("Model inference failed", e);
            return new Result("Error: " + e.getMessage());
        }
    }
}

這段程式碼展示了一個根據Cloudflow的模型服務處理器實作。它包含兩個主要流程：

  1. 設定流處理 - 從名為"config-in"的源讀取模型伺服器URL，並更新處理器的狀態
  2. 資料處理流 - 從"data-in"源讀取輸入資料，呼叫模型服務進行推論，然後將結果傳送到"result-out"接收器

inferWithModel方法展示瞭如何使用Java的HttpClient傳送預測請求到模型伺服器。這種方式的優點是可以在不重新佈署應用的情況下，透過設定流動態更新模型服務的URL。

模型監控與洞察

除了基本的推論處理外，可以引入額外的串流元件來取得模型服務的洞察，如解釋和漂移檢測。這些元件通常使用相同的動態控制流模式，但針對不同的模型伺服器端點。

利用模型服務構建批處理應用

雖然串流處理越來越受歡迎，但批處理仍然是許多企業資料處理的根本。在批處理中整合模型服務有幾種策略。

批處理的基本方法與效能挑戰

最簡單的批處理應用實作是依序處理資料集中的每個樣本，為每個樣本呼叫模型服務。雖然這種實作可行，但由於處理每個元素的網路開銷，效能通常不佳。

批次處理策略

提高處理速度的一種流行方法是使用批次處理（batching）。以TensorFlow Serving為例，它支援兩種批次處理方法：

  1. 伺服器端批次處理 - 透過設定--enable_batching和--batching_parameters_file標誌啟用。當伺服器端達到完整批次時，推論請求會在內部合併為單個大請求（張量），並在合併請求上執行TensorFlow工作階段。

  2. 客戶端批次處理 - 在客戶端將多個輸入組合在一起，形成單個請求。

# 伺服器端批次處理設定範例
{
  "max_batch_size": 128,          # 批次的最大大小
  "batch_timeout_micros": 5000,   # 等待形成批次的最長時間(微秒)
  "max_enqueued_batches": 100,    # 佇列中允許的最大批次數
  "num_batch_threads": 8          # 處理批次的執行緒數
}

這個JSON設定展示了TensorFlow Serving的伺服器端批次處理設定。關鍵引數包括：

  • max_batch_size 控制單個批次可以包含的最大樣本數
  • batch_timeout_micros 定義了即使批次未滿，也會處理的超時間
  • max_enqueued_batches 限制了佇列中等待處理的批次數量
  • num_batch_threads 指定了用於處理批次的執行緒數

這些引數需要根據硬體資源（特別是CPU與GPU）和延遲/吞吐量需求進行調整。例如，GPU通常能處理更大的批次，但可能需要較長的批次形成時間。

多執行緒批處理

儘管批次處理可以顯著提高批處理推論的效能，但對於達到效能目標通常還不夠。另一種流行的效能改進方法是多執行緒處理。

這種方法的核心思想是佈署模型伺服器的多個例項，將資料處理分割為多個執行緒，並允許每個執行緒負責對其負責的部分資料進行推論。

透過串流實作批處理

實作多執行緒的一種方式是透過串流技術來進行批處理。這可以透過實作讀取源資料並將每條記錄寫入Kafka進行處理的軟體元件來完成。這種方法有效地將批處理轉變為串流處理，透過架構設計實作更好的可擴充套件性。

public class BatchToStreamProcessor {
    public static void main(String[] args) {
        // 設定Kafka生產者
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        Producer<String, String> producer = new KafkaProducer<>(props);
        
        // 讀取批處理資料集
        try (BufferedReader reader = new BufferedReader(new FileReader("dataset.csv"))) {
            String line;
            while ((line = reader.readLine()) != null) {
                // 將每條記錄傳送到Kafka主題
                producer.send(new ProducerRecord<>("inference-requests", line));
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
        
        System.out.println("批處理資料已轉換為串流請求");
    }
}

這段程式碼展示瞭如何將批處理作業轉換為串流處理的基本方法。它使用Kafka生產者將批處理資料集中的每一行傳送到名為"inference-requests"的Kafka主題。

這種方法的優點包括：

  1. 可以利用Kafka的分割槽功能實作自然的平行處理
  2. 可以輕鬆擴充套件消費者（模型推論處理器）的數量來提高吞吐量
  3. 提供了內建的緩衝機制，防止模型服務過載
  4. 允許使用相同的處理邏輯同時處理批次和實時資料

這種模式實際上將MapReduce的概念應用於模型推論，其中"map"階段是分散的模型推論處理。

選擇合適的串流技術與架構模式

在選擇串流處理技術和架構模式時，我建議考慮以下因素：

團隊技術能力與組織結構

如果你的組織有專門的平台團隊負責基礎設施管理，串流處理引擎可能是更好的選擇。相反，如果各個開發團隊擁有更多自主權，串流處理函式庫可能更適合。

應用需求與複雜性

對於需要複雜事件處理、狀態管理和精確一次語義的應用，完整的串流處理引擎通常更合適。而對於較簡單的使用案例，串流處理函式庫可能提供足夠的功能，同時降低複雜性。

整合與佈署考量

評估現有系統和佈署策略。如果你已經在使用Kubernetes，Cloudflow可能是一個理想的選擇，因為它專為該環境設計。如果你的基礎設施主要根據Kafka，則Kafka Streams可能提供最順暢的整合體驗。

模型服務整合方式

根據模型複雜性和效能需求選擇合適的模型服務整合方式。對於需要低延遲的應用，嵌入式模型可能更合適；而對於需要頻繁更新或計算密集型模型，外部模型服務可能是更好的選擇。

串流處理技術的選擇不僅關乎技術特性，還涉及組織結構、團隊能力和長期維護考量。透過仔細評估這些因素，可以選擇最適合特定需求的技術和架構模式，從而構建高效、可擴充套件與易於維護的串流應用。

模型服務整合：連線AI與實際應用的橋樑

在AI技術日益成熟的今天，如何將訓練好的機器學習模型有效整合到實際應用中，已成為許多開發團隊面臨的關鍵挑戰。模型服務（Model Serving）作為連線模型訓練與實際應用的橋樑，其重要性不言而喻。無論是批次處理大量資料，還是即時回應使用者請求的串流應用，選擇合適的模型服務架構都會直接影響系統的效能、可擴充套件性與維護成本。

在實作AI應用時，我發現許多開發者往往專注於模型訓練與最佳化，卻忽略了模型佈署與服務整合的複雜性。這種不平衡的關注往往導致即使擁有高準確度的模型，在實際應用中也無法發揮其潛力。本文將分享如何在不同應用場景中有效整合模型服務，並提供實用的架構設計與實作。

模型服務架構概覽

在深入討論具體實作前，讓我們先了解模型服務的基本架構。現代模型服務架構通常包含三個核心層次：

  1. 服務層：負責接收請求、路由和負載平衡，如Istio或Ambassador等服務網格
  2. 模型伺服器：執行實際推論的核心元件，如TensorFlow Serving或KFServing
  3. 應用整合層：將模型服務與業務邏輯整合的層次，可能是批次處理或串流處理系統

這三層架構提供了彈性與可擴充套件性，允許每一層獨立擴充套件以應對不同的負載需求。在實際應用中，我們可以根據使用場景選擇不同的實作方式，從簡單的REST API呼叫到複雜的串流處理架構。

玄貓

玄貓

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