使用 Go 建立 Dataflow 管道

本頁說明如何使用 Go 適用的 Apache Beam SDK 建構定義管道的程式。接著，您會在本地和 Dataflow 服務上執行管道。如要瞭解 WordCount 管道，請觀看「如何在 Apache Beam 中使用 WordCount」影片。

事前準備

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

  2. Install the Google Cloud CLI.

  3. If you're using an external identity provider (IdP), you must first sign in to the gcloud CLI with your federated identity.

  4. To initialize the gcloud CLI, run the following command: 

    gcloud init

  5. Create or select a Google Cloud project.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

  6. Make sure that billing is enabled for your Google Cloud project.

  7. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs: 

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

  8. Create local authentication credentials for your user account: 

    gcloud auth application-default login

    If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.

  9. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:myemail@example.com.

    • Replace ROLE with each individual role.

  10. Install the Google Cloud CLI.

  11. If you're using an external identity provider (IdP), you must first sign in to the gcloud CLI with your federated identity.

  12. To initialize the gcloud CLI, run the following command: 

    gcloud init

  13. Create or select a Google Cloud project.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

  14. Make sure that billing is enabled for your Google Cloud project.

  15. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs: 

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

  16. Create local authentication credentials for your user account: 

    gcloud auth application-default login

    If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.

  17. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:myemail@example.com.

    • Replace ROLE with each individual role.
    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. Create a Cloud Storage bucket and configure it as follows:
    • Set the storage class to S (標準)。
    • 將儲存空間位置設定為下列項目： US (美國)。
    • 將 BUCKET_NAME 替換成 不重複的值區名稱。請勿在值區名稱中加入任何機密資訊，因為值區命名空間屬於全域性質，而且會公開顯示。 
      • gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US
    • 複製 Google Cloud 專案 ID 和 Cloud Storage 值區名稱。 您會在後續步驟中用到這些值。

設定開發環境

Apache Beam SDK 是一種用於資料管道的開放原始碼程式設計模型。您可以使用 Apache Beam 程式定義管道，然後選擇執行器 (例如 Dataflow) 來執行管道。

使用 Go 適用的 Apache Beam SDK 時，建議您使用最新版 Go。如果尚未安裝最新版 Go，請參閱 Go 的下載及安裝指南，下載並安裝適用於您作業系統的 Go。

如要確認已安裝的 Go 版本，請在本機終端機執行下列指令：

go version

執行 Beam 字數統計範例

Go 適用的 Apache Beam SDK 包含wordcount管道範例。wordcount 範例會執行下列動作：

  1. 讀取文字檔案做為輸入。根據預設，這個範例會讀取位於 Cloud Storage 值區的文字檔，資源名稱為 gs://dataflow-samples/shakespeare/kinglear.txt。
  2. 將每一行剖析為字詞。
  3. 對代碼化的字詞執行頻率計數。

如要在本機電腦上執行最新版本的 Beam wordcount 範例，請按照下列步驟操作：

  1. 使用 git clone 指令複製 apache/beam GitHub 存放區：

    git clone https://github.com/apache/beam.git

  2. 切換至 beam/sdks/go 目錄：

    cd beam/sdks/go

  3. 使用下列指令執行管道：

    go run examples/wordcount/wordcount.go \
  --input gs://dataflow-samples/shakespeare/kinglear.txt \
  --output outputs

    input 標記會指定要讀取的檔案，output 標記則會指定頻率計數輸出的檔案名稱。

管道完成後，請查看輸出結果：

more outputs*

如要退出，請按 q。

修改管道程式碼

Beam wordcount 管道會區分大小寫字詞。下列步驟說明如何建立自己的 Go 模組、修改 wordcount 管道，讓管道不區分大小寫，並在 Dataflow 上執行。

建立 Go 模組

如要變更管道程式碼，請按照下列步驟操作。

  1. 在所選位置為 Go 模組建立目錄：

    mkdir wordcount
cd wordcount

  2. 建立 Go 模組。在本範例中，請使用 example/dataflow 做為模組路徑。

    go mod init example/dataflow

  3. 從 Apache Beam GitHub 存放區下載 wordcount 程式碼的最新副本。將這個檔案放入您建立的 wordcount 目錄。

  4. 如果您使用非 Linux 作業系統，則必須取得 Go unix 套件。必須有這個套件，才能在 Dataflow 服務上執行管道。

    go get -u golang.org/x/sys/unix

  5. 確認 go.mod 檔案與模組的原始碼相符：

    go mod tidy

執行未修改的管道

確認未經修改的 wordcount 管道可在本機執行。

  1. 在終端機中，建構並在本機執行管道：

     go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
     --output outputs

  2. 查看輸出結果：

     more outputs*

  3. 如要退出，請按 q。

變更管道程式碼

如要變更管道，使其不區分大小寫，請修改程式碼，將 strings.ToLower 函式套用至所有字詞。

  1. 在您選擇的編輯器中開啟 wordcount.go 檔案。

  2. 檢查 init 區塊 (為求清楚起見，已移除註解)：

     func init() {
   register.DoFn3x0[context.Context, string, func(string)](&extractFn{})
   register.Function2x1(formatFn)
   register.Emitter1[string]()
 }

  3. 新增一行程式碼，註冊 strings.ToLower 函式：

     func init() {
   register.DoFn3x0[context.Context, string, func(string)](&extractFn{})
   register.Function2x1(formatFn)
   register.Emitter1[string]()
   register.Function1x1(strings.ToLower)
 }

  4. 檢查 CountWords 函式：

     func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {
   s = s.Scope("CountWords")

   // Convert lines of text into individual words.
   col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines)

   // Count the number of times each word occurs.
   return stats.Count(s, col)
 }

  5. 如要將字詞轉換為小寫，請新增 ParDo，並對每個字詞套用 strings.ToLower：

     func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {
   s = s.Scope("CountWords")

   // Convert lines of text into individual words.
   col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines)

   // Map all letters to lowercase.
   lowercaseWords := beam.ParDo(s, strings.ToLower, col)

   // Count the number of times each word occurs.
   return stats.Count(s, lowercaseWords)
 }

  6. 儲存檔案。

在本機執行更新後的管道

在本機執行更新後的 wordcount pipeline，並確認輸出內容已變更。

  1. 建構並執行修改後的 wordcount 管道：

     go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
     --output outputs

  2. 查看修改後管道的輸出結果。所有字詞都應為小寫。

     more outputs*

  3. 如要退出，請按 q。

在 Dataflow 服務上執行管道

如要在 Dataflow 服務上執行更新後的 wordcount 範例，請使用下列指令：

go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner dataflow \
    --project PROJECT_ID \
    --region DATAFLOW_REGION \
    --staging_location gs://BUCKET_NAME/binaries/

更改下列內容：

  • BUCKET_NAME：Cloud Storage bucket 名稱。

  • PROJECT_ID：專案 ID。 Google Cloud

  • DATAFLOW_REGION：要部署 Dataflow 工作的區域。例如：europe-west1。 如需可用位置清單，請參閱 Dataflow 位置。--region 標記會覆寫中繼資料伺服器、本機用戶端或環境變數中設定的預設地區。

查看結果

您可以在Google Cloud 控制台中查看 Dataflow 工作清單。前往 Google Cloud 控制台的 Dataflow「Jobs」(工作) 頁面。

前往「Jobs」(工作) 頁面

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

透過 Dataflow 執行管道時，結果會儲存於 Cloud Storage 值區。使用Google Cloud 控制台或本機終端機查看輸出結果。

控制台

如要在 Google Cloud 控制台中查看結果，請前往 Cloud Storage 的「Bucket」頁面。

前往「Buckets」(值區) 頁面

在專案的值區清單中，按一下您先前建立的儲存空間值區。工作建立的輸出檔案會顯示在 results 目錄中。

終端機

從終端機或使用 Cloud Shell 查看結果。

  1. 如要列出輸出檔案，請使用 gcloud storage ls 指令：

    gcloud storage ls gs://BUCKET_NAME/results/outputs* --long

    將 BUCKET_NAME 替換為指定輸出 Cloud Storage 值區的名稱。

  2. 如要查看輸出檔案中的結果，請使用 gcloud storage cat 指令：

    gcloud storage cat gs://BUCKET_NAME/results/outputs*

清除所用資源

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

  1. In the Google Cloud console, go to the Cloud Storage Buckets page.

    Go to Buckets

  2. Click the checkbox for the bucket that you want to delete.
  3. To delete the bucket, click Delete, and then follow the instructions.

  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. Optional: Revoke the authentication credentials that you created, and delete the local credential file. 

    gcloud auth application-default revoke

  6. Optional: Revoke credentials from the gcloud CLI. 

    gcloud auth revoke

後續步驟