將 Pub/Sub 連線至 Apache Kafka

本文說明如何使用 Pub/Sub Group Kafka Connector 整合 Apache Kafka 和 Pub/Sub。

關於 Pub/Sub Group Kafka Connector

Apache Kafka 是開放原始碼平台，可串流事件。這項技術通常用於分散式架構，可讓鬆耦合的元件之間進行通訊。Pub/Sub 是代管服務，用於以非同步方式傳送及接收訊息。與 Kafka 一樣，您可以使用 Pub/Sub 在雲端架構中的元件之間進行通訊。

您可以使用 Pub/Sub Group Kafka Connector 整合這兩個系統。Connector JAR 包含下列連接器：

  • 接收器連接器會從一或多個 Kafka 主題讀取記錄，並發布至 Pub/Sub。
  • 來源連接器會從 Pub/Sub 主題讀取訊息，並發布至 Kafka。

以下是 Pub/Sub Group Kafka Connector 的一些使用情境：

  • 您要將以 Kafka 為基礎的架構遷移至 Cloud de Confiance by S3NS。
  • 您有一個前端系統，會將事件儲存在Cloud de Confiance by S3NS以外的 Kafka 中，但您也使用 Cloud de Confiance by S3NS 執行部分後端服務，這些服務需要接收 Kafka 事件。
  • 從地端部署 Kafka 解決方案收集記錄，並傳送至Cloud de Confiance by S3NS 進行資料分析。
  • 您有一個使用 Cloud de Confiance by S3NS的前端系統，但您也使用 Kafka 在地端部署環境儲存資料。

連接器需要 Kafka Connect，這是用於在 Kafka 和其他系統之間串流資料的架構。如要使用連接器，您必須在 Kafka 叢集旁執行 Kafka Connect。

本文假設您已熟悉 Kafka 和 Pub/Sub。建議您先完成其中一個 Pub/Sub 快速入門導覽課程，再閱讀本文。

Pub/Sub 連接器不支援 IAM 與 Kafka Connect ACL 之間 Cloud de Confiance 的任何整合。

開始使用連接器

本節將逐步引導您完成下列工作：

  1. 設定 Pub/Sub Group Kafka Connector。
  2. 將 Kafka 的事件傳送至 Pub/Sub。
  3. 將訊息從 Pub/Sub 傳送至 Kafka。

必要條件

安裝 Kafka

按照 Apache Kafka 快速入門導覽課程，在本機電腦上安裝單一節點 Kafka。請完成快速入門導覽課程中的下列步驟：

  1. 下載並解壓縮最新版 Kafka。
  2. 啟動 Kafka 環境。
  3. 建立 Kafka 主題。

驗證

Pub/Sub Group Kafka Connector 必須向 Pub/Sub 驗證，才能傳送及接收 Pub/Sub 訊息。如要設定驗證，請按照下列步驟操作：

  1. 安裝 Google Cloud CLI。

  2. 設定 gcloud CLI，使用您的聯合身分。

    詳情請參閱「 使用聯合身分登入 gcloud CLI」。

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

    gcloud init

  4. 建立或選取 Cloud de Confiance 專案。

    選取或建立專案所需的角色

    • 選取專案：選取專案時，不需要具備特定 IAM 角色，只要您在專案中獲派角色，即可選取該專案。
    • 建立專案：如要建立專案，您需要專案建立者角色 (roles/resourcemanager.projectCreator)，其中包含 resourcemanager.projects.create 權限。瞭解如何授予角色。

    • 建立 Cloud de Confiance 專案：

      gcloud projects create PROJECT_ID

      將 PROJECT_ID 替換為要建立的專案名稱。 Cloud de Confiance

    • 選取您建立的 Cloud de Confiance 專案：

      gcloud config set project PROJECT_ID

      將 PROJECT_ID 替換為 Cloud de Confiance 專案名稱。

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

    gcloud auth application-default login

    如果系統傳回驗證錯誤，請確認您已 使用聯合身分登入 gcloud CLI。

  6. 將角色授予使用者帳戶。針對下列各個 IAM 角色執行下列指令一次：roles/pubsub.admin

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

    更改下列內容：

下載連接器 JAR

將連接器 JAR 檔案下載至本機電腦。詳情請參閱 GitHub 讀我檔案中的「取得連接器」一節。

複製連接器設定檔

  1. 複製或下載連接器的 GitHub 存放區。

    git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
cd java-pubsub-group-kafka-connector

  2. 將 config 目錄的內容複製到 Kafka 安裝目錄的 config 子目錄。

    cp config/* [path to Kafka installation]/config/

這些檔案包含連接器的設定。

更新 Kafka Connect 設定

  1. 前往包含您下載的 Kafka Connect 二進位檔的目錄。
  2. 在 Kafka Connect 二進位目錄中，使用文字編輯器開啟名為 config/connect-standalone.properties 的檔案。
  3. 如果 plugin.path property 已註解排除，請取消註解。

  4. 更新 plugin.path property，加入連接器 JAR 的路徑。

    範例：

    plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar

  5. 將 offset.storage.file.filename 屬性設為本機檔案名稱。在獨立模式下，Kafka 會使用這個檔案儲存偏移資料。

    範例：

    offset.storage.file.filename=/tmp/connect.offsets

將 Kafka 的事件轉送至 Pub/Sub

本節說明如何啟動接收器連接器、將事件發布至 Kafka，然後從 Pub/Sub 讀取轉送的訊息。

  1. 使用 Google Cloud CLI 建立 Pub/Sub 主題和訂閱項目。

    gcloud pubsub topics create PUBSUB_TOPIC
gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

    更改下列內容：

    • PUBSUB_TOPIC：Pub/Sub 主題的名稱，用於接收來自 Kafka 的訊息。
    • PUBSUB_SUBSCRIPTION：主題的 Pub/Sub 訂閱項目名稱。

  2. 在文字編輯器中開啟 /config/cps-sink-connector.properties 檔案。為下列屬性新增值，這些屬性在註解中標示為 "TODO"：

    topics=KAFKA_TOPICS
cps.project=PROJECT_ID
cps.topic=PUBSUB_TOPIC

    更改下列內容：

    • KAFKA_TOPICS：以半形逗號分隔的 Kafka 主題清單，用於讀取資料。
    • PROJECT_ID：包含 Pub/Sub 主題的 Cloud de Confiance by S3NS 專案。
    • PUBSUB_TOPIC：接收 Kafka 訊息的 Pub/Sub 主題。

  3. 在 Kafka 目錄中執行下列指令：

    bin/connect-standalone.sh \
  config/connect-standalone.properties \
  config/cps-sink-connector.properties

  4. 按照 Apache Kafka 快速入門中的步驟，將一些事件寫入 Kafka 主題。

  5. 使用 gcloud CLI 從 Pub/Sub 讀取事件。

    gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack

將訊息從 Pub/Sub 轉送至 Kafka

本節說明如何啟動來源連接器、將訊息發布至 Pub/Sub，以及從 Kafka 讀取轉送的訊息。

  1. 使用 gcloud CLI 建立含有訂閱項目的 Pub/Sub 主題。

    gcloud pubsub topics create PUBSUB_TOPIC
gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

    更改下列內容：

    • PUBSUB_TOPIC：Pub/Sub 主題的名稱。
    • PUBSUB_SUBSCRIPTION：Pub/Sub 訂閱項目名稱。

  2. 在文字編輯器中開啟名為 /config/cps-source-connector.properties 的檔案。為下列屬性新增值，這些屬性在註解中標示為 "TODO"：

    kafka.topic=KAFKA_TOPIC
cps.project=PROJECT_ID
cps.subscription=PUBSUB_SUBSCRIPTION

    更改下列內容：

    • KAFKA_TOPIC：用於接收 Pub/Sub 訊息的 Kafka 主題。
    • PROJECT_ID：包含 Pub/Sub 主題的 Cloud de Confiance by S3NS 專案。
    • PUBSUB_TOPIC：Pub/Sub 主題。

  3. 在 Kafka 目錄中執行下列指令：

    bin/connect-standalone.sh \
  config/connect-standalone.properties \
  config/cps-source-connector.properties

  4. 使用 gcloud CLI 將訊息發布至 Pub/Sub。

    gcloud pubsub topics publish PUBSUB_TOPIC --message="message 1"

  5. 從 Kafka 讀取訊息。按照 Apache Kafka 快速入門導覽課程中的步驟，從 Kafka 主題讀取訊息。

訊息轉換

Kafka 記錄包含鍵和值，兩者都是可變長度的位元組陣列。Kafka 記錄也可以選擇性地包含標頭 (即鍵/值組合)。Pub/Sub 訊息主要包含兩個部分：郵件內文，以及零或多個鍵/值屬性。

Kafka Connect 會使用轉換器，將鍵和值序列化為 Kafka，以及從 Kafka 序列化。 如要控管序列化作業，請在連接器設定檔中設定下列屬性：

  • key.converter：用於序列化記錄鍵的轉換器。
  • value.converter：用於序列化記錄值的轉換器。

Pub/Sub 訊息的主體是 ByteString 物件，因此最有效率的轉換方式是直接複製酬載。因此，建議您盡可能使用會產生原始資料類型 (整數、浮點數、字串或位元組結構定義) 的轉換器，避免對相同的郵件內文進行反序列化和重新序列化。

從 Kafka 轉換至 Pub/Sub

接收器連接器會將 Kafka 記錄轉換為 Pub/Sub 訊息，轉換方式如下：

  • Kafka 記錄鍵會以名為 "key" 的屬性名稱形式儲存在 Pub/Sub 訊息中。
  • 根據預設，連接器會捨棄 Kafka 記錄中的所有標頭。不過，如果將 headers.publish 設定選項設為 true，連接器會將標頭寫入為 Pub/Sub 屬性。如果任何標頭超出 Pub/Sub 的訊息屬性限制，連接器就會略過這些標頭。
  • 如果是整數、浮點數、字串和位元組結構定義，連接器會將 Kafka 記錄值的位元組直接傳遞至 Pub/Sub 訊息內文。
  • 如果是 struct 結構定義，連接器會將每個欄位寫入為 Pub/Sub 訊息的屬性。舉例來說，如果欄位是 { "id"=123 }，產生的 Pub/Sub 訊息就會有 "id"="123" 屬性。欄位值一律會轉換為字串。系統不支援將對應和結構體類型做為結構體內的欄位類型。
  • 如果是對應結構定義，連接器會將每個鍵/值組合寫入為 Pub/Sub 訊息的屬性。舉例來說，如果對應項目是 {"alice"=1,"bob"=2}，產生的 Pub/Sub 訊息就會有 "alice"="1" 和 "bob"="2" 這兩項屬性。鍵和值會轉換為字串。

結構體和對應項目的結構定義有一些額外行為：

  • 您可以選擇設定 messageBodyName 設定屬性，指定特定結構體欄位或對應鍵做為訊息主體。欄位或鍵的值會儲存為郵件內文中的 ByteString。如果未設定 messageBodyName，則結構體和對應項架構的訊息內文會是空白。

  • 如果是陣列值，連接器只支援原始陣列類型。陣列中的值序列會串連成單一 ByteString 物件。

從 Pub/Sub 轉換為 Kafka

來源連接器會將 Pub/Sub 訊息轉換為 Kafka 記錄，轉換方式如下：

  • Kafka 記錄鍵：預設鍵為 null。您也可以設定 kafka.key.attribute 設定選項，指定要用做鍵的 Pub/Sub 訊息屬性。在這種情況下，連結器會尋找具有該名稱的屬性，並將記錄鍵設為屬性值。如果沒有指定屬性，記錄鍵會設為 null。

  • Kafka 記錄值。連接器會依下列方式寫入記錄值：

    • 如果 Pub/Sub 訊息沒有自訂屬性，連接器會使用 value.converter 指定的轉換器，將 Pub/Sub 訊息內文直接寫入 Kafka 記錄值，做為 byte[] 類型。

    • 如果 Pub/Sub 訊息具有自訂屬性且 kafka.record.headers 為 false，連接器會將 struct 寫入記錄值。這個結構體包含每個屬性的一個欄位，以及一個名為 "message" 的欄位，其值為 Pub/Sub 訊息主體 (以位元組形式儲存)：

      {
  "message": "<Pub/Sub message body>",
  "<attribute-1>": "<value-1>",
  "<attribute-2>": "<value-2>",
  ....
}

      在這種情況下，您必須使用與 struct 結構定義相容的 value.converter，例如 org.apache.kafka.connect.json.JsonConverter。

    • 如果 Pub/Sub 訊息含有自訂屬性且 kafka.record.headers 為 true，連接器會將屬性寫入 Kafka 記錄標頭。它會使用 value.converter 指定的轉換器，將 Pub/Sub 訊息主體直接寫入 Kafka 記錄值，做為 byte[] 型別。

  • Kafka 記錄標頭。根據預設，標頭為空白，除非您將 kafka.record.headers 設為 true。

設定選項

除了 Kafka Connect API 提供的設定外，Pub/Sub Group Kafka Connector 也支援接收器和來源設定，如「Pub/Sub 連接器設定」一文所述。

取得支援

如需協助，請建立支援單。 如有一般問題和討論，請在 GitHub 存放區中建立問題。

後續步驟