將 Managed Service for Apache Kafka 資料寫入 BigQuery

如要將 Managed Service for Apache Kafka 主題中的 AVRO 或 JSON 訊息載入 BigQuery 資料表，可以使用 Dataflow 範本。本文將說明如何使用 Google Cloud 控制台，根據這個範本設定管道。您也可以使用 REST API 或 Google Cloud CLI 設定這項管道。

使用的Google Cloud 產品

Kafka 至 BigQuery Dataflow 範本會使用下列計費 Google Cloud 產品。使用 Pricing Calculator，根據您的預測用量來產生預估費用。

  • BigQuery：BigQuery Google Cloud是無伺服器的資料倉儲。在這項解決方案中，BigQuery 是 Kafka 資料的目的地。BigQuery Streaming Write API 用於將資料寫入 BigQuery。接著，您可以使用 BigQuery 透過 SQL 查詢分析這項資料、使用 BigQuery ML 建立機器學習模型，以及為商業智慧應用程式提供支援。
  • Dataflow：Dataflow 是全代管的資料處理服務。Kafka 至 BigQuery Dataflow 會使用 Dataflow 建立管道，從 Kafka 主題讀取資料、執行任何必要轉換，然後寫入 BigQuery。Dataflow 的自動調度和自我修復功能可確保管道穩定有效率地運作。
  • Cloud Storage：雖然 Cloud Storage 並未直接參與核心資料流程，但如果您使用 Avro 二進位編碼處理 Kafka 訊息，可以將 Avro 結構定義檔案儲存在 Cloud Storage。這樣 Dataflow 管道才能正確解讀資料結構。

此外，這項解決方案也使用 Managed Service for Apache Kafka。

  • Managed Service for Apache Kafka：這項 Google Cloud 服務 可協助您執行 Apache Kafka。Managed Service for Apache Kafka 可讓您專注於建構事件導向系統和串流資料管道，不必費心管理基礎架構。如要進一步瞭解 Managed Service for Apache Kafka 定價，請參閱定價指南。

事前準備

啟動 Kafka 至 BigQuery Dataflow 範本前，請務必完成下列步驟：

  1. 建立 Managed Service for Apache Kafka 叢集和主題。

    如要建立叢集和主題，其中一種方法是按照 Managed Service for Apache Kafka 快速入門指南操作。

    如果主題包含 Avro 記錄，如要瞭解其他資源需求，請參閱「設定 Kafka 訊息格式」

  2. 啟用下列 Google Cloud API：

    • Dataflow

    • BigQuery

    • Cloud Storage

    gcloud services enable dataflow.googleapis.com bigquery.googleapis.com \
storage.googleapis.com

  3. 建立 BigQuery 資料集和資料表。確認資料表的結構定義與 Kafka 輸入主題的結構定義相符。

    如果要在同一主題中使用多個結構定義，並寫入多個資料表，則不必在設定管道前建立資料表。

    如要進一步瞭解如何建立 BigQuery 資料集和資料表，請參閱「建立資料集」和「建立具有結構定義的空白資料表」。

將代管 Kafka 用戶端角色授予 Dataflow 工作者服務帳戶

如要將 Dataflow 工作連結至 Managed Service for Apache Kafka，請將特定權限授予 Dataflow worker 服務帳戶。這個服務帳戶是 Dataflow 工作中所有 worker VM 使用的身分，且這些 VM 發出的任何要求都會使用這個帳戶。

如要允許存取 Kafka 資源，請將 roles/managedkafka.client 角色授予 Dataflow worker 服務帳戶。這個角色包含建立連線所需的 managedkafka.clusters.connect 權限。

如要進一步瞭解 Worker 服務帳戶，請參閱「Google Cloud 上的管道安全性與權限」。

如要將代管 Kafka 用戶端角色授予 Dataflow 服務帳戶，請按照下列步驟操作：

控制台

  1. 前往 Google Cloud 控制台的「IAM」(身分與存取權管理) 頁面。
    前往 IAM 頁面
  2. 確認專案已設為 Managed Service for Apache Kafka 用戶端要存取的消費者專案。
  3. 按一下「授予存取權」。
  4. 在新頁面的「新增主體」中，輸入您使用的 Dataflow 工作人員服務帳戶電子郵件地址。
  5. 在「指派角色」部分，選取「代管 Kafka 用戶端」角色。
  6. 按一下 [儲存]。

gcloud CLI

  1. 在 Google Cloud 控制台中啟用 Cloud Shell。

    啟用 Cloud Shell

    Google Cloud 控制台底部會開啟 Cloud Shell 工作階段，並顯示指令列提示。Cloud Shell 是已安裝 Google Cloud CLI 的殼層環境，並已設定適用於您目前專案的值。工作階段可能要幾秒鐘的時間才能初始化。

  2. 執行 gcloud projects add-iam-policy-binding 指令：

    gcloud projects add-iam-policy-binding PROJECT_ID \
  --member serviceAccount:SERVICE_ACCOUNT_EMAIL \
  --role roles/managedkafka.client

    更改下列內容：

    • PROJECT_ID 是專案 ID。

    • SERVICE_ACCOUNT_EMAIL 是 Dataflow 工作站服務帳戶的電子郵件地址。

啟動 Kafka 到 BigQuery 的 Dataflow 範本

您可以從控制台的叢集詳細資料頁面，啟動 Kafka to BigQuery Dataflow 範本。

  1. 前往 Google Cloud 控制台的「Cluster」(叢集) 頁面。

    前往「Clusters」(叢集)

    系統會列出您在專案中建立的叢集。

  2. 如要查看叢集詳細資料頁面，請按一下叢集名稱。
  3. 在叢集詳細資料頁面中，針對「主題」，按一下任何主題的「BigQuery Export」圖示。

    系統隨即會開啟「Create a Dataflow job using template "Kafka to BigQuery"」(使用「Kafka 至 BigQuery」範本建立 Dataflow 工作) 頁面。

根據下列各節的資訊，設定範本中的欄位。

輸入工作名稱

在「Job name」(工作名稱) 欄位中，輸入 Dataflow 工作的名稱。

專案中所有目前正在執行的工作名稱不得重複。

為管道選擇區域端點

在「Regional endpoint」(區域端點) 欄位中，將區域端點設為 Kafka 叢集或 BigQuery 資料集的位置，盡量減少跨區域資料移轉費用。

Dataflow 工作人員可獨立於 Kafka 叢集區域執行。不過，如果您在 Kafka 叢集所在區域以外啟動工作人員，就會產生區域間輸出費用。

如要查看叢集位置，請按照「列出 Managed Service for Apache Kafka 叢集」一文中的步驟操作。

選擇 Dataflow 範本

在「Dataflow template」(Dataflow 範本) 欄位中，保留預設值「Kafka to BigQuery」(Kafka 到 BigQuery)。

設定來源

  1. 在「來源」中，保留「Managed Service for Apache Kafka」的預設值。

  2. 系統會根據您選擇的主題，預先選取 Kafka 叢集、Kafka 主題和 Kafka 來源驗證模式的預設值。保留這些值。

設定 Kafka 訊息格式

Dataflow 範本支援三種訊息格式：

  • Avro Confluent 線路格式：每則 Kafka 訊息都包含魔術位元組、結構定義 ID 和 Avro 二進位編碼記錄。根據預設，訊息金鑰不會載入 BigQuery。

    如果是 Avro (Confluent 線路格式)，您可以使用單一或多個結構定義：

    • 單一結構定義：所有訊息都遵循單一預先定義的 Avro 結構定義。

    • 多個結構定義：訊息可使用不同結構定義。這項功能僅支援 Avro (Confluent 線路格式)。

  • Avro (二進位編碼)：訊息只包含記錄的酬載，不含任何中繼資料。您必須提供上傳至 Cloud Storage 的 Avro 結構定義檔案 (.avsc)。所有訊息都必須遵守這個單一結構定義。

  • JSON：記錄不需要預先定義的結構定義。不符合 BigQuery 資料表結構定義的記錄會傳送至 dead-letter 佇列 (如有設定)，或記錄錯誤訊息。支援的格式為 {"field": "value"} 格式。系統不支援「[{"name": "field", "value": "value"}]」格式。

Avro Confluent 線路格式

如果選擇這個選項做為 Kafka 訊息格式，請設定下列其他設定：

結構定義來源：這個欄位會告知管道結構定義的位置。 您可以選擇下列其中一個選項：

  • 結構定義儲存庫：結構定義會儲存在 Confluent 結構定義儲存庫中。 這項功能有助於演進結構定義及管理多個版本。 請確認結構定義登錄可供 Managed Service for Apache Kafka 叢集網路存取，且與 Dataflow 工作人員位於相同區域。您可以在單一和多個結構定義情境中使用結構定義登錄。 設定下列其他設定：

    • 結構定義儲存庫連線網址：提供連線至結構定義儲存庫的網址。

    • 驗證模式：如果登錄檔需要驗證，請選取「OAuth」或「TLS」。否則請選取「無」。

  • 單一結構定義檔案：如果所有訊息都遵循檔案中定義的單一固定結構定義，請選擇這個選項。

    • Avro 結構定義檔案的 Cloud Storage 檔案：用於解碼主題中所有訊息的 Avro 結構定義檔案路徑。

Managed Service for Apache Kafka 不提供結構定義登錄。範本僅支援將驗證憑證傳遞至與 Confluent 線路格式相容的結構定義登錄。

Avro 二進位編碼

如果選擇這個選項做為 Kafka 訊息格式，請設定下列其他設定：

  • Avro 結構定義檔案的 Cloud Storage 檔案：用於解碼主題中所有訊息的 Avro 結構定義檔案路徑。

JSON

如果選擇這個選項做為 Kafka 訊息格式，就不需要其他設定。

設定將 Kafka 金鑰保留在 BigQuery 資料表中

如果啟用「將 Kafka 訊息鍵保留在 BigQuery 資料表中」選項，您也必須在目標資料表中新增名為 _key 的 BYTES 類型欄位。雖然鍵可能代表結構化資料，但範本會將其視為位元組陣列。

指定位移選項

  1. 如要避免在個別工作人員或整個管道需要重新啟動時重新處理訊息，請選取「將偏移量提交至 Kafka」選項。 確保管道從中斷處繼續處理，避免重複處理和潛在的資料不一致問題。

  2. 在「Enter Consumer Group ID」(輸入消費者群組 ID) 欄位中，輸入這個管道群組的專屬名稱。在大多數情況下，您會希望管道讀取每則訊息一次，且可重新啟動。

  3. 在「Default Kafka start offset」欄位中，Dataflow 管道提供兩種起始偏移選項。請選取下列其中一個選項：

    • 最早：從 Kafka 主題的開頭處理訊息。

    • 最新：從最新的可用偏移量開始處理訊息。

設定目的地

這些選項可控管資料管道將資料寫入 BigQuery 的方式。

「資料表名稱策略」欄位會決定資料在 BigQuery 資料表中的整理方式。您有以下兩種選擇：

  • 單一資料表名稱 (預設)：Kafka 主題的所有資料都會匯入單一 BigQuery 資料表。這是最簡單的選項，如果 Kafka 資料具有相同結構定義，建議使用這個選項。

    在「BigQuery output table」(BigQuery 輸出資料表) 中，指定要儲存 Kafka 主題所有資料的資料表名稱。

  • 動態資料表名稱：根據 Kafka 訊息的結構定義建立多個 BigQuery 資料表。如果 Kafka 訊息具有不同結構定義，建議使用這個選項。只有在管道使用 Avro Confluent Wire 或結構定義登錄檔處理訊息結構時，才能使用這個選項。

    Kafka 訊息中的每個不重複結構定義，都會在 BigQuery 中建立專屬資料表。 系統會根據結構定義名稱自動產生表格名稱。

    如果選取「動態資料表名稱」，請設定下列額外欄位：

    • BigQuery 輸出專案：BigQuery 資料集所在的 Google Cloud 專案。

    • BigQuery 輸出資料集：專案中的資料集，用於建立資料表。

    • BigQuery 資料表命名前置字串 (選用)：為自動產生的資料表名稱加上前置字串，方便整理。

設定 BigQuery Storage Write API

在「Number of streams for the BigQuery Storage Write API」(BigQuery Storage Write API 的串流數量) 中，從 0 (預設值) 開始。這個選項可讓管道自動判斷最佳串流數量。

這項設定可控管用於將資料寫入 BigQuery 的平行串流數量。數字越大，寫入速度就越快，尤其是在處理大量資料時。最佳數量取決於資料和 BigQuery 設定。

設定觸發頻率

針對「BigQuery Write API 的觸發頻率 (以秒為單位)」選項，請從 5 秒 (預設值) 開始。在大多數情況下，這都是不錯的平衡點。

這項設定會決定資料寫入 BigQuery 的頻率。

數字越小，寫入頻率越高。這可以縮短延遲時間，但可能會增加 BigQuery 費用。數字越大，寫入頻率越低。這樣做可能較符合成本效益，但可能會增加延遲時間。

設定無效信件佇列

有時訊息會因毀損、資料類型不相容，或與目的地 BigQuery 資料表的結構定義不符而無法處理。

如要處理這些情況，請在範本中啟用無效信件佇列，並提供資料表名稱。範本會使用標準化結構定義建立資料表。錯誤訊息會寫入另一個 BigQuery 資料表。

設定加密機制

根據預設，所有靜態和傳輸中的資料都會經過加密 Google-owned and Google-managed encryption key。如果您有客戶自行管理的加密金鑰 (CMEK)，可以選取自己的金鑰。如要進一步瞭解如何設定 CMEK，請參閱「設定訊息加密」。

設定網路

您必須在 Dataflow 範本中指定叢集的網路和子網路。範本的「選用參數」部分可讓您定義 Dataflow 工作站的網路。

根據預設，Kafka 到 BigQuery 的 Dataflow 範本會在專案的預設網路中佈建 Dataflow 工作人員。如要讓 Managed Service for Apache Kafka 叢集透過 Dataflow 將資料傳送至 BigQuery，請確保 Dataflow 工作者可以存取叢集的網路。

如果 Kafka 叢集未連線至專案預設網路中的子網路，建議您為 Kafka 叢集使用專案的預設網路。

如要進一步瞭解如何使用 Dataflow 管道設定網路，請參閱下列文章：

如果設定 Dataflow 網路時遇到問題，請參閱 Dataflow 網路疑難排解指南。

設定選用的 Dataflow 參數

只有在瞭解設定對 Dataflow worker 的影響時，才設定選用參數。設定錯誤可能會影響效能或費用。如需各個選項的詳細說明，請參閱「選用參數」。

監控

Kafka to BigQuery 的 Dataflow 範本提供監控體驗，可讓您在控制台中探索記錄、指標和錯誤。這套監控工具是 Dataflow 使用者介面的一部分。

您可以在「作業指標」分頁中建立自訂資訊主頁，對於 Kafka 到 BigQuery 範本，建議您設定「工作指標」資訊主頁，監控下列項目：

  • 處理量：在任何時間點處理的資料量。 這有助於監控工作中的資料流程，並找出潛在的效能問題。

    詳情請參閱「監控 Dataflow 處理量」。

  • 資料更新間隔：資料元素的時間戳記與事件在管道中處理時間的差距 (以秒為單位)。這有助於找出效能和資料來源瓶頸，或頻繁重試。

    詳情請參閱「監控 Dataflow 資料更新間隔」。

  • 待處理作業：待處理的位元組數量。 這項資訊會影響自動調度資源決策。

如要進一步瞭解如何監控 Dataflow，請參閱 Dataflow 監控說明文件。

疑難排解

如果 Dataflow 管道發生效能問題，Dataflow 提供了一整套疑難排解和診斷工具。

以下是兩種常見情況和相應的疑難排解指南：

如需 Dataflow 管道偵錯的一般總覽，請參閱「疑難排解及偵錯 Dataflow 管道」。

已知限制

  • 範本不支援將驗證憑證傳遞至您的結構定義登錄。

  • 建立 Kafka 到 BigQuery 的 Dataflow 工作時，請確認 Google Cloud 專案已設為包含 Managed Service for Apache Kafka 叢集的專案。

後續步驟