Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Membuat konektor Sink Pub/Sub

Konektor Pub/Sub Sink mengalirkan pesan dari topik Kafka ke topik Pub/Sub, sehingga Anda dapat mengintegrasikan aplikasi berbasis Kafka dengan Pub/Sub.

Kasus penggunaan untuk konektor Pub/Sub Sink mencakup hal berikut:

Arsitektur berbasis peristiwa. Memicu pemrosesan Pub/Sub dari pesan yang dipublikasikan ke Kafka.
Integrasi layanan. Mengirim data dari Kafka ke layanan atau aplikasi Google Cloud lain yang menggunakan data dari Pub/Sub.
Memicu notifikasi atau tindakan real-time berdasarkan data yang diproses.

Sebelum memulai

Sebelum membuat konektor Pub/Sub Sink, pastikan Anda memiliki hal berikut:

Buat cluster Managed Service untuk Apache Kafka untuk cluster Connect Anda. Ini adalah cluster Kafka utama yang terkait dengan cluster Connect. Cluster ini juga merupakan cluster sumber yang membentuk salah satu ujung pipeline konektor.
Buat cluster Connect untuk menghosting konektor Pub/Sub Sink Anda.
Membuat topik Pub/Sub.
Buat dan konfigurasikan topik Kafka dalam cluster sumber. Data berpindah dari topik Kafka ini ke topik Pub/Sub tujuan.

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang diperlukan guna membuat konektor, minta administrator untuk memberi Anda peran IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat konektor. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat konektor:

Buat konektor: managedkafka.connectors.create

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Memberikan izin untuk memublikasikan ke topik Pub/Sub

Agen Layanan Kafka Terkelola harus memiliki izin untuk memublikasikan pesan ke topik Pub/Sub. Berikan peran Pub/Sub Publisher (roles/pubsub.publisher) kepada agen layanan di project yang berisi topik Pub/Sub.

Jika cluster Connect Anda berada di project yang berbeda dengan cluster Managed Service untuk Apache Kafka, lihat Membuat cluster Connect di project yang berbeda.

Cara kerja konektor Pub/Sub Sink

Konektor Pub/Sub Sink mengambil pesan dari satu atau beberapa topik Kafka dan memublikasikannya ke topik Pub/Sub.

Berikut adalah perincian mendetail tentang cara konektor Pub/Sub Sink menyalin data:

Konektor menggunakan pesan dari satu atau beberapa topik Kafka dalam cluster sumber.
Konektor menulis pesan ke ID topik Pub/Sub target yang ditentukan menggunakan properti konfigurasi cps.topic. Properti ini wajib diisi.
Konektor juga memerlukan project Google Cloud yang berisi topik Pub/Sub untuk ditentukan menggunakan properti konfigurasi cps.project. Properti ini wajib diisi.
Konektor juga dapat secara opsional menggunakan endpoint Pub/Sub kustom yang ditentukan menggunakan properti cps.endpoint. Endpoint defaultnya adalah "pubsub.googleapis.com:443".
Untuk mengoptimalkan performa, konektor akan mem-buffer pesan sebelum memublikasikannya ke Pub/Sub. Anda dapat mengonfigurasi maxBufferSize, maxBufferBytes, maxDelayThresholdMs, maxOutstandingRequestBytes, dan maxOutstandingMessages untuk mengontrol buffering.
Data Kafka memiliki tiga komponen: header, kunci, nilai. Konektor menggunakan pengonversi kunci dan nilai untuk mengubah data pesan Kafka menjadi format yang diharapkan oleh Pub/Sub. Saat menggunakan skema nilai struct atau peta, properti messageBodyName menentukan kolom atau kunci yang akan digunakan sebagai isi pesan Pub/Sub.
Konektor dapat menyertakan topik, partisi, offset, dan stempel waktu Kafka sebagai atribut pesan dengan menggunakan properti metadata.publish yang ditetapkan ke true.
Konektor dapat menyertakan header pesan Kafka sebagai atribut pesan Pub/Sub dengan menggunakan properti headers.publish yang ditetapkan ke true.
Konektor dapat menyertakan kunci pengurutan untuk pesan Pub/Sub menggunakan properti orderingKeySource. Opsi untuk nilainya mencakup "none" (default), "key", dan "partition".
Properti tasks.max mengontrol tingkat paralelisme untuk konektor. Meningkatkan tasks.max dapat meningkatkan throughput, tetapi paralelisme yang sebenarnya dibatasi oleh jumlah partisi dalam topik Kafka.

Properti konektor Pub/Sub Sink

Saat membuat konektor Pub/Sub Sink, Anda harus menentukan properti berikut.

Nama konektor

Nama unik untuk konektor dalam cluster Connect. Untuk panduan penamaan resource, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.

Jenis plugin konektor

Pilih Pub/Sub Sink sebagai jenis plugin konektor. Hal ini menentukan arah aliran data, yaitu dari Kafka ke Pub/Sub dan implementasi konektor spesifik yang digunakan. Jika Anda tidak menggunakan antarmuka pengguna untuk mengonfigurasi konektor, Anda juga harus menentukan class konektor.

Topik Kafka

Topik Kafka tempat konektor menggunakan pesan. Anda dapat menentukan satu atau beberapa topik, atau menggunakan ekspresi reguler untuk mencocokkan beberapa topik. Misalnya, topic.* untuk mencocokkan semua topik yang diawali dengan "topic". Topik ini harus ada dalam cluster Managed Service untuk Apache Kafka yang terkait dengan cluster Connect Anda.

Topik Pub/Sub

Topik Pub/Sub yang ada tempat konektor memublikasikan pesan. Pastikan akun layanan cluster Connect memiliki peran roles/pubsub.publisher di project topik, seperti yang dijelaskan dalam Sebelum memulai.

Konfigurasi

Bagian ini memungkinkan Anda menentukan properti konfigurasi tambahan khusus konektor.

Karena data dalam topik Kafka dapat dalam berbagai format seperti Avro, JSON, atau byte mentah, bagian penting dari konfigurasi melibatkan penentuan konverter. Konverter menerjemahkan data dari format yang digunakan dalam topik Kafka Anda ke format internal Kafka Connect yang terstandardisasi. Konektor Pub/Sub Sink kemudian mengambil data internal ini dan mengubahnya menjadi format yang diperlukan oleh Pub/Sub sebelum menulisnya.

Untuk mengetahui informasi umum selengkapnya tentang peran pengonversi di Kafka Connect, jenis pengonversi yang didukung, dan opsi konfigurasi umum, lihat pengonversi.

Berikut beberapa konfigurasi khusus untuk konektor Pub/Sub Sink:

cps.project: Menentukan project ID Google Cloud yang berisi topik Pub/Sub.
cps.topic: Menentukan topik Pub/Sub tempat data dipublikasikan.
cps.endpoint: Menentukan endpoint Pub/Sub yang akan digunakan.

Untuk mengetahui daftar properti konfigurasi yang tersedia khusus untuk konektor ini, lihat Konfigurasi konektor Pub/Sub Sink.

Membuat konektor Pub/Sub Sink

Sebelum membuat konektor, tinjau dokumentasi untuk Properti konektor Pub/Sub Sink.

Konsol

Di konsol Google Cloud , buka halaman Connect Clusters.

Buka Connect Clusters
Klik cluster Connect yang ingin Anda buatkan konektornya.

Halaman Connect cluster details akan ditampilkan.
Klik Create connector.

Halaman Create Kafka connector akan ditampilkan.
Untuk nama konektor, masukkan string.

Untuk panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.
Untuk Connector plugin, pilih Pub/Sub Sink.
Di bagian Topics, pilih Select a list of Kafka topics atau Use a topic regex. Kemudian, pilih atau masukkan topik Kafka tempat konektor ini menggunakan pesan. Topik ini ada di cluster Kafka terkait Anda.
Untuk Select a Cloud Pub/Sub topic, pilih topik Pub/Sub tempat konektor ini memublikasikan pesan. Topik ditampilkan dalam format nama resource lengkap: projects/{project}/topics/{topic}.
(Opsional) Konfigurasi setelan tambahan di bagian Configurations. Di sinilah Anda akan menentukan properti seperti tasks.max, key.converter, dan value.converter, seperti yang dibahas di bagian sebelumnya.
Pilih Task restart policy. Untuk mengetahui informasi selengkapnya, lihat Kebijakan memulai ulang tugas.
Klik Create.

gcloud

Jalankan perintah gcloud managed-kafka connectors create:

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

Ganti kode berikut:

CONNECTOR_ID: ID atau nama konektor.
CONNECT_CLUSTER_ID: ID cluster Connect tempat konektor dibuat.
LOCATION: Lokasi cluster Connect.
CONFIG_FILE: Jalur ke file konfigurasi YAML atau JSON lokal.

Berikut adalah contoh file konfigurasi untuk konektor Pub/Sub Sink:

connector.class: "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector"
name: "CPS_SINK_CONNECTOR_ID"
tasks.max: "1"
topics: "GMK_TOPIC_ID"
value.converter: "org.apache.kafka.connect.storage.StringConverter"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
cps.topic: "CPS_TOPIC_ID"
cps.project: "GCP_PROJECT_ID"

Ganti kode berikut:

CPS_SINK_CONNECTOR_ID: ID atau nama konektor Pub/Sub Sink. Untuk panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka. Nama konektor tidak dapat diubah.
GMK_TOPIC_ID: ID topik Managed Service untuk Apache Kafka yang datanya dibaca oleh konektor Pub/Sub Sink.
CPS_TOPIC_ID: ID topik Pub/Sub tempat data dipublikasikan.
GCP_PROJECT_ID: ID project Google Cloud tempat topik Pub/Sub Anda berada.

Terraform

Anda dapat menggunakan resource Terraform untuk membuat konektor.

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

  configs = {
    "connector.class" = "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector"
    "name"            = "my-pubsub-sink-connector"
    "tasks.max"       = "3"
    "topics"          = "TOPIC_NAME"
    "cps.topic"       = "CPS_TOPIC_NAME"
    "cps.project"     = "CPS_PROJECT_NAME"
    "value.converter" = "org.apache.kafka.connect.storage.StringConverter"
    "key.converter"   = "org.apache.kafka.connect.storage.StringConverter"
  }

  provider = google-beta
}

Untuk mempelajari cara menerapkan atau menghapus konfigurasi Terraform, lihat Perintah dasar Terraform.

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Go API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi(ADC). Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

import (
	"context"
	"fmt"
	"io"

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

// createPubSubSinkConnector creates a Pub/Sub Sink connector.
func createPubSubSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, valueConverter, keyConverter, cpsTopic, cpsProject, tasksMax 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_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// valueConverter := "org.apache.kafka.connect.storage.StringConverter"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// cpsTopic := "CPS_TOPIC_ID"
	// cpsProject := "GCP_PROJECT_ID"
	// tasksMax := "3"
	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 Sink sample connector configuration
	config := map[string]string{
		"connector.class": "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector",
		"name":            connectorID,
		"tasks.max":       tasksMax,
		"topics":          topics,
		"value.converter": valueConverter,
		"key.converter":   keyConverter,
		"cps.topic":       cpsTopic,
		"cps.project":     cpsProject,
	}

	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 sink connector: %s\n", resp.Name)
	return nil
}

Java

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Java API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.


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 CreatePubSubSinkConnector {

  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-sink-connector";
    String pubsubProjectId = "my-pubsub-project-id";
    String pubsubTopicName = "my-pubsub-topic";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector";
    String maxTasks = "3";
    String valueConverter = "org.apache.kafka.connect.storage.StringConverter";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createPubSubSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        pubsubProjectId,
        pubsubTopicName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        valueConverter,
        keyConverter);
  }

  public static void createPubSubSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String pubsubProjectId,
      String pubsubTopicName,
      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("topics", kafkaTopicName);
    configMap.put("value.converter", valueConverter);
    configMap.put("key.converter", keyConverter);
    configMap.put("cps.topic", pubsubTopicName);
    configMap.put("cps.project", pubsubProjectId);

    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 Sink connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Python API Managed Service untuk Apache Kafka.

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.sink.CloudPubSubSinkConnector",
    "name": connector_id,
    "tasks.max": tasks_max,
    "topics": topics,
    "value.converter": value_converter,
    "key.converter": key_converter,
    "cps.topic": cps_topic,
    "cps.project": cps_project,
}

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}")

Setelah membuat konektor, Anda dapat mengedit, menghapus, menjeda, menghentikan, atau memulai ulang konektor.

Apa langkah selanjutnya?

Apache Kafka® adalah merek dagang terdaftar dari Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lainnya.

Membuat konektor Sink Pub/Sub Tetap teratur dengan koleksi Simpan dan kategorikan konten berdasarkan preferensi Anda.