使用自訂容器和 C++ 程式庫

在本教學課程中，您將建立管道，使用 C++ 程式庫搭配自訂容器，執行 Dataflow HPC 高度平行工作流程。本教學課程說明如何使用 Dataflow 和 Apache Beam 執行網格運算應用程式，這類應用程式需要將資料分散到在許多核心上執行的函式中。

本教學課程會說明如何先使用 Direct Runner 執行管道，然後使用 Dataflow Runner 執行管道。在本機執行管道，即可在部署前測試管道。

這個範例會使用 Cython 繫結和 GMP 程式庫的函式。無論使用哪個程式庫或繫結工具，您都可以將相同原則套用至管道。

您可以在 GitHub 取得範例程式碼。

目標

• 建立使用 C++ 程式庫搭配自訂容器的管道。

• 使用 Dockerfile 建構 Docker 容器映像檔。

• 將程式碼和依附元件封裝至 Docker 容器中。

• 在本機執行管道進行測試。

• 在分散式環境中執行管道。

費用

在本文件中，您會使用下列 Google Cloud的計費元件：

• Artifact Registry
• Cloud Build
• Cloud Storage
• Compute Engine
• Dataflow

如要根據預測用量估算費用，請使用 Pricing Calculator。

初次使用 Google Cloud 的使用者可能符合免費試用期資格。

完成本文所述工作後，您可以刪除建立的資源，避免繼續計費，詳情請參閱「清除所用資源」。

事前準備

登入 Google Cloud 帳戶。如果您是 Google Cloud新手，歡迎 建立帳戶，親自評估產品在實際工作環境中的成效。新客戶還能獲得價值 $300 美元的免費抵免額，可用於執行、測試及部署工作負載。
安裝 Google Cloud CLI。
若您採用的是外部識別資訊提供者 (IdP)，請先使用聯合身分登入 gcloud CLI。
執行下列指令，初始化 gcloud CLI：
gcloud init
建立或選取 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 專案名稱。
確認專案已啟用計費功能 Google Cloud 。
啟用 Cloud Storage、Cloud Storage JSON、Compute Engine、Dataflow、Resource Manager、Artifact Registry 和 Cloud Build API：
啟用 API 時所需的角色
如要啟用 API，您需要具備服務使用情形管理員 IAM 角色 (roles/serviceusage.serviceUsageAdmin)，其中包含 serviceusage.services.enable 權限。瞭解如何授予角色。
gcloud services enable compute.googleapis.com dataflow.googleapis.com storage_component storage_api cloudresourcemanager.googleapis.com artifactregistry.googleapis.com cloudbuild.googleapis.com
為使用者帳戶建立本機驗證憑證：
gcloud auth application-default login
如果系統傳回驗證錯誤，且您使用外部識別資訊提供者 (IdP)，請確認您已 使用聯合身分登入 gcloud CLI。
將角色授予使用者帳戶。針對下列每個 IAM 角色，執行一次下列指令： roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
更改下列內容：
• PROJECT_ID：專案 ID。
• USER_IDENTIFIER：使用者帳戶的 ID。 例如：[e-mail]。
• ROLE：授予使用者帳戶的 IAM 角色。

安裝 Google Cloud CLI。

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

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

gcloud init

建立或選取 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 專案名稱。

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

啟用 Cloud Storage、Cloud Storage JSON、Compute Engine、Dataflow、Resource Manager、Artifact Registry 和 Cloud Build API：

啟用 API 時所需的角色

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

gcloud services enable compute.googleapis.com dataflow.googleapis.com storage_component storage_api cloudresourcemanager.googleapis.com artifactregistry.googleapis.com cloudbuild.googleapis.com

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

gcloud auth application-default login

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

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

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

更改下列內容：

• PROJECT_ID：專案 ID。
• USER_IDENTIFIER：使用者帳戶的 ID。 例如：[e-mail]。
• ROLE：授予使用者帳戶的 IAM 角色。

1. 為新管道建立使用者代管的 worker 服務帳戶，並將必要角色授予該服務帳戶。

   1. 如要建立服務帳戶，請執行 gcloud iam service-accounts create 指令：

      gcloud iam service-accounts create parallelpipeline \
          --description="Highly parallel pipeline worker service account" \
          --display-name="Highly parallel data pipeline access"

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

      • roles/dataflow.admin
      • roles/dataflow.worker
      • roles/storage.objectAdmin
      • roles/artifactregistry.reader

      gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:parallelpipeline@PROJECT_ID.iam.gserviceaccount.com" --role=SERVICE_ACCOUNT_ROLE

      將 SERVICE_ACCOUNT_ROLE 替換為各個角色。

   3. 將可為服務帳戶建立存取權杖的角色授予 Google 帳戶：

      gcloud iam service-accounts add-iam-policy-binding parallelpipeline@PROJECT_ID.iam.gserviceaccount.com --member="user:EMAIL_ADDRESS" --role=roles/iam.serviceAccountTokenCreator

下載程式碼範例並變更目錄

下載程式碼範例，然後變更目錄。 GitHub 存放區中的程式碼範例提供執行這個管道所需的所有程式碼。準備好建構自己的管道時，可以將這段程式碼範例做為範本。

複製 beam-cpp-example 存放區。

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

   git clone https://github.com/GoogleCloudPlatform/dataflow-sample-applications.git
   

2. 切換至應用程式目錄：

   cd dataflow-sample-applications/beam-cpp-example
   

管道程式碼

您可以自訂本教學課程中的管道程式碼。這個 pipeline 會完成下列工作：

• 動態產生輸入範圍內的所有整數。
• 透過 C++ 函式執行整數，並篩除錯誤值。
• 將錯誤值寫入側邊管道。
• 計算每個停止時間的發生次數，並將結果正規化。
• 列印輸出內容、格式化結果，並將結果寫入文字檔。
• 建立具有單一元素的 PCollection。
• 使用 map 函式處理單一元素，並將頻率 PCollection 做為側邊輸入內容傳遞。
• 處理 PCollection 並產生單一輸出內容。

範例檔案如下所示：

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#


import argparse
import logging
import os
import sys


def run(argv):
  # Import here to avoid __main__ session pickling issues.
  import io
  import itertools
  import matplotlib.pyplot as plt
  import collatz

  import apache_beam as beam
  from apache_beam.io import restriction_trackers
  from apache_beam.options.pipeline_options import PipelineOptions

  class RangeSdf(beam.DoFn, beam.RestrictionProvider):
    """An SDF producing all the integers in the input range.

    This is preferable to beam.Create(range(...)) as it produces the integers
    dynamically rather than materializing them up front.  It is an SDF to do
    so with perfect dynamic sharding.
    """
    def initial_restriction(self, desired_range):
      start, stop = desired_range
      return restriction_trackers.OffsetRange(start, stop)

    def restriction_size(self, _, restriction):
      return restriction.size()

    def create_tracker(self, restriction):
      return restriction_trackers.OffsetRestrictionTracker(restriction)

    def process(self, _, active_range=beam.DoFn.RestrictionParam()):
      for i in itertools.count(active_range.current_restriction().start):
        if active_range.try_claim(i):
          yield i
        else:
          break

  class GenerateIntegers(beam.PTransform):
    def __init__(self, start, stop):
      self._start = start
      self._stop = stop

    def expand(self, p):
      return (
          p
          | beam.Create([(self._start, self._stop + 1)])
          | beam.ParDo(RangeSdf()))

  parser = argparse.ArgumentParser()
  parser.add_argument('--start', dest='start', type=int, default=1)
  parser.add_argument('--stop', dest='stop', type=int, default=10000)
  parser.add_argument('--output', default='./out.png')

  known_args, pipeline_args = parser.parse_known_args(argv)
  # Store this as a local to avoid capturing the full known_args.
  output_path = known_args.output

  with beam.Pipeline(options=PipelineOptions(pipeline_args)) as p:

    # Generate the integers from start to stop (inclusive).
    integers = p | GenerateIntegers(known_args.start, known_args.stop)

    # Run them through our C++ function, filtering bad records.
    # Requires apache beam 2.34 or later.
    stopping_times, bad_values = (
        integers
        | beam.Map(collatz.total_stopping_time).with_exception_handling(
            use_subprocess=True))

    # Write the bad values to a side channel.
    bad_values | 'WriteBadValues' >> beam.io.WriteToText(
        os.path.splitext(output_path)[0] + '-bad.txt')

    # Count the occurrence of each stopping time and normalize.
    total = known_args.stop - known_args.start + 1
    frequencies = (
        stopping_times
        | 'Aggregate' >> (beam.Map(lambda x: (x, 1)) | beam.CombinePerKey(sum))
        | 'Normalize' >> beam.MapTuple(lambda x, count: (x, count / total)))

    if known_args.stop <= 10:
      # Print out the results for debugging.
      frequencies | beam.Map(print)
    else:
      # Format and write them to a text file.
      (
          frequencies
          | 'Format' >> beam.MapTuple(lambda count, freq: f'{count}, {freq}')
          | beam.io.WriteToText(os.path.splitext(output_path)[0] + '.txt'))

    # Define some helper functions.
    def make_scatter_plot(xy):
      x, y = zip(*xy)
      plt.plot(x, y, '.')
      png_bytes = io.BytesIO()
      plt.savefig(png_bytes, format='png')
      png_bytes.seek(0)
      return png_bytes.read()

    def write_to_path(path, content):
      """Most Beam IOs write multiple elements to some kind of a container
      file (e.g. strings to lines of a text file, avro records to an avro file,
      etc.)  This function writes each element to its own file, given by path.
      """
      # Write to a temporary path and to a rename for fault tolerence.
      tmp_path = path + '.tmp'
      fs = beam.io.filesystems.FileSystems.get_filesystem(path)
      with fs.create(tmp_path) as fout:
        fout.write(content)
      fs.rename([tmp_path], [path])

    (
        p
        # Create a PCollection with a single element.
        | 'CreateSingleton' >> beam.Create([None])
        # Process the single element with a Map function, passing the frequency
        # PCollection as a side input.
        # This will cause the normally distributed frequency PCollection to be
        # colocated and processed as a single unit, producing a single output.
        | 'MakePlot' >> beam.Map(
            lambda _,
            data: make_scatter_plot(data),
            data=beam.pvalue.AsList(frequencies))
        # Pair this with the desired filename.
        |
        'PairWithFilename' >> beam.Map(lambda content: (output_path, content))
        # And actually write it out, using MapTuple to split the tuple into args.
        | 'WriteToOutput' >> beam.MapTuple(write_to_path))


if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  run(sys.argv)

設定開發環境

1. 使用 Python 適用的 Apache Beam SDK。

2. 安裝 GMP 程式庫：

   apt-get install libgmp3-dev
   

3. 如要安裝依附元件，請使用 requirements.txt 檔案。

   pip install -r requirements.txt
   

4. 如要建構 Python 繫結，請執行下列指令。

   python setup.py build_ext --inplace
   

您可以透過本教學課程自訂 requirements.txt 檔案。範例檔案包含下列依附元件：

#
#    Licensed to the Apache Software Foundation (ASF) under one or more
#    contributor license agreements.  See the NOTICE file distributed with
#    this work for additional information regarding copyright ownership.
#    The ASF licenses this file to You under the Apache License, Version 2.0
#    (the "License"); you may not use this file except in compliance with
#    the License.  You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

apache-beam[gcp]==2.46.0
cython==0.29.24
pyparsing==2.4.2
matplotlib==3.4.3

在本機執行管道

在本機執行管道有助於測試。 在本機執行管道，確認管道執行及運作方式符合預期後，再將管道部署至分散式環境。

您可以使用下列指令在本機執行管道。 這個指令會輸出名為 out.png 的圖片。

python pipeline.py

建立 Google Cloud 資源

本節說明如何建立下列資源：

• 做為暫時儲存位置和輸出位置的 Cloud Storage bucket。
• 用於封裝管道程式碼和依附元件的 Docker 容器。

建立 Cloud Storage 值區

首先，請使用 Google Cloud CLI 建立 Cloud Storage bucket。這個 bucket 是 Dataflow 管道的暫時儲存位置。

如要建立 bucket，請使用 gcloud storage buckets create 指令：

gcloud storage buckets create gs://BUCKET_NAME --location=LOCATION

更改下列內容：

• BUCKET_NAME：Cloud Storage bucket 的名稱，必須符合 bucket 命名規定。Cloud Storage bucket 名稱不得重複。
• LOCATION：值區的位置。

建立及建構容器映像檔

您可以自訂本教學課程的 Dockerfile。 範例檔案如下所示：

注意：在分散式環境中執行管道時，您可能需要更新 Dockerfile 中的 Python 版本。使用自訂容器映像檔時，映像檔中的 Python 解譯器子版本必須與管道建構時使用的版本相符。

FROM apache/beam_python3.9_sdk:2.46.0

# Install a C++ library.
RUN apt-get update
RUN apt-get install -y libgmp3-dev

# Install Python dependencies.
COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt

# Install the code and some python bindings.
COPY pipeline.py pipeline.py
COPY collatz.pyx collatz.pyx
COPY setup.py setup.py
RUN python setup.py install

這個 Dockerfile 包含 FROM、COPY 和 RUN 指令，您可以在 Dockerfile 參考資料中瞭解這些指令。

1. 如要上傳構件，請建立 Artifact Registry 存放區。每個存放區只能包含單一支援格式的構件。

   所有存放區內容都會使用 Google-owned and Google-managed encryption keys 或客戶自行管理的加密金鑰加密。Artifact Registry 預設使用Google-owned and Google-managed encryption keys ，不需要為這個選項進行任何設定。

   您必須至少具備存放區的 Artifact Registry 寫入者存取權。

   執行下列指令來建立新的存放區。這個指令會使用 --async 旗標並立即傳回，不會等待執行中的作業完成。

   gcloud artifacts repositories create REPOSITORY \
      --repository-format=docker \
      --location=LOCATION \
      --async
   

   將 REPOSITORY 替換為存放區名稱。專案中每個存放區位置的存放區名稱不得重複。

2. 建立 Dockerfile。

   如要讓套件成為 Apache Beam 容器的一部分，必須在 requirements.txt 檔案中指定這些套件。請勿在 requirements.txt 檔案中指定 apache-beam。Apache Beam 容器已包含 apache-beam。

   注意：使用 Dockerfile 安裝的依附元件只會在啟動管道時提供。需求檔案中必須包含工作依附元件。如範例所示，這個檔案不需要包含 Apache Beam，建議您將兩者分開，以便加快啟動速度。

3. 推送或提取映像檔前，請先設定 Docker，驗證傳送至 Artifact Registry 的要求。如要為 Docker 存放區設定驗證機制，請執行下列指令：

   gcloud auth configure-docker LOCATION-docker.pkg.dev
   

   這個指令會更新 Docker 設定。您現在可以在 Google Cloud 專案中連結 Artifact Registry，以推送映像檔。

4. 使用 Cloud Build 建構 Docker 映像檔 Dockerfile。

   更新下列指令中的路徑，與您建立的 Dockerfile 相符。這個指令會建構檔案，並推送至 Artifact Registry 存放區。

   gcloud builds submit --tag LOCATION-docker.pkg.dev/PROJECT_ID/REPOSITORY/dataflow/cpp_beam_container:latest .
   

將程式碼和依附元件封裝至 Docker 容器中

1. 如要在分散式環境中執行這項管道，請將程式碼和依附元件封裝到 Docker 容器中。

   docker build . -t cpp_beam_container
   

2. 封裝程式碼和依附元件後，您可以在本機執行管道進行測試。

   python pipeline.py \
      --runner=PortableRunner \
      --job_endpoint=embed \
      --environment_type=DOCKER \
      --environment_config="docker.io/library/cpp_beam_container"
   

   這項指令會將輸出內容寫入 Docker 映像檔。如要查看輸出內容，請使用 --output 執行管道，並將輸出內容寫入 Cloud Storage bucket。例如，執行下列指令。

   python pipeline.py \
      --runner=PortableRunner \
      --job_endpoint=embed \
      --environment_type=DOCKER \
      --environment_config="docker.io/library/cpp_beam_container" \
      --output=gs://BUCKET_NAME/out.png
   

執行管道

現在，您可以參照含有管道程式碼的檔案，並傳遞管道所需的參數，在 Dataflow 中執行 Apache Beam 管道。

在殼層或終端機中，使用 Dataflow Runner 執行管道。

python pipeline.py \
    --runner=DataflowRunner \
    --project=PROJECT_ID \
    --region=REGION \
    --temp_location=gs://BUCKET_NAME/tmp \
    --sdk_container_image="LOCATION-docker.pkg.dev/PROJECT_ID/REPOSITORY/dataflow/cpp_beam_container:latest" \
    --experiment=use_runner_v2 \
    --output=gs://BUCKET_NAME/out.png

執行指令來執行管道後，Dataflow 會傳回工作 ID，工作狀態為「已加入佇列」。工作狀態可能需要幾分鐘才會變成「Running」，之後您就能存取工作圖。

查看結果

查看寫入 Cloud Storage bucket 的資料。使用 gcloud storage ls 指令列出值區頂層的內容：

gcloud storage ls gs://BUCKET_NAME

如果成功，指令會傳回類似以下的訊息：

gs://BUCKET_NAME/out.png

清除所用資源

為避免因為本教學課程所用資源，導致系統向 Google Cloud 收取費用，請刪除含有相關資源的專案，或者保留專案但刪除個別資源。

刪除專案

如要避免付費，最簡單的方法就是刪除您為了本教學課程所建立的專案。 Google Cloud

注意：刪除專案會造成以下結果：

• 專案中的所有內容都會遭到刪除。如果使用現有專案來進行本文中的任務，刪除專案將一併移除當中已完成的其他任務'。
• 自訂專案 ID 會消失。當您之前建立這個專案時，可能建立了想要在日後使用的自訂專案 ID。如要保留使用該專案 ID 的網址 (例如 appspot.com 網址)，請刪除在該專案中選取的資源，而不是刪除整個專案。

如果打算探索多種架構、教學課程或快速入門導覽課程，重複使用專案可避免超出專案配額限制。

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

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

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

刪除個別資源

如要重複使用專案，請刪除您為本教學課程建立的資源。

清理 Google Cloud 專案資源

1. 刪除 Artifact Registry 存放區。

   gcloud artifacts repositories delete REPOSITORY \
      --location=LOCATION --async
   

2. 刪除 Cloud Storage bucket 和其中的物件。單獨使用這個 bucket 不會產生任何費用。

   gcloud storage rm gs://BUCKET_NAME --recursive
   

撤銷憑證

1. 撤銷您授予使用者管理的 Worker 服務帳戶的角色。 針對下列每個 IAM 角色執行一次下列指令：

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.objectAdmin
   • roles/artifactregistry.reader

   gcloud projects remove-iam-policy-binding PROJECT_ID \
     --member=serviceAccount:parallelpipeline@PROJECT_ID.iam.gserviceaccount.com \
     --role=SERVICE_ACCOUNT_ROLE

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

   gcloud auth application-default revoke

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

   gcloud auth revoke

後續步驟

• 查看 GitHub 上的範例應用程式。
• 在 Dataflow 中使用自訂容器。
• 進一步瞭解如何搭配 Apache Beam 使用容器環境。
• 查看 Google Cloud 的參考架構、圖表和最佳做法。歡迎瀏覽我們的 Cloud Architecture Center。