自動水平調度資源

啟用自動水平調度資源功能後，Dataflow 會為工作選擇適當數量的 worker 執行個體，並視需要新增或移除 worker。Dataflow 會根據 worker 的平均 CPU 使用率和管道的平行處理程度，調整資源規模。管道的平行處理級別，是指在任何指定時間，以最有效率的方式處理資料所需的執行緒數量估計值。

批次和串流管道都支援水平自動調度資源。

批次自動調度資源

所有批次管道預設會啟用水平自動調度資源。Dataflow 會根據管道各階段預估的工作總量，自動選擇 worker 數量。這項預估值取決於輸入大小和目前的處理量。Dataflow 每 30 秒會根據執行進度重新評估工作量。當預估工作總量增加或減少時，Dataflow 會動態調度資源，增加或減少 worker 數量。

工作站數量與工作量呈次線性關係。舉例來說，如果工作量增加一倍，工作站數量不會增加一倍。

如果發生下列任一情況，Dataflow 會維持或減少 worker 數量，以節省閒置資源：

  • worker 的平均 CPU 使用率低於 5%。
  • 由於無法平行處理的工作 (例如壓縮檔案或未分割的 I/O 模組導致資料無法分割)，平行處理會受到限制。
  • 平行處理程度固定，例如寫入 Cloud Storage 中的現有檔案時。

如要設定 worker 數量上限，請設定 --maxNumWorkers 管道選項。預設值為 2,000。 如要設定 worker 數量的下限，請設定 --dataflow-service-options=min_num_workers 服務選項。這些標記為選用。

串流自動調度資源

對於串流工作，Dataflow 可透過水平自動調度資源功能，根據負載和資源使用率的變化，彈性調整 worker 數量。

根據預設，使用 Streaming Engine 的串流工作會啟用水平自動調度資源。如要為未使用 Streaming Engine 的串流工作啟用水平自動調度資源，請在啟動管道時設定下列管道選項：

Java

--autoscalingAlgorithm=THROUGHPUT_BASED
--maxNumWorkers=MAX_WORKERS

將 MAX_WORKERS 替換為 worker 執行個體數量上限。

Python

--autoscaling_algorithm=THROUGHPUT_BASED
--max_num_workers=MAX_WORKERS

將 MAX_WORKERS 替換為 worker 執行個體數量上限。

Go

--autoscaling_algorithm=THROUGHPUT_BASED
--max_num_workers=MAX_WORKERS

將 MAX_WORKERS 替換為 worker 執行個體數量上限。

如要設定 worker 數量的下限，請設定 --dataflow-service-options=min_num_workers 服務選項。設定這個值後，水平自動調度資源功能就不會將 worker 數量縮減至低於指定數量。這個標記是選用的。

如果串流工作使用 Streaming Engine，您也可以設定延遲時間級別，藉此設定待處理工作閾值。

對於已在執行的串流工作，您可以使用執行中工作更新，更新 worker 數量下限、worker 數量上限或延遲時間級別。如要調整設定，請設定 min-num-workers、max-num-workers 或 latency-tier 旗標。詳情請參閱「調整串流管道的水平自動調度資源」。

停用水平自動調度資源功能

如要停用水平自動調度資源功能，請在執行工作時設定下列管道選項。

Java

--autoscalingAlgorithm=NONE

如果停用自動水平調度資源功能，Dataflow 會根據 --numWorkers 選項設定 worker 數量。

Python

--autoscaling_algorithm=NONE

如果停用自動水平調度資源功能，Dataflow 會根據 --num_workers 選項設定 worker 數量。

Go

--autoscaling_algorithm=NONE

如果停用自動水平調度資源功能，Dataflow 會根據 --num_workers 選項設定 worker 數量。

自訂來源

如果您建立自訂資料來源，可以實作方法，為水平自動調度資源演算法提供更多資訊，進而提升效能：

Java

有界來源

  • 在 BoundedSource 子類別中，實作 getEstimatedSizeBytes 方法。Dataflow 服務會使用 getEstimatedSizeBytes，計算管道一開始要使用的 worker 數量。
  • 在 BoundedReader 子類別中，實作 getFractionConsumed 方法。Dataflow 服務會使用 getFractionConsumed 追蹤讀取進度，並在讀取期間收斂至正確的使用者數量。

無界限來源

來源必須向 Dataflow 服務回報待處理工作。待處理作業是指來源尚未處理的輸入內容，以位元組為單位。如要向服務通知待處理項目，請在 UnboundedReader 類別中實作下列任一方法。

  • getSplitBacklogBytes() - 來源的目前組別待處理作業。服務會匯總所有分割的待處理工作。
  • getTotalBacklogBytes() - 所有組別的全域待處理作業。在某些情況下，待處理工作量不適用於每個分割，只能在所有分割中計算。只有第一個分割 (分割 ID「0」) 需要提供待處理工作總量。

Apache Beam 存放區包含多個實作 UnboundedReader 類別的自訂來源範例。

Python

有界來源

  • 在 BoundedSource 子類別中，實作 estimate_size 方法。Dataflow 服務會使用 estimate_size，計算管道要使用的初始 worker 數量。
  • 在 RangeTracker 子類別中，實作 fraction_consumed 方法。Dataflow 服務會使用 fraction_consumed 追蹤讀取進度，並在讀取期間收斂至正確的使用者數量。

Go

有界來源

  • 在 RangeTracker 中實作 GetProgress() 方法。Dataflow 服務會使用 GetProgress 追蹤讀取進度，並在讀取期間收斂至正確的使用 worker 數量。

限制

  • 在執行 Dataflow Prime 的工作中，水平自動調度資源功能會在垂直自動調度資源期間和之後最多 10 分鐘內停用。詳情請參閱「對自動水平調度資源的影響」。
  • 如果管道未使用 Dataflow Shuffle，Dataflow 可能無法有效縮減 worker，因為 worker 可能已將重組資料儲存在本機磁碟。
  • Apache Beam SDK 2.60.0 以上版本支援 PeriodicImpulse 轉換，並可搭配串流自動調度資源功能。如果管道使用 PeriodicImpulse 搭配舊版 SDK，Dataflow worker 就不會按照預期縮減。

後續步驟