將 Kafka 訊息寫入 BigQuery

本教學課程說明如何使用 Kafka Connect，將 Managed Service for Apache Kafka 叢集的訊息寫入 BigQuery。

在本教學課程中，您將建立 Connect 叢集，然後使用 BigQuery 接收器連接器寫入現有的 BigQuery 資料表。在此情境中，BigQuery 資料表會定義 Kafka 訊息的結構定義，Kafka 訊息必須符合資料表結構定義。詳情請參閱「BigQuery 接收器連接器的結構定義」。

事前準備

控制台

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

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Managed Kafka API.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the API

  5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Managed Kafka API.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the API

    8.

  8. 請確認您在專案中擁有下列一或多個角色： 代管 Kafka 叢集編輯者、 代管 Kafka Connect 叢集編輯者、 代管 Kafka 連接器編輯者、 代管 Kafka 主題編輯者、 BigQuery 資料擁有者

    檢查角色

    1. 前往 Google Cloud 控制台的「IAM」頁面。

      前往「IAM」頁面
    2. 選取專案。

    3. 在「主體」欄中，找出所有識別您或您所屬群組的資料列。如要瞭解自己所屬的群組，請與管理員聯絡。

    4. 針對指定或包含您的所有列，請檢查「角色」欄，確認角色清單是否包含必要角色。

    授予角色

    1. 前往 Google Cloud 控制台的「IAM」頁面。

      前往「IAM」頁面
    2. 選取專案。
    3. 按一下「Grant access」(授予存取權)。

    4. 在「New principals」(新增主體) 欄位中，輸入您的使用者 ID。 這通常是指 Google 帳戶的電子郵件地址。

    5. 按一下「選取角色」，然後搜尋角色。
    6. 如要授予其他角色，請按一下「Add another role」(新增其他角色)，然後新增其他角色。
    7. 按一下「Save」(儲存)。

gcloud

  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. 啟用 Managed Kafka API：

    啟用 API 時所需的角色

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

    gcloud services enable managedkafka.googleapis.com

  8. 安裝 Google Cloud CLI。

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

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

    gcloud init

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

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

  13. 啟用 Managed Kafka API：

    啟用 API 時所需的角色

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

    gcloud services enable managedkafka.googleapis.com
    14.

  14. 將角色授予使用者帳戶。針對下列每個 IAM 角色，執行一次下列指令： roles/managedkafka.clusterEditor, roles/managedkafka.connectClusterEditor, roles/managedkafka.connectorEditor, roles/managedkafka.topicEditor, roles/bigquery.dataOwner 

    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 角色。

建立 BigQuery 資料表

在這個步驟中，將建立具有下列結構定義的 BigQuery 資料表：

資料欄名稱 資料類型
name STRING
id INTEGER

建立資料集

如要建立 BigQuery 資料集，請按照下列步驟操作：

控制台

  1. 開啟「BigQuery」BigQuery頁面。

    前往 BigQuery 頁面

  2. 在「Explorer」面板中，選取要建立資料集的專案。

  3. 展開 「查看動作」選項，然後點選「建立資料集」。

  4. 在「Create dataset」(建立資料集) 頁面：

    • 在「Dataset ID」(資料集 ID) 中輸入資料集名稱。

    • 針對「位置類型」，選擇資料集的地理位置。

gcloud

如要建立新的資料集，請使用 bq mk 指令，並加上 --dataset 旗標。

bq mk --location REGION \
  --dataset PROJECT_ID:DATASET_NAME

更改下列內容：

  • PROJECT_ID：專案 ID
  • DATASET_NAME：資料集名稱
  • REGION：資料集的位置

詳情請參閱「建立資料集」。

建立含有結構定義的資料表

接著，建立具有結構定義的新 BigQuery 資料表：

控制台

  1. 前往「BigQuery」頁面

    前往 BigQuery 頁面

  2. 在「Explorer」面板中展開專案，然後選取資料集。

  3. 在「Dataset」(資料集) 資訊部分，按一下「Create table」(建立資料表)。

  4. 在「Create table from」(建立資料表來源) 清單中，選取「Empty table」(空白資料表)。

  5. 在「Table」(資料表) 方塊中，輸入資料表名稱。

  6. 在「Schema」(結構定義) 部分中，按一下「Edit as Text」(以文字形式編輯)。

  7. 貼上下列結構定義：

    name:STRING,
id:INTEGER

  8. 按一下「Create table」(建立資料表)。

gcloud

如要建立新資料表，請使用 bq mk 指令搭配 --table 旗標。

bq mk --table \
  PROJECT_ID:DATASET_NAME.TABLE_NAME \
  name:STRING,id:INTEGER

更改下列內容：

  • PROJECT_ID：專案 ID
  • DATASET_NAME：資料集名稱
  • TABLE_NAME：要建立的資料表名稱

詳情請參閱「建立含有結構定義的空白資料表」。

根據預設，BigQuery 接收器連接器會使用主題名稱做為 BigQuery 資料表名稱。如要覆寫這個行為，請設定 topic2TableMap 設定屬性。詳情請參閱「BigQuery Sink 連接器的運作方式」。

建立 Managed Service for Apache Kafka 資源

在本節中，您將建立下列 Managed Service for Apache Kafka 資源：

  • 具有主題的 Kafka 叢集。
  • 具有 BigQuery 接收器連接器的 Connect 叢集。

建立 Kafka 叢集

在這個步驟中，您會建立 Managed Service for Apache Kafka 叢集。建立叢集最多可能需要 30 分鐘。

控制台

  1. 前往「Managed Service for Apache Kafka」>「Clusters」(叢集) 頁面。

    前往「Clusters」(叢集)

  2. 點選 「Create」(建立)。

  3. 在「Cluster name」(叢集名稱) 方塊中輸入叢集的名稱。

  4. 在「Region」(區域) 清單中，選取叢集的位置。選擇與 BigQuery 資料表相同的區域。

  5. 在「網路設定」部分，設定可存取叢集的子網路：

    1. 在「Project」(專案) 部分，選取專案。
    2. 在「Network」(網路) 中選取虛擬私有雲網路。
    3. 在「Subnet」(子網路) 中，選取子網路。
    4. 按一下 [完成]。

  6. 點選「建立」。

叢集建立期間，叢集狀態為 Creating。叢集建立完成後，狀態會顯示為 Active。

gcloud

如要建立 Kafka 叢集，請執行 managed-kafka clusters create 指令。

gcloud managed-kafka clusters create KAFKA_CLUSTER \
--location=REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \
--async

更改下列內容：

  • KAFKA_CLUSTER：Kafka 叢集名稱
  • REGION：叢集位置，請選擇與 BigQuery 資料表相同的區域
  • PROJECT_ID：專案 ID
  • SUBNET_NAME：要建立叢集的子網路，例如 default

如要瞭解支援的位置，請參閱「Managed Service for Apache Kafka 位置」。

這項指令會以非同步方式執行，並傳回作業 ID：

Check operation [projects/PROJECT_ID/locations/REGION/operations/OPERATION_ID] for status.

如要追蹤建立作業的進度，請使用 gcloud managed-kafka operations describe 指令：

gcloud managed-kafka operations describe OPERATION_ID \
  --location=REGION

詳情請參閱「監控叢集建立作業」。

建立 Kafka 主題

建立 Managed Service for Apache Kafka 叢集後，請建立 Kafka 主題。主題名稱應與 BigQuery 資料表名稱相同。

控制台

  1. 前往「Managed Service for Apache Kafka」>「Clusters」(叢集) 頁面。

    前往「Clusters」(叢集)

  2. 按一下叢集名稱。

  3. 在叢集詳細資料頁面中，按一下 「Create Topic」(建立主題)。

  4. 在「Topic name」(主題名稱) 方塊中，輸入與 BigQuery 資料表相同的名稱。

  5. 點選「建立」。

gcloud

如要建立主題，請執行 managed-kafka topics create 指令。

gcloud managed-kafka topics create TABLE_NAME \
--cluster=KAFKA_CLUSTER \
--location=REGION \
--partitions=10 \
--replication-factor=3

更改下列內容：

  • TABLE_NAME：BigQuery 資料表名稱，也就是主題名稱
  • KAFKA_CLUSTER：Kafka 叢集的名稱
  • REGION：您建立 Kafka 叢集的區域

建立連結叢集

在這個步驟中，您會建立 Connect 叢集。建立 Connect 叢集最多可能需要 30 分鐘。

開始這個步驟前，請確認 Managed Service for Apache Kafka 叢集已完全建立。

控制台

  1. 前往「Managed Service for Apache Kafka」>「Connect Clusters」(連線叢集) 頁面。

    前往「Connect Clusters」(連結叢集)

  2. 點選 「Create」(建立)。

  3. 在「Connect cluster name」(Connect 叢集名稱) 中輸入字串。例如：my-connect-cluster。

  4. 在「主要 Kafka 叢集」中，選取您先前建立的 Kafka。

  5. 點選「建立」。

叢集建立期間，叢集狀態為 Creating。叢集建立完成後，狀態會顯示為 Active。

gcloud

如要建立 Connect 叢集，請執行 gcloud managed-kafka connect-clusters create 指令。

gcloud managed-kafka connect-clusters create CONNECT_CLUSTER \
  --location=REGION \
  --cpu=12 \
  --memory=12GiB \
  --primary-subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \
  --kafka-cluster=KAFKA_CLUSTER \
  --async

更改下列內容：

  • CONNECT_CLUSTER：Connect 叢集名稱
  • REGION：您建立 Kafka 叢集的區域
  • PROJECT_ID：專案 ID
  • SUBNET_NAME：建立 Kafka 叢集的子網路
  • KAFKA_CLUSTER：Kafka 叢集名稱

這項指令會以非同步方式執行，並傳回作業 ID：

Check operation [projects/PROJECT_ID/locations/REGION/operations/OPERATION_ID] for status.

如要追蹤建立作業的進度，請使用 gcloud managed-kafka operations describe 指令：

gcloud managed-kafka operations describe OPERATION_ID \
  --location=REGION

詳情請參閱「監控叢集建立作業」。

授予 IAM 角色

將 BigQuery 資料編輯者 Identity and Access Management (IAM) 角色授予代管 Kafka 服務帳戶。這個角色可讓連接器寫入 BigQuery 資料表。

控制台

  1. 前往 Google Cloud 控制台的「IAM」(身分與存取權管理) 頁面。

    前往「身分與存取權管理」頁面

  2. 選取「包含 Google 提供的角色授予項目」。

  3. 找到「Managed Kafka Service Account」(受管理 Kafka 服務帳戶) 列，然後按一下 「Edit principal」(編輯主體)。

  4. 點選「新增其他角色」，然後選取「BigQuery 資料編輯者」角色。

  5. 按一下 [儲存]。

如要進一步瞭解如何授予角色，請參閱「使用控制台授予 IAM 角色」。

gcloud

如要將 IAM 角色授予服務帳戶，請執行 gcloud projects add-iam-policy-binding 指令。

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --role=roles/bigquery.dataEditor

更改下列內容：

  • PROJECT_ID：專案 ID
  • PROJECT_NUMBER：您的專案編號

如要找出專案編號，請使用 gcloud projects describe 指令。

建立 BigQuery 接收器連接器

在這個步驟中，您會建立 BigQuery Sink 連接器。這個連接器會從一或多個主題讀取訊息，並將訊息寫入 BigQuery。

控制台

  1. 前往「Managed Service for Apache Kafka」>「Connect Clusters」(連線叢集) 頁面。

    前往「Connect Clusters」(連結叢集)

  2. 按一下 Connect 叢集名稱。

  3. 按一下 「建立連接器」。

  4. 在「Connector name」(連接器名稱) 中輸入字串。範例：bigquery-connector。

  5. 在「連接器外掛程式」清單中，選取「BigQuery Sink」。

  6. 在「主題」部分，選取先前建立的主題，然後按一下「確定」。

  7. 在「資料集」中，輸入 BigQuery 資料集的名稱， 格式如下： PROJECT_ID.DATASET_NAME。 範例：my-project.dataset1。

  8. 在「Configurations」(設定) 編輯方塊中，加入以下這行程式碼：

    bigQueryPartitionDecorator=false

  9. 點選「建立」。

gcloud

如要建立 BigQuery Sink 連接器，請執行 gcloud managed-kafka connectors create 指令。

gcloud managed-kafka connectors create CONNECTOR_NAME \
  --location=REGION \
  --connect-cluster=CONNECT_CLUSTER \
  --configs=connector.class=com.wepay.kafka.connect.bigquery.BigQuerySinkConnector,\
key.converter=org.apache.kafka.connect.storage.StringConverter,\
value.converter=org.apache.kafka.connect.json.JsonConverter,\
value.converter.schemas.enable=false,\
tasks.max=3,\
project=PROJECT_ID,\
defaultDataset=DATASET_NAME,\
topics=TABLE_NAME,\
bigQueryPartitionDecorator=false

更改下列內容：

  • CONNECTOR_NAME：連接器名稱，例如 bigquery-connector
  • CONNECT_CLUSTER：Connect 叢集名稱
  • REGION：您建立 Connect 叢集的區域
  • PROJECT_ID：專案 ID
  • DATASET_NAME：BigQuery 資料集的名稱。
  • TABLE_NAME：Kafka 主題的名稱，與 BigQuery 資料表相同。

將 bigQueryPartitionDecorator 設定參數設為 false，可防止連接器在資料表名稱中加入分區修飾符 (例如 "$"yyyyMMdd")。

查看結果

如要查看結果，請將訊息傳送至 Kafka 主題。請使用下列格式撰寫郵件內文：

{ "name": "STRING_VALUE", "id": INTEGER_VALUE }

您可以透過多種方式將訊息傳送至 Managed Service for Apache Kafka，包括：

如要在 BigQuery 中查看記錄，請對資料表執行下列查詢：

控制台

  1. 開啟「BigQuery」BigQuery頁面。

    前往 BigQuery 頁面

  2. 在查詢編輯器中執行下列查詢：

    SELECT * FROM `PROJECT_ID.DATASET_NAME.TABLE_NAME`
LIMIT 1000

    請替換下列變數：

    • PROJECT_ID：您 Google Cloud專案的名稱
    • DATASET_NAME：BigQuery 資料集名稱
    • TABLE_NAME：BigQuery 資料表的名稱

gcloud

使用 bq query 指令查詢資料表：

bq query --use_legacy_sql=false 'SELECT * FROM `PROJECT_ID.DATASET_NAME.TABLE_NAME`'

請替換下列變數：

  • PROJECT_ID：您 Google Cloud專案的名稱
  • DATASET_NAME：BigQuery 資料集名稱
  • TABLE_NAME：BigQuery 資料表的名稱

清除所用資源

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

控制台

  1. 刪除 Connect 叢集。

    1. 前往「Managed Service for Apache Kafka」>「Connect Clusters」(連線叢集) 頁面。

      前往「Connect Clusters」(連結叢集)

    2. 選取 Connect 叢集，然後按一下「刪除」。

  2. 刪除 Kafka 叢集。

    1. 前往「Managed Service for Apache Kafka」>「Clusters」(叢集) 頁面。

      前往「Clusters」(叢集)

    2. 選取 Kafka 叢集，然後按一下「Delete」(刪除)。

  3. 刪除 BigQuery 資料表和資料集。

    1. 前往「BigQuery」頁面

      前往 BigQuery 頁面

    2. 在「Explorer」窗格中展開專案，然後選取資料集。

    3. 展開「動作」選項，然後點按「刪除」。

    4. 在「Delete dataset」(刪除資料集) 對話方塊中，在欄位輸入 delete，然後按一下「Delete」(刪除)。

gcloud

  1. 如要刪除 Connect 叢集，請使用 gcloud managed-kafka connect-clusters delete 指令。

    gcloud managed-kafka connect-clusters delete CONNECT_CLUSTER \
  --location=REGION --async

  2. 如要刪除 Kafka 叢集，請使用 gcloud managed-kafka clusters delete 指令。

    gcloud managed-kafka clusters delete KAFKA_CLUSTER \
  --location=REGION --async

  3. 如要同時刪除 BigQuery 資料集和 BigQuery 資料表，請使用 bq rm 指令。

    bq rm --recursive --dataset PROJECT_ID:DATASET_NAME

後續步驟