使用 Python 建立 Dataflow 管道

本文說明如何使用 Apache Beam SDK for Python 建構程式，定義管道。接著，您可以使用直接本機執行器或雲端執行器 (例如 Dataflow) 執行管道。如要瞭解 WordCount 管道，請觀看「How to use WordCount in Apache Beam」影片。

如要直接在 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，您必須具備 serviceusage.services.enable 權限。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「服務使用情形管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。 

    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.supportUser, roles/datastream.admin, roles/monitoring.metricsScopesViewer, roles/cloudaicompanion.settingsAdmin 

    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，您必須具備 serviceusage.services.enable 權限。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「服務使用情形管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。 

    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.supportUser, roles/datastream.admin, roles/monitoring.metricsScopesViewer, roles/cloudaicompanion.settingsAdmin 

    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，並設定如下：
    • 將儲存空間級別設為 S (Standard)。
    • 將儲存空間位置設定為下列區域： US (美國)。
    • 將 BUCKET_NAME 替換成不重複的 bucket 名稱。請勿在 bucket 名稱中加入任何機密資訊，因為 bucket 命名空間全域通用並會公開顯示。 
    gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US
  20. 複製 Google Cloud 專案 ID 和 Cloud Storage bucket 名稱。您稍後會在本文件中用到這些值。

設定環境

在本節中，請使用命令提示字元設定獨立的 Python 虛擬環境，以便使用 venv 執行管道專案。這樣一來，您就能將一個專案的依附元件與其他專案的依附元件隔離。

如果沒有現成可用的命令提示字元，則可使用 Cloud Shell。Cloud Shell 已安裝 Python 3 的套件管理工具，因此您可以跳過這個步驟，直接建立虛擬環境。

如要安裝 Python 並建立虛擬環境，請按照下列步驟操作：

  1. 確認系統中已安裝 Python 3 和 pip： 
    python --version
python -m pip --version
  2. 視需要安裝 Python 3，然後設定 Python 虛擬環境：按照「 設定 Python 開發環境」頁面的「安裝 Python」和「設定 venv」部分提供的操作說明進行。

完成快速入門後，您可以執行 deactivate 來停用虛擬環境。

取得 Apache Beam SDK

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

如要下載及安裝 Apache Beam SDK，請按照下列步驟操作：

  1. 確認您位於前一節建立的 Python 虛擬環境中。 確認提示以 <env_name> 開頭，其中 env_name 是虛擬環境的名稱。
  2. 安裝最新版本的 Apache Beam SDK for Python：
    3. pip install apache-beam[gcp]

在本機執行管道

如要瞭解管道在本機的執行方式，請使用 apache_beam 套件隨附的 wordcount 範例，其中包含現成的 Python 模組。

wordcount 管道範例會執行下列作業：

  1. 以文字檔做為輸入內容。

    這個文字檔位於資源名稱為 gs://dataflow-samples/shakespeare/kinglear.txt 的 Cloud Storage bucket 中。

  2. 將每一行剖析為字詞。
  3. 對代碼化的字詞執行頻率計數。

如要在本機暫存 wordcount 管道，請按照下列步驟操作：

  1. 在本機終端機中執行 wordcount 範例： 
    python -m apache_beam.examples.wordcount \
  --output outputs
  2. 查看管道的輸出內容： 
    more outputs*
  3. 如要退出，請按 q 鍵。
在本機執行管道可測試及偵錯 Apache Beam 程式。 您可以在 Apache Beam GitHub 上查看wordcount.py原始碼。

在 Dataflow 服務上執行管道

在本節中，請在 Dataflow 服務上執行 apache_beam 套件中的 wordcount 範例管道。這個範例會將 DataflowRunner 指定為 --runner 的參數。
  • 執行管道： 
    python -m apache_beam.examples.wordcount \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://BUCKET_NAME/tmp/

    更改下列內容：

    • DATAFLOW_REGION：要部署 Dataflow 工作的區域，例如 europe-west1

      --region 旗標會覆寫中繼資料伺服器、本機用戶端或環境變數中設定的預設區域。

    • BUCKET_NAME：您先前複製的 Cloud Storage bucket 名稱
    • PROJECT_ID：您先前複製的 Google Cloud 專案 ID

查看結果

透過 Dataflow 執行管道時，結果會儲存在 Cloud Storage bucket。 在本節中，請使用 Google Cloud 控制台或本機終端機，確認管道正在執行。

Google Cloud 控制台

如要在 Google Cloud 控制台中查看結果，請按照下列步驟操作：

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

    前往「Jobs」(工作) 頁面

    「Jobs」(工作) 頁面會顯示 wordcount 工作的詳細資料，包括一開始的「執行中」狀態，以及後來的「已完成」狀態。

  2. 前往 Cloud Storage 的「Buckets」(值區) 頁面。

    前往「Buckets」(值區) 頁面

  3. 在專案的值區清單中，按一下您先前建立的儲存空間值區。

    在 wordcount 目錄中，會顯示工作建立的輸出檔案。

本機終端機

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

  1. 如要列出輸出檔案，請使用 gcloud storage ls 指令： 
    gcloud storage ls gs://BUCKET_NAME/results/outputs* --long

    2. 請將 BUCKET_NAME 改成管道程式中使用的 Cloud Storage bucket 名稱。

  2. 如要在輸出檔案中查看結果，請使用 gcloud storage cat 指令：
    gcloud storage cat gs://BUCKET_NAME/results/outputs*

修改管道程式碼

前例中的 wordcount 管道會區分大小寫字詞。 下列步驟說明如何修改管道，讓 wordcount 管道不區分大小寫。
  1. 在本機電腦上，從 Apache Beam GitHub 存放區下載最新版本的wordcount程式碼。
  2. 從本機終端機執行管道： 
    python wordcount.py --output outputs
  3. 查看結果： 
    more outputs*
  4. 如要退出，請按 q 鍵。
  5. 在您選用的編輯器中開啟 wordcount.py 檔案。
  6. 在 run 函式中，檢查管道步驟： 
    counts = (
        lines
        | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))

    在 split 之後，這些行會分割成字串形式的字詞。

  7. 如要將字串轉換為小寫，請修改 split 後的行： 
    counts = (
        lines
        | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
        | 'lowercase' >> beam.Map(str.lower)
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))
    這項修改會將 str.lower 函式對應至每個字詞。這行程式碼等同於 beam.Map(lambda word: str.lower(word))。
  8. 儲存檔案，然後執行修改後的 wordcount 工作： 
    python wordcount.py --output outputs
  9. 查看修改後的管道結果： 
    more outputs*
  10. 如要退出，請按 q 鍵。
  11. 在 Dataflow 服務上執行修改後的管道： 
    python wordcount.py \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://BUCKET_NAME/tmp/

    更改下列內容：

    • DATAFLOW_REGION：要部署 Dataflow 工作的區域
    • BUCKET_NAME：Cloud Storage bucket 名稱
    • PROJECT_ID：您的 Google Cloud 專案 ID

清除所用資源

為了避免系統向您的 Google Cloud 帳戶收取本頁面所用資源的費用，請刪除含有這些資源的 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

後續步驟