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

建立 Pub/Sub 來源連接器

Pub/Sub 來源連接器會將訊息從 Pub/Sub 串流至 Managed Service for Apache 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 訂閱項目讀取訊息的權限。在包含訂閱項目的專案中，將下列 IAM 角色授予服務代理：

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

如果 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」(Pub/Sub 來源)。
在「Cloud Pub/Sub subscription」(Cloud Pub/Sub 訂閱項目) 清單中，選取 Pub/Sub 訂閱項目。連接器會從這個訂閱項目提取訊息。訂閱項目會顯示為完整資源名稱：projects/{project}/subscriptions/{subscription}。
在「Kafka topic」(Kafka 主題) 清單中，選取要寫入訊息的 Kafka 主題。
選用：在「Configurations」(設定) 方塊中，新增設定屬性或編輯預設屬性。詳情請參閱「設定連接器」。
選取「Task restart policy」(任務重新啟動政策)。詳情請參閱「工作重新啟動政策」。
點選「建立」。

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 訂閱項目所在的 Google Cloud專案 ID。
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® 是 Apache Software Foundation 或其關係企業在美國和/或其他國家/地區的註冊商標。

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

總覽

事前準備

必要角色和權限

所需權限

授予從 Pub/Sub 讀取的權限

建立 Pub/Sub 來源連接器

控制台

gcloud

Terraform

Go

Java

Python

設定連接器

提取模式

Pub/Sub 端點

Kafka 分區

轉換者

訊息轉換

記錄金鑰

記錄標頭

記錄價值

位元組陣列記錄值

結構記錄值

後續步驟

建立 Pub/Sub 來源連接器