使用 Java 建立 Dataflow 管道

本文說明如何設定 Google Cloud 專案、使用 Java 適用的 Apache Beam SDK 建立範例管道，以及在 Dataflow 服務上執行範例管道。這個管道會從 Cloud Storage 讀取文字檔案、計算檔案中不重複的字數，然後將字數寫回 Cloud Storage。如要瞭解 WordCount 管道，請觀看「如何在 Apache Beam 中使用 WordCount」影片。

本教學課程需要使用 Maven，但您也可以將範例專案從 Maven 轉換為 Gradle。詳情請參閱「 選用：從 Maven 轉換為 Gradle」。

如要直接在 Google Cloud 控制台中，按照這項工作的逐步指南操作，請按一下「Guide me」(逐步引導)：

「Guide me」(逐步引導)

事前準備

  1. 登入 Google Cloud 帳戶。如果您是 Google Cloud新手，歡迎 建立帳戶，親自評估產品在實際工作環境中的成效。新客戶還能獲得價值 $300 美元的免費抵免額，可用於執行、測試及部署工作負載。

  2. 安裝 Google Cloud CLI。

  3. 若您採用的是外部識別資訊提供者 (IdP)，請先使用聯合身分登入 gcloud CLI。

  4. 執行下列指令，初始化 gcloud CLI：

    gcloud init

  5. 建立或選取 Google Cloud 專案。

    選取或建立專案所需的角色

    • 選取專案：選取專案時，不需要具備特定 IAM 角色，只要您已獲授角色，即可選取任何專案。
    • 建立專案：如要建立專案，您需要具備專案建立者角色 (roles/resourcemanager.projectCreator)，其中包含 resourcemanager.projects.create 權限。瞭解如何授予角色。

    • 建立 Google Cloud 專案：

      gcloud projects create PROJECT_ID

      將 PROJECT_ID 替換為您要建立的 Google Cloud 專案名稱。

    • 選取您建立的 Google Cloud 專案：

      gcloud config set project PROJECT_ID

      將 PROJECT_ID 替換為 Google Cloud 專案名稱。

  6. 確認專案已啟用計費功能 Google Cloud 。

  7. 啟用 Dataflow、Compute Engine、Cloud Logging、Cloud Storage、Google Cloud Storage JSON、BigQuery、Cloud Pub/Sub、Cloud Datastore 和 Cloud Resource Manager API：

    啟用 API 時所需的角色

    如要啟用 API，您需要具備服務使用情形管理員 IAM 角色 (roles/serviceusage.serviceUsageAdmin)，其中包含 serviceusage.services.enable 權限。瞭解如何授予角色。

    gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com

  8. 為使用者帳戶建立本機驗證憑證： 

    gcloud auth application-default login

    如果系統傳回驗證錯誤，且您使用外部識別資訊提供者 (IdP)，請確認您已 使用聯合身分登入 gcloud CLI。

  9. 將角色授予使用者帳戶。針對下列每個 IAM 角色，執行一次下列指令： roles/iam.serviceAccountUser 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

    更改下列內容：

    • PROJECT_ID：專案 ID。
    • USER_IDENTIFIER：使用者帳戶的 ID。 例如：myemail@example.com。
    • ROLE：授予使用者帳戶的 IAM 角色。

  10. 安裝 Google Cloud CLI。

  11. 若您採用的是外部識別資訊提供者 (IdP)，請先使用聯合身分登入 gcloud CLI。

  12. 執行下列指令，初始化 gcloud CLI：

    gcloud init

  13. 建立或選取 Google Cloud 專案。

    選取或建立專案所需的角色

    • 選取專案：選取專案時，不需要具備特定 IAM 角色，只要您已獲授角色，即可選取任何專案。
    • 建立專案：如要建立專案，您需要具備專案建立者角色 (roles/resourcemanager.projectCreator)，其中包含 resourcemanager.projects.create 權限。瞭解如何授予角色。

    • 建立 Google Cloud 專案：

      gcloud projects create PROJECT_ID

      將 PROJECT_ID 替換為您要建立的 Google Cloud 專案名稱。

    • 選取您建立的 Google Cloud 專案：

      gcloud config set project PROJECT_ID

      將 PROJECT_ID 替換為 Google Cloud 專案名稱。

  14. 確認專案已啟用計費功能 Google Cloud 。

  15. 啟用 Dataflow、Compute Engine、Cloud Logging、Cloud Storage、Google Cloud Storage JSON、BigQuery、Cloud Pub/Sub、Cloud Datastore 和 Cloud Resource Manager API：

    啟用 API 時所需的角色

    如要啟用 API，您需要具備服務使用情形管理員 IAM 角色 (roles/serviceusage.serviceUsageAdmin)，其中包含 serviceusage.services.enable 權限。瞭解如何授予角色。

    gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com cloudresourcemanager.googleapis.com

  16. 為使用者帳戶建立本機驗證憑證： 

    gcloud auth application-default login

    如果系統傳回驗證錯誤，且您使用外部識別資訊提供者 (IdP)，請確認您已 使用聯合身分登入 gcloud CLI。

  17. 將角色授予使用者帳戶。針對下列每個 IAM 角色，執行一次下列指令： roles/iam.serviceAccountUser 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

    更改下列內容：

    • PROJECT_ID：專案 ID。
    • USER_IDENTIFIER：使用者帳戶的 ID。 例如：myemail@example.com。
    • ROLE：授予使用者帳戶的 IAM 角色。
    18.

  18. 將角色授予 Compute Engine 預設服務帳戶。針對下列每個 IAM 角色，執行一次下列指令：

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.objectAdmin
    gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" --role=SERVICE_ACCOUNT_ROLE
    • 將 PROJECT_ID 替換為專案 ID。
    • 將 PROJECT_NUMBER 替換為專案編號。 如要找出專案編號，請參閱「識別專案」一文，或使用 gcloud projects describe 指令。
    • 將 SERVICE_ACCOUNT_ROLE 替換為各個角色。
  19. 建立 Cloud Storage bucket，然後依照下列指示來設定 bucket：
    • 將儲存空間級別設為 S (Standard)。
    • 將儲存空間位置設定為下列區域： US (美國)。
    • 將 BUCKET_NAME 替換成不重複的值區名稱。請勿在 bucket 名稱中加入任何機密資訊，因為 bucket 命名空間全域通用並會公開顯示。 
    gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US
  20. 複製下列內容，因為後續章節會用到：
    • Cloud Storage bucket 名稱。
    • 您的 Google Cloud 專案 ID。如要找出這個 ID，請參閱「識別專案」。
  21. 下載並安裝 Java Development Kit (JDK) 11 版。(Dataflow 仍支援第 8 版)。確認 JAVA_HOME 環境變數已設定，且指向您的 JDK 安裝。
  22. 按照適用於您作業系統的 Maven 安裝指南，下載並安裝 Apache Maven。

取得管道程式碼

Apache Beam SDK 是用於資料處理管道的開放原始碼程式設計模型。您可以使用 Apache Beam 程式來定義這些管道，並選擇 Dataflow 等執行器來執行管道。

  1. 在殼層或終端機中，使用 Maven 原型外掛程式在電腦上建立 Maven 專案，其中包含 Apache Beam SDK 的 WordCount 範例：
    mvn archetype:generate \
    -DarchetypeGroupId=org.apache.beam \
    -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
    -DarchetypeVersion=2.71.0 \
    -DgroupId=org.example \
    -DartifactId=word-count-beam \
    -Dversion="0.1" \
    -Dpackage=org.apache.beam.examples \
    -DinteractiveMode=false

    這項指令會在目前的目錄下建立名為 word-count-beam 的新目錄。word-count-beam 目錄包含簡易的 pom.xml 檔案，以及一系列計算文字檔字數的範例管道。

  2. 確認 word-count-beam 目錄包含 pom.xml 檔案：

    Linux 或 macOS

    cd word-count-beam/
ls

    輸出內容如下：

    pom.xml   src

    Windows

    cd word-count-beam/
dir

    輸出內容如下：

    pom.xml   src
  3. 確認 Maven 專案包含範例管道：

    Linux 或 macOS

    ls src/main/java/org/apache/beam/examples/

    輸出內容如下：

    DebuggingWordCount.java   WindowedWordCount.java   common
MinimalWordCount.java   WordCount.java

    Windows

    dir src/main/java/org/apache/beam/examples/

    輸出內容如下：

    DebuggingWordCount.java   WindowedWordCount.java   common
MinimalWordCount.java   WordCount.java

如要詳細瞭解這些範例中使用的 Apache Beam 概念，請參閱 Apache Beam WordCount 範例。下一節的操作說明會使用 WordCount.java。

在本機執行管道

  • 在殼層或終端機中，從 word-count-beam 目錄在本機執行 WordCount 管道：
    mvn compile exec:java \
    -Dexec.mainClass=org.apache.beam.examples.WordCount \
    -Dexec.args="--output=counts"

    輸出檔案的前置字串為 counts，並寫入 word-count-beam 目錄。其中包含輸入文字中的不重複字詞，以及每個字詞的出現次數。

在 Dataflow 服務上執行管道

  • 在殼層或終端機中，從 word-count-beam 目錄在 Dataflow 服務上建構並執行 WordCount 管道：
    mvn -Pdataflow-runner compile exec:java \
    -Dexec.mainClass=org.apache.beam.examples.WordCount \
    -Dexec.args="--project=PROJECT_ID \
    --gcpTempLocation=gs://BUCKET_NAME/temp/ \
    --output=gs://BUCKET_NAME/output \
    --runner=DataflowRunner \
    --region=REGION"

    更改下列內容：

    • PROJECT_ID：您的 Google Cloud 專案 ID
    • BUCKET_NAME：Cloud Storage bucket 的名稱
    • REGION：Dataflow 區域，例如 us-central1

查看結果

  1. 前往 Google Cloud 控制台的 Dataflow「Jobs」(工作) 頁面。

    前往「Jobs」(工作) 頁面

    「Jobs」(工作) 頁面會顯示所有可用工作的詳細資料，包括狀態。「wordcount」工作的「Status」(狀態) 一開始為「Running」(執行中)，然後更新為「Succeeded」(成功)。

  2. 在 Google Cloud 控制台，前往「Cloud Storage bucket」頁面。

    前往「Buckets」(值區) 頁面

    「Buckets」(值區) 頁面會顯示專案中所有儲存空間值區的清單。

  3. 按一下您建立的儲存空間 bucket。

    「Bucket details」(Bucket 詳細資料) 頁面會顯示 Dataflow 工作建立的輸出檔案和暫存檔案。

清除所用資源

為了避免系統向您的 Google Cloud 帳戶收取本頁面所用資源的費用，請刪除含有這些資源的 Google Cloud 專案。

刪除專案

如要避免付費，最簡單的方法就是刪除您為快速入門導覽建立的專案。 Google Cloud

  1. 前往 Google Cloud 控制台的「Manage resources」(管理資源) 頁面。

    前往「Manage resources」(管理資源)

  2. 在專案清單中選取要刪除的專案，然後點選「Delete」(刪除)。
  3. 在對話方塊中輸入專案 ID，然後按一下 [Shut down] (關閉) 以刪除專案。

刪除個別資源

如要保留本快速入門導覽課程中使用的專案，請刪除個別資源： Google Cloud

  1. 前往 Google Cloud 控制台的 Cloud Storage「Buckets」(值區) 頁面。

    前往「Buckets」(值區) 頁面

  2. 按一下要刪除的值區旁的核取方塊。
  3. 如要刪除值區，請依序點選 「Delete」(刪除)，然後按照指示操作。

  4. 撤銷您授予 Compute Engine 預設服務帳戶的角色。針對下列每個 IAM 角色執行一次下列指令：

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.objectAdmin
    gcloud projects remove-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=SERVICE_ACCOUNT_ROLE

  5. 選用：撤銷您建立的驗證憑證，並刪除本機憑證檔案。

    gcloud auth application-default revoke

  6. 選用：從 gcloud CLI 撤銷憑證。 

    gcloud auth revoke

後續步驟