使用 BatchElements 轉換動態分組元素

為何要批次處理元素？

Apache Beam 提供 BatchElements 轉換來分攤後續操作的處理時間。這最適合應用於每次調用都有大量固定成本，而每次調用中處理的每個元素的成本較小的操作之前。簡而言之，與一次處理一個元素相比，批次處理元素是在一次處理多個元素時效率更高的操作之前的最佳化步驟。

範例：RunInference 中的批次處理

在 RunInference 的上下文中，批次處理元素主要用於減少對用於推論的模型進行的呼叫次數。這對於在工作節點推論環境中的效率非常重要，但在考慮將推論呼叫傳送到遠端服務時，影響更大。當 API 速率限制是一個考量時，盡可能批次處理元素是降低超出配額風險的重要方法。RunInference 架構總是 呼叫到 BatchElements，進一步突顯此步驟對於推論工作負載的重要性。

在捆綁中批次處理

用於批次處理元素的主要方法是在元素捆綁中進行批次處理。會迭代捆綁，在達到目標批次大小之前將元素附加到捆綁，然後發射。保證相同批次中的元素存在於相同的事件時間視窗中，因為批次處理函式會感知視窗。如果在到達捆綁末端時有不完整的批次被緩衝，則會發射該批次。

參數調整考量

此方式批次處理的大部分調整都圍繞用於基於下游效能估計最佳批次大小的參數。_BatchSizeEstimator 物件維護關於發射的捆綁的批次大小和時鐘時間的資料點，然後嘗試使用線性迴歸求解當前理想的批次大小。BatchElements 的 文件字串詳細說明了這一點和此處處理的參數。對於大多數使用者來說，最重要的考量是最小和最大批次大小（將它們設定為相同的值會產生固定批次大小，而不是動態批次大小）和元素大小函式。後者允許使用者定義一個 lambda，該 lambda 可以個別調整每個輸入的大小，而不是簡單地將每個元素計為大小 1。如果輸入在傳遞給模型時具有明顯不同的處理時間，這可能非常有用；例如，將文字主體傳送到模型可以按字元長度調整大小。

跨捆綁批次處理

在捆綁中批次處理元素有一個主要的缺點：如果捆綁很小（甚至是單個元素），則該操作實際上不會執行任何操作。在串流管道中尤其如此，在串流管道中，我們預期捆綁只包含 1 到 2 個元素。為了能夠使用小捆綁進行批次處理，我們必須能夠跨捆綁批次處理元素。如需深入瞭解其運作方式的技術說明，跨捆綁批次處理（也稱為有狀態批次處理）的設計文件概述了程式碼；但是，高層次的解釋是，我們利用 Beam 的 狀態 API 來儲存跨捆綁的元素，當我們達到所需的捆綁大小或將捆綁緩衝了某個最大時間量時，將它們發射到下游。

若要啟用有狀態批次處理，請在 Apache Beam 2.52.0 或更高版本中將 max_batch_duration_secs 參數傳遞至 BatchElements。這將鍵入輸入元素以確保可以使用狀態 API，然後使用有狀態的批次處理函式。當批次達到當前的目標批次大小或批次已緩衝的時間大於或等於 max_batch_duration_secs 時，將會發射批次。應注意，基於時間的行為是使用 計時器 API 實作的，並且是盡力而為的，因此批次在發射之前可能被持有的實際時間長度可能會大於設定的最大值。

跨捆綁批次處理確實有其自身的缺點

參數調整考量

除了在捆綁中進行批次處理所使用的所有調整參數外，調整 max_batch_duration_secs 參數將對轉換的吞吐量產生重大影響。選擇最大批次持續時間時，需要在吞吐量和一致的批次大小之間做出權衡。較大的值通常會降低吞吐量，因為不完整的批次在傳送到推論呼叫本身之前會被保留更長時間；但是，這將更一致地產生完整批次。另一方面，較小的值通常具有較高的吞吐量，但可能會導致較小的批次大小。您更重視哪一種取決於您的使用案例。

還需要注意的是，這些趨勢並不總是絕對成立！在串流情境中，元素被注入管道的頻率存在相當大的差異。您可能可以使用較短的最大批次持續時間來達到所需的批次大小，但這與使用較長的最大持續時間持續達到該批次大小是不同的。最好將這些描述的權衡視為平均情況來考量。