在自訂容器中執行 Dataflow 工作

本文說明如何使用自訂容器執行 Dataflow 管道。

如要瞭解如何建立容器映像檔，請參閱「為 Dataflow 建構自訂容器映像檔」。

執行管道時，請使用 Apache Beam SDK 啟動管道，並確保 SDK 的版本和語言版本與自訂容器映像檔中的 SDK 相同。這個步驟可避免因不相容的依附元件或 SDK 導致非預期的錯誤。

在本機測試

在 Dataflow 中執行管道之前，建議先在本機測試容器映像檔，這樣可以更快速地進行測試和偵錯。

如要進一步瞭解 Apache Beam 的特定用法，請參閱「使用自訂容器映像檔執行管道」的 Apache Beam 指南。

使用 PortableRunner 進行基本測試

如要驗證是否可以提取遠端容器映像檔，並執行簡單的管道，請使用 Apache Beam PortableRunner。使用 PortableRunner 時，工作提交作業會在本地環境中進行，而 DoFn 執行作業則會在 Docker 環境中進行。

使用 GPU 時，Docker 容器可能無法存取 GPU。如要測試含有 GPU 的容器，請使用直接執行器，並按照「使用 GPU」頁面「使用獨立 VM 進行偵錯」一節中的步驟，在含有 GPU 的獨立 VM 上測試容器映像檔。

以下範例會執行管道：

Java

mvn compile exec:java -Dexec.mainClass=com.example.package.MyClassWithMain \
    -Dexec.args="--runner=PortableRunner \
    --jobEndpoint=REGION \
    --defaultEnvironmentType=DOCKER \
    --defaultEnvironmentConfig=IMAGE_URI \
    --inputFile=INPUT_FILE \
    --output=OUTPUT_FILE"

Python

python path/to/my/pipeline.py \
  --runner=PortableRunner \
  --job_endpoint=REGION \
  --environment_type=DOCKER \
  --environment_config=IMAGE_URI \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE

Go

go path/to/my/pipeline.go \
  --runner=PortableRunner \
  --job_endpoint=REGION \
  --environment_type=DOCKER \
  --environment_config=IMAGE_URI \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE

更改下列內容：

  • REGION：要使用的作業服務區域，格式為地址和通訊埠。例如：localhost:3000。使用 embed 執行程序內工作服務。
  • IMAGE_URI：自訂容器映像檔 URI。
  • INPUT_FILE：可讀取為文字檔案的輸入檔案。SDK 測試架構
    容器映像檔必須能存取這個檔案，無論是預先載入容器映像檔，還是遠端檔案都沒問題。
  • OUTPUT_FILE：輸出內容的寫入路徑。這個路徑可以是遠端路徑，也可以是容器上的本機路徑。

pipeline 順利完成後，請查看控制台記錄，確認 pipeline 已順利完成，且使用了 IMAGE_URI 指定的遠端映像檔。

管道執行完畢後，儲存在容器中的檔案不會出現在本機檔案系統，且容器會停止運作。您可以使用 docker cp，從已停止的容器檔案系統複製檔案。

您也可以使用以下程式碼：

  • 將輸出內容提供給 Cloud Storage 等遠端檔案系統。 您可能需要手動設定存取權以進行測試，包括憑證檔案或應用程式預設憑證。
  • 如要快速偵錯，請新增暫時的記錄。

使用 Direct Runner

如要更深入地在本機測試容器映像檔和管道，請使用 Apache Beam Direct Runner。

您可以在與容器映像檔相符的本機環境中測試，或在執行中的容器上啟動管道，藉此驗證管道是否與容器無關。

Java

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  mvn compile exec:java -Dexec.mainClass=com.example.package.MyClassWithMain ...

Python

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  python path/to/my/pipeline.py ...

Go

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  go path/to/my/pipeline.go ...

將 IMAGE_URI 替換為自訂容器映像檔 URI。

這些範例假設所有管道檔案 (包括管道本身) 都位於自訂容器中，已從本機檔案系統掛接，或是遠端檔案，且可供 Apache Beam 和容器存取。舉例來說，如要使用 Maven (mvn) 執行先前的 Java 範例，必須在容器中暫存 Maven 及其依附元件。詳情請參閱 Docker 說明文件中的「儲存空間」和「docker run」。

在 Direct Runner 上進行測試的目標，是測試自訂容器環境中的管道，而不是測試使用預設 ENTRYPOINT 執行容器。修改 ENTRYPOINT (例如 docker run --entrypoint ...)，直接執行管道，或允許在容器中手動執行指令。

如果您依賴特定設定 (以在 Compute Engine 上執行容器為基礎)，可以直接在 Compute Engine VM 上執行容器。詳情請參閱「Containers on Compute Engine」。

啟動 Dataflow 工作

在 Dataflow 上啟動 Apache Beam 管道時，請指定容器映像檔的路徑。請勿將 :latest 標記與自訂圖片搭配使用。為建構版本加上日期或專屬 ID 標記。如果發生錯誤，使用這類代碼或許就能將管道執行作業還原為先前已知的正常設定，並檢查變更。

Java

使用 --sdkContainerImage 為 Java 執行階段指定 SDK 容器映像檔。

使用 --experiments=use_runner_v2 啟用 Runner v2。

Python

如果使用 2.30.0 以上版本的 SDK，請使用管道選項 --sdk_container_image 指定 SDK 容器映像檔。

如果是舊版 SDK，請使用管道選項 --worker_harness_container_image 指定要用於工作站架構的容器映像檔位置。

自訂容器僅支援 Dataflow Runner v2。如要啟動批次 Python 管道，請設定 --experiments=use_runner_v2 標記。如果您要啟動串流 Python 管道，則不必指定實驗，因為串流 Python 管道預設會使用 Runner v2。

Go

如果使用 2.40.0 以上版本的 SDK，請使用管道選項 --sdk_container_image 指定 SDK 容器映像檔。

如果是舊版 SDK，請使用管道選項 --worker_harness_container_image 指定要用於工作站架構的容器映像檔位置。

所有版本的 Go SDK 都支援自訂容器，因為這些容器預設使用 Dataflow Runner v2。

以下範例說明如何使用自訂容器啟動批次 WordCount 範例。

Java

mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
   -Dexec.args="--runner=DataflowRunner \
                --inputFile=INPUT_FILE \
                --output=OUTPUT_FILE \
                --project=PROJECT_ID \
                --region=REGION \
                --gcpTempLocation=TEMP_LOCATION \
                --diskSizeGb=DISK_SIZE_GB \
                --experiments=use_runner_v2 \
                --sdkContainerImage=IMAGE_URI"

Python

使用 Python 適用的 Apache Beam SDK 2.30.0 以上版本：

python -m apache_beam.examples.wordcount \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --temp_location=TEMP_LOCATION \
  --runner=DataflowRunner \
  --disk_size_gb=DISK_SIZE_GB \
  --experiments=use_runner_v2 \
  --sdk_container_image=IMAGE_URI

Go

wordcount --input gs://dataflow-samples/shakespeare/kinglear.txt \
          --output gs://<your-gcs-bucket>/counts \
          --runner dataflow \
          --project your-gcp-project \
          --region your-gcp-region \
          --temp_location gs://<your-gcs-bucket>/tmp/ \
          --staging_location gs://<your-gcs-bucket>/binaries/ \
          --sdk_container_image=IMAGE_URI

更改下列內容：

  • INPUT_FILE：執行範例時，Dataflow 讀取的 Cloud Storage 輸入路徑。
  • OUTPUT_FILE：範例管道寫入的 Cloud Storage 輸出路徑。這個檔案包含字數。
  • PROJECT_ID：專案的 ID。 Google Cloud
  • REGION：用於部署 Dataflow 工作的區域。
  • TEMP_LOCATION：Cloud Storage 路徑，供 Dataflow 暫存管道執行期間建立的臨時工作檔案。
  • DISK_SIZE_GB：選用。如果容器很大，請考慮增加預設開機磁碟大小，以免磁碟空間不足。
  • IMAGE_URI：SDK 自訂容器映像檔 URI。 請一律使用已加上版本資訊的容器 SHA 或標記。請勿使用 :latest 標記或可變動標記。