Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

建立 Pub/Sub 來源連接器

Pub/Sub 來源連接器會將訊息從 Pub/Sub 串流至 Kafka，讓您將 Pub/Sub 與以 Kafka 為基礎的應用程式和資料管道整合。

Pub/Sub 來源連接器的用途包括：

即時擷取資料。將雲端服務或其他應用程式的資料發布至 Pub/Sub，然後將資料複製到 Kafka 進行串流處理。
事件導向架構。從發布至 Pub/Sub 的訊息觸發 Kafka 處理作業。

這個連接器會從 Pub/Sub 訂閱項目讀取訊息，將每則訊息轉換為 Kafka 記錄，然後將記錄寫入 Kafka 主題。根據預設，連接器會建立下列 Kafka 記錄：

Kafka 記錄鍵為 null。
Kafka 記錄值是 Pub/Sub 訊息資料 (以位元組表示)。
Kafka 記錄標頭為空白。

不過，您可以設定這項行為。詳情請參閱「設定連接器」。

事前準備

建立 Pub/Sub 來源連接器前，請確認您具備下列項目：

具有訂閱項目的 Pub/Sub 主題。
Managed Service for Apache Kafka 叢集。
Kafka 叢集中的 Kafka 主題。
Connect 叢集。建立 Connect 叢集時，請將 Managed Service for Apache Kafka 叢集設為主要 Kafka 叢集。

所需角色和權限

如要取得建立連線器所需的權限，請要求管理員授予您專案的代管 Kafka 連線器編輯者 (roles/managedkafka.connectorEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這個預先定義的角色具備建立連接器所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

如要建立連接器，必須具備下列權限：

建立連接器： managedkafka.connectors.create

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

授予從 Pub/Sub 讀取的權限

代管 Kafka 服務帳戶必須具備從 Pub/Sub 訂閱項目讀取訊息的權限。在包含 Pub/Sub 訂閱項目的專案中，將下列 IAM 角色授予服務帳戶：

Pub/Sub 訂閱者 (roles/pubsub.subscriber)
Pub/Sub 檢視者 (roles/pubsub.viewer)

代管 Kafka 服務帳戶的格式如下： service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com，其中 PROJECT_NUMBER 是 Connect 叢集的專案編號。

如果 Connect 叢集與 Managed Service for Apache Kafka 叢集位於不同專案，請參閱「在不同專案中建立 Connect 叢集」。

建立 Pub/Sub 來源連接器

控制台

前往 Google Cloud 控制台的「Connect Clusters」(連結叢集) 頁面。

前往「Connect Clusters」頁面
按一下要建立連接器的 Connect 叢集。
按一下「Create connector」(建立連接器)。
輸入連接器名稱字串。

如要查看連接器命名準則，請參閱 Managed Service for Apache Kafka 資源命名指南。
在「Connector plugin」部分，選取「Pub/Sub Source」。
在「Cloud Pub/Sub subscription」(Cloud Pub/Sub 訂閱項目) 清單中，選取 Pub/Sub 訂閱項目。連接器會從這個訂閱項目提取訊息。訂閱項目會以完整資源名稱顯示：projects/{project}/subscriptions/{subscription}。
在「Kafka 主題」清單中，選取要寫入訊息的 Kafka 主題。
選用：在「Configurations」方塊中，新增設定屬性或編輯預設屬性。詳情請參閱「設定連接器」。
選取「任務重新啟動政策」。詳情請參閱「工作重新啟動政策」。
點選「建立」。

gcloud

執行 gcloud managed-kafka connectors create 指令：
```
gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE
```
更改下列內容：
- CONNECTOR_ID：連接器的 ID 或名稱。如要查看連接器命名準則，請參閱 Managed Service for Apache Kafka 資源命名指南。連接器名稱無法變更。
- LOCATION：Connect 叢集的位置。
- CONNECT_CLUSTER_ID：建立連接器的 Connect 叢集 ID。
- CONFIG_FILE：YAML 或 JSON 設定檔的路徑。

以下是設定檔範例：

connector.class: "com.google.pubsub.kafka.source.CloudPubSubSourceConnector"
cps.project: "PROJECT_ID"
cps.subscription: "PUBSUB_SUBSCRIPTION_ID"
kafka.topic: "KAFKA_TOPIC_ID"
value.converter: "org.apache.kafka.connect.converters.ByteArrayConverter"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
tasks.max: "3"

更改下列內容：

PROJECT_ID：Pub/Sub 訂閱項目所在的專案 ID。 Google Cloud
PUBSUB_SUBSCRIPTION_ID：要從中提取資料的 Pub/Sub 訂閱項目 ID。
KAFKA_TOPIC_ID：寫入資料的 Kafka 主題 ID。

您必須提供 cps.project、cps.subscription 和 kafka.topic 設定屬性。如需其他設定選項，請參閱「設定連接器」。

Terraform

您可以使用 Terraform 資源建立連接器。

resource "google_managed_kafka_connector" "example-pubsub-source-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-pubsub-source-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "connector.class"  = "com.google.pubsub.kafka.source.CloudPubSubSourceConnector"
    "name"             = "my-pubsub-source-connector"
    "tasks.max"        = "3"
    "kafka.topic"      = "GMK_TOPIC_ID"
    "cps.subscription" = "CPS_SUBSCRIPTION_ID"
    "cps.project"      = data.google_project.default.project_id
    "value.converter"  = "org.apache.kafka.connect.converters.ByteArrayConverter"
    "key.converter"    = "org.apache.kafka.connect.storage.StringConverter"
  }

  provider = google-beta
}

如要瞭解如何套用或移除 Terraform 設定，請參閱「基本 Terraform 指令」。

Go

在試用這個範例之前，請先按照「安裝用戶端程式庫」中的 Go 設定說明操作。詳情請參閱 Managed Service for Apache Kafka Go API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證(ADC)。詳情請參閱「為本機開發環境設定 ADC」。

import (
	"context"
	"fmt"
	"io"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"
)

// createPubSubSourceConnector creates a Pub/Sub Source connector.
func createPubSubSourceConnector(w io.Writer, projectID, region, connectClusterID, connectorID, kafkaTopic, cpsSubscription, cpsProject, tasksMax, valueConverter, keyConverter string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "CPS_SOURCE_CONNECTOR_ID"
	// kafkaTopic := "GMK_TOPIC_ID"
	// cpsSubscription := "CPS_SUBSCRIPTION_ID"
	// cpsProject := "GCP_PROJECT_ID"
	// tasksMax := "3"
	// valueConverter := "org.apache.kafka.connect.converters.ByteArrayConverter"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	// Pub/Sub Source sample connector configuration
	config := map[string]string{
		"connector.class":  "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
		"name":             connectorID,
		"tasks.max":        tasksMax,
		"kafka.topic":      kafkaTopic,
		"cps.subscription": cpsSubscription,
		"cps.project":      cpsProject,
		"value.converter":  valueConverter,
		"key.converter":    keyConverter,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created Pub/Sub source connector: %s\n", resp.Name)
	return nil
}

Java

在試用這個範例之前，請先按照「安裝用戶端程式庫」中的 Java 設定說明操作。詳情請參閱 Managed Service for Apache Kafka Java API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定 ADC」。


import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreatePubSubSourceConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-pubsub-source-connector";
    String pubsubProjectId = "my-pubsub-project-id";
    String subscriptionName = "my-subscription";
    String kafkaTopicName = "pubsub-topic";
    String connectorClass = "com.google.pubsub.kafka.source.CloudPubSubSourceConnector";
    String maxTasks = "3";
    String valueConverter = "org.apache.kafka.connect.converters.ByteArrayConverter";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createPubSubSourceConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        pubsubProjectId,
        subscriptionName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        valueConverter,
        keyConverter);
  }

  public static void createPubSubSourceConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String pubsubProjectId,
      String subscriptionName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String valueConverter,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("name", connectorId);
    configMap.put("tasks.max", maxTasks);
    configMap.put("kafka.topic", kafkaTopicName);
    configMap.put("cps.subscription", subscriptionName);
    configMap.put("cps.project", pubsubProjectId);
    configMap.put("value.converter", valueConverter);
    configMap.put("key.converter", keyConverter);

    Connector connector = Connector.newBuilder()
        .setName(
            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
        .putAllConfigs(configMap)
        .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
          .setConnectorId(connectorId)
          .setConnector(connector)
          .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created Pub/Sub Source connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

在試用這個範例之前，請先按照「安裝用戶端程式庫」中的 Python 設定說明操作。詳情請參閱 Managed Service for Apache Kafka Python API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定 ADC」。

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "connector.class": "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
    "name": connector_id,
    "tasks.max": tasks_max,
    "kafka.topic": kafka_topic,
    "cps.subscription": cps_subscription,
    "cps.project": cps_project,
    "value.converter": value_converter,
    "key.converter": key_converter,
}

connector = Connector()
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

建立連接器後，您可以編輯、刪除、暫停、停止或重新啟動連接器。

設定連接器

本節說明可在連接器上設定的部分設定屬性。

如需這個連接器的專屬屬性完整清單，請參閱「Pub/Sub 來源連接器設定」。

拉動模式

提取模式會指定連接器擷取 Pub/Sub 訊息的方式。支援的模式如下：

提取模式 (預設)。系統會分批提取訊息。如要啟用這項模式，請設定 cps.streamingPull.enabled=false.。如要設定批次大小，請設定 cps.maxBatchSize 屬性。

如要進一步瞭解提取模式，請參閱「Pull API」。
串流提取模式。從 Pub/Sub 擷取訊息時，可達到最高處理量和最低延遲時間。如要啟用這項模式，請設定 cps.streamingPull.enabled=true。

如要進一步瞭解串流提取模式，請參閱「StreamingPull API」。

如果啟用串流提取功能，您可以設定下列設定屬性來調整效能：
- cps.streamingPull.flowControlBytes：每個工作未處理的訊息位元組數量上限。
- cps.streamingPull.flowControlMessages：每個工作未處理的訊息數量上限。
- cps.streamingPull.maxAckExtensionMs：連接器延長訂閱期限的時間上限 (以毫秒為單位)。
- cps.streamingPull.maxMsPerAckExtension：連接器每次延長訂閱期限的時間上限 (以毫秒為單位)。
- cps.streamingPull.parallelStreams：要從訂閱項目提取訊息的串流數量。

Pub/Sub 端點

連接器預設會使用全域 Pub/Sub 端點。如要指定端點，請將 cps.endpoint 屬性設為端點位址。如要進一步瞭解端點，請參閱「Pub/Sub 端點」。

Kafka 分區

根據預設，連接器會寫入主題中的單一分割區。如要指定連接器寫入的分區數量，請設定 kafka.partition.count 屬性。值不得超過主題的分割區計數。

如要指定連接器將訊息指派給分割區的方式，請設定 kafka.partition.scheme 屬性。詳情請參閱「Pub/Sub 來源連接器設定」。

轉換者

將鍵轉換器設為 org.apache.kafka.connect.storage.StringConverter。

視連接器設定而定，將值轉換器設為下列其中一個：

org.apache.kafka.connect.converters.ByteArrayConverter
org.apache.kafka.connect.json.JsonConverter

詳情請參閱「記錄值」。

訊息轉換

Pub/Sub 來源連接器會將 Pub/Sub 訊息轉換為 Kafka 記錄。以下各節說明轉換程序。

記錄鍵

金鑰轉換器必須是 org.apache.kafka.connect.storage.StringConverter。

錄製鍵預設為 null。
如要使用 Pub/Sub 訊息屬性做為鍵，請將 kafka.key.attribute 設為屬性的名稱。例如：kafka.key.attribute=username。
如要使用 Pub/Sub 排序鍵做為鍵，請設定 kafka.key.attribute=orderingKey。

記錄標頭

根據預設，記錄標頭為空白。

如果 kafka.record.headers 為 true，Pub/Sub 訊息屬性會寫入為記錄標頭。如要加入排序鍵，請設定 cps.makeOrderingKeyAttribute=true。

記錄價值

記錄值會以位元組陣列或 struct 型別的形式寫入。

位元組陣列記錄值

如果 kafka.record.headers 為 true，或 Pub/Sub 訊息沒有自訂屬性，連接器會將訊息資料寫入位元組陣列。將值轉換器設為 org.apache.kafka.connect.converters.ByteArrayConverter。

結構體記錄值

如果 kafka.record.headers 為 false，且訊息至少有一個自訂屬性，連接器會將記錄值寫入為 struct。將值轉換器設為 org.apache.kafka.connect.json.JsonConverter。

struct 包含下列欄位：

message：Pub/Sub 訊息資料 (以位元組表示)。
每個 Pub/Sub 訊息屬性都有一個欄位。如要加入排序鍵，請設定 cps.makeOrderingKeyAttribute=true。

舉例來說，如果訊息有 username 屬性，記錄值會如下所示：

{
  "message":"MESSAGE_DATA",
  "username":"Alice"
}

如果 value.converter.schemas.enable 是 true，則 struct 會同時包含酬載和結構定義：

{
  "schema":
    {
      "type":"struct",
      "fields": [
        {
          "type":"bytes",
          "optional":false,
          "field":"message"
        },
        {
          "type":"string",
          "optional":false,
          "field":"username"
        }
      ],
      "optional":false
    },
    "payload": {
      "message":"MESSAGE_DATA",
      "username":"Alice"
    }
}

後續步驟

Apache Kafka® 是 The Apache Software Foundation 或其在美國和/或其他國家/地區的關聯企業的註冊商標。

建立 Pub/Sub 來源連接器 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

事前準備

所需角色和權限

所需權限

授予從 Pub/Sub 讀取的權限

建立 Pub/Sub 來源連接器

控制台

gcloud

Terraform

Go

Java

Python

設定連接器

拉動模式

Pub/Sub 端點

Kafka 分區

轉換者

訊息轉換

記錄鍵

記錄標頭

記錄價值

位元組陣列記錄值

結構體記錄值

後續步驟

建立 Pub/Sub 來源連接器