如要將資料從 Google Cloud Managed Service for Apache Kafka 主題載入至 Cloud Storage bucket，可以使用 Dataflow 範本。您可以使用 Google Cloud 控制台、REST API 或 Google Cloud CLI。
本文說明如何使用 Google Cloud 控制台設定 Kafka to Cloud Storage Dataflow 範本。
Google Cloud 使用的產品
Kafka to 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 管道轉移的資料。
此外，這項解決方案也使用 Google Cloud Managed Service for Apache Kafka。
- Google Cloud Managed Service for Apache Kafka：這項 Google Cloud 服務 可協助您執行 Apache Kafka。提供管道的來源資料。 您需要現有的 Managed Service for Apache Kafka 叢集和主題， 其中包含要轉移至 Cloud Storage 的資料。如要進一步瞭解 Google Cloud Managed Service for Apache Kafka 定價，請參閱定價指南。
事前準備
啟動 Kafka to Cloud Storage Dataflow 範本前，請確認已完成下列步驟：
建立 Managed Service for Apache Kafka 叢集和主題。
如要建立叢集和主題，其中一種方法是按照 Managed Service for Apache Kafka 快速入門導覽課程操作。
如果主題包含 Avro 記錄，請參閱「指定訊息格式」，瞭解其他資源需求。
啟用下列 Google Cloud API：
Dataflow
Cloud Storage
gcloud services enable dataflow.googleapis.com storage-api.googleapis.com \建立 Cloud Storage bucket。
如要進一步瞭解如何建立 Cloud Storage bucket，請參閱「建立 bucket」。
將代管 Kafka 用戶端角色授予 Dataflow 工作站服務帳戶
如要將 Dataflow 工作連線至 Managed Service for Apache Kafka，請授予 Dataflow worker 服務帳戶特定權限。這個服務帳戶是 Dataflow 工作中所有工作站 VM 使用的身分，且這些 VM 發出的任何要求都會使用這個帳戶。
如要允許存取 Kafka 資源，您必須將 roles/managedkafka.client 角色授予 Dataflow 工作者服務帳戶。這個角色包含建立連線所需的 managedkafka.clusters.connect 權限。
如要進一步瞭解工作站服務帳戶，請參閱「Google Cloud 上的管道安全與權限」。
如要將代管 Kafka 用戶端角色授予 Dataflow 服務帳戶，請按照下列步驟操作：
控制台
- 前往 Google Cloud 控制台的「IAM」(身分與存取權管理) 頁面。
前往 IAM - 確認專案已設為 Managed Service for Apache Kafka 用戶端要存取的消費者專案。
- 按一下「授予存取權」。
- 在新頁面的「新增主體」中，輸入您使用的 Dataflow 工作人員服務帳戶電子郵件地址。
- 在「指派角色」中，選取「Managed Kafka client」(受管理 Kafka 用戶端) 角色。
- 按一下 [儲存]。
gcloud CLI
-
在 Google Cloud 控制台中啟用 Cloud Shell。
Google Cloud 主控台底部會開啟一個 Cloud Shell 工作階段，並顯示指令列提示。Cloud Shell 是已安裝 Google Cloud CLI 的殼層環境，並已針對您目前的專案設定好相關值。工作階段可能要幾秒鐘的時間才能初始化。
-
執行
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 範本。
-
前往 Google Cloud 控制台的「Cluster」(叢集) 頁面。
系統會列出您在專案中建立的叢集。
- 如要查看叢集詳細資料頁面，請按一下叢集名稱。
- 在叢集詳細資料頁面中，按一下「匯入資料」。
「使用範本『Kafka 至 Kafka』建立 Dataflow 工作」頁面隨即開啟。
- 在範本中，將「Dataflow 範本」更新為「Kafka to Cloud Storage」。
請根據下列各節的資訊，設定範本中的欄位。
輸入工作名稱
在「Job name」(工作名稱) 欄位中，輸入 Dataflow 工作的名稱。
專案中所有目前正在執行的工作名稱不得重複。
為管道選擇區域端點
在「Regional endpoint」(區域端點) 欄位中，將區域端點設為 Kafka 叢集所在位置，盡量減少跨區域資料移轉費用。
Dataflow 工作站可獨立於 Kafka 叢集區域執行。不過，如果您在 Kafka 叢集所在區域以外啟動工作站，則會產生區域間輸出費用。
如要查看叢集位置，請按照「列出 Managed Service for Apache Kafka 叢集」一文中的步驟操作。
設定來源
在「來源」中，保留「Managed Service for Apache Kafka」的預設值。
保留「Kafka cluster」(Kafka 叢集) 和「Kafka source authentication mode」(Kafka 來源驗證模式) 的預設值。
在「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"}]」格式。
Google Cloud Managed Service for Apache Kafka 不提供結構定義登錄。範本僅支援將驗證憑證傳遞至與 Confluent 線路格式相容的結構定義登錄。
Avro Confluent 線路格式
如果您選擇這個選項做為 Kafka 訊息格式，請設定下列額外設定：
結構定義來源：這個欄位會告知管道結構定義的位置。 您可以選擇下列其中一個選項：
結構定義儲存庫：結構定義會儲存在 Confluent 結構定義儲存庫中。這項功能有助於演進結構定義及管理多個版本。 請確認結構定義登錄可供 Managed Service for Apache Kafka 叢集網路存取，且與 Dataflow 工作人員位於相同區域。您可以在單一和多個結構定義情境中使用結構定義登錄。設定下列其他設定：
結構定義儲存庫連線網址：提供連線至結構定義儲存庫的網址。
驗證模式：如果登錄檔需要驗證，請選取「OAuth」或「TLS」。否則請選取「無」。
單一結構定義檔案：如果所有訊息都遵循檔案中定義的單一固定結構定義，請選擇這個選項。
- Cloud Storage 檔案 (Avro 結構定義檔案)：用於解碼主題中所有訊息的 Avro 結構定義檔案路徑。
Avro 二進位編碼
如果您選擇這個選項做為 Kafka 訊息格式，請設定下列額外設定：
- Cloud Storage 檔案 (Avro 結構定義檔案)：用於解碼主題中所有訊息的 Avro 結構定義檔案路徑。
JSON
如果您選擇這個選項做為 Kafka 訊息格式，則無需進行其他設定。
指定 Kafka 偏移
如要避免在個別 worker 或整個管道需要重新啟動時重新處理訊息，請選取「將偏移量提交至 Kafka」選項。這可確保管道從中斷處繼續處理，避免重複處理及潛在的資料不一致問題。
在「Enter Consumer Group ID」(輸入消費者群組 ID) 欄位中，輸入這個管道群組的專屬名稱。在大多數情況下，您會希望管道讀取每則訊息一次，且可重新啟動。
在「Default Kafka start offset」(預設 Kafka 開始偏移) 欄位中，Dataflow 管道提供兩種開始偏移選項。請選取下列其中一個選項：
最早：從 Kafka 主題的開頭處理訊息。
最新：從最新的可用偏移量開始處理訊息。
設定目的地
這些選項可控管資料管道將資料寫入 Cloud Storage 的方式。
在「Destination」(目的地) 部分，輸入 bucket 路徑，並加入輸出檔案的檔案名稱前置字串。檔案前置字串結尾必須為斜線。例如：
gs://test-bucket/test-prefix/在「Window duration」(時間範圍) 中，輸入將資料寫入 Cloud Storage 的時間範圍。根據資料處理需求，選擇適當的格式 (秒數為
Ns、分鐘數為Nm、時數為Nh)。在「要寫入的檔案輸出檔案名稱前置字元」中，您可以提供要加到每個輸出檔案的前置字元，方便整理及識別檔案。
將「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 to Cloud Storage 的 Dataflow 範本提供監控體驗，讓您在控制台中探索記錄、指標和錯誤。這套監控工具是 Dataflow 使用者介面的一部分。
您可以在「作業指標」分頁中建立自訂資訊主頁。對於「Kafka to Cloud Storage」Dataflow 範本，建議您設定「工作指標」資訊主頁，監控下列項目：
處理量：在任何時間點處理的資料量。 這有助於監控工作中的資料流程，並找出潛在的效能問題。
詳情請參閱「監控 Dataflow 輸送量」。
資料更新間隔：資料元素的時間戳記與管道處理事件的時間差 (以秒為單位)。這有助於找出效能和資料來源瓶頸，或頻繁重試。
詳情請參閱「監控 Dataflow 資料更新間隔」。
待處理作業：等待處理的位元組數量。 這項資訊會影響自動調度資源決策。
如要進一步瞭解如何監控 Dataflow，請參閱 Dataflow 監控說明文件。
疑難排解
如果 Dataflow 管道發生效能問題，Dataflow 提供了一整套疑難排解和診斷工具。
以下是兩種常見情境和相應的疑難排解指南：
如需 Dataflow 管道偵錯的一般總覽，請參閱「排解及偵錯 Dataflow 管道」。
已知限制
範本不支援將驗證憑證傳遞至您的結構定義登錄。
建立 Kafka to Cloud Storage Dataflow 工作時，請確認 Google Cloud project 設定為包含 Managed Service for Apache Kafka 叢集的專案。