將 Managed Service for Apache Kafka 資料寫入 Cloud Storage

如要將資料從 Managed Service for Apache Kafka 主題載入 Cloud Storage bucket，可以使用 Dataflow 範本。您可以使用 Google Cloud 控制台、REST API 或 Google Cloud CLI。

本文說明如何使用 Google Cloud 控制台，設定 Kafka 到 Cloud Storage Dataflow 範本。

使用的Google Cloud 產品

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

  • Dataflow：Dataflow 是全代管的資料處理服務。Kafka 至 Cloud Storage Dataflow 範本會使用 Dataflow 建立管道，從 Kafka 主題讀取資料、執行任何必要轉換，然後將資料寫入 Cloud Storage。Dataflow 的自動調度和自我修復功能可確保管道穩定有效率地運作。
  • Cloud Storage：做為 Kafka 資料的目的地。 您需要 Cloud Storage bucket，才能儲存 Dataflow 管道轉移的資料。

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

  • Managed Service for Apache Kafka：這項 Google Cloud 服務 可協助您執行 Apache Kafka。提供管道的來源資料。 您需要現有的 Managed Service for Apache Kafka 叢集和主題， 其中包含要轉移至 Cloud Storage 的資料。如要進一步瞭解 Managed Service for Apache Kafka 定價，請參閱定價指南。

事前準備

啟動 Kafka to Cloud Storage Dataflow 範本前，請先完成下列步驟：

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

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

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

  2. 啟用下列 Google Cloud API：

    • Dataflow

    • Cloud Storage

    gcloud services enable dataflow.googleapis.com storage-api.googleapis.com \

  3. 建立 Cloud Storage bucket。

    如要進一步瞭解如何建立 Cloud Storage bucket，請參閱「建立 bucket」。

將代管 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 到 Cloud Storage 的 Dataflow 範本

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

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

    前往「Clusters」(叢集)

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

  2. 如要查看叢集詳細資料頁面，請按一下叢集名稱。
  3. 在叢集詳細資料頁面中，按一下「匯入資料」。

    「使用範本『Kafka 至 Kafka』建立 Dataflow 工作」頁面隨即開啟。

  4. 在範本中，將「Dataflow 範本」更新為「Kafka to Cloud Storage」。

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

輸入工作名稱

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

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

為管道選擇區域端點

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

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

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

設定來源

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

  2. 保留「Kafka cluster」(Kafka 叢集) 和「Kafka source authentication mode」(Kafka 來源驗證模式) 的預設值。

  3. 在「Kafka topic」(Kafka 主題) 中，從可用主題清單中選取主題。

設定 Kafka 訊息格式

Dataflow 範本支援下列訊息格式：

  • Avro Confluent 線路格式：每則 Kafka 訊息都包含魔術位元組、結構定義 ID 和 Avro 二進位編碼記錄。

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

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

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

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

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

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

Avro Confluent 線路格式

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

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

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

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

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

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

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

Avro 二進位編碼

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

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

JSON

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

指定 Kafka 偏移

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

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

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

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

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

設定目的地

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

  1. 在「Destination」(目的地) 部分，輸入 bucket 路徑，並加入輸出檔案的檔案名稱前置字元。檔案前置字串結尾必須為斜線。例如： gs://test-bucket/test-prefix/

  2. 在「Window duration」(時間範圍) 中，輸入將資料寫入 Cloud Storage 的時間範圍。根據資料處理需求，選擇適當的格式 (秒數為 Ns、分鐘數為 Nm、時數為 Nh)。

  3. 在「Output filename prefix of the files to write」(要寫入的檔案輸出檔案名稱前置字元) 部分，您可以提供要新增至每個輸出檔案的前置字元，方便整理及識別檔案。

  4. 將「Maximum output shards」(輸出分片數量上限) 設為零。 您可以指定寫入檔案時要產生的分片數量。增加數量可提高處理量，但也會導致重組成本增加，進而提高費用。如果將數字設為零，服務會選取最佳數字。

設定無效信件佇列

有時郵件會因損毀、資料型別不相容或結構定義不符而無法處理。

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

設定加密機制

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

設定網路

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

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

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

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

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

設定選用的 Dataflow 參數

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

監控

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

您可以在「作業指標」分頁中建立自訂資訊主頁，如果是 Kafka 至 Cloud Storage Dataflow 範本，建議您設定「工作指標」資訊主頁，監控下列項目：

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

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

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

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

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

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

疑難排解

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

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

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

已知限制

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

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

後續步驟