Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Crea un connettore sink Pub/Sub

I connettori Pub/Sub Sink trasmettono i messaggi dagli argomenti Kafka agli argomenti Pub/Sub, consentendoti di integrare le tue applicazioni basate su Kafka con Pub/Sub.

Ecco alcuni casi d'uso per i connettori di sink Pub/Sub:

Architetture basate su eventi. Attiva l'elaborazione Pub/Sub dai messaggi pubblicati su Kafka.
Integrazione dei servizi. Invia i dati da Kafka ad altri servizi Google Cloud o applicazioni che utilizzano Pub/Sub.
Attivare notifiche o azioni in tempo reale in base ai dati elaborati.

Prima di iniziare

Prima di creare un connettore sink Pub/Sub, assicurati di disporre di quanto segue:

Crea un cluster Managed Service per Apache Kafka per il tuo cluster Connect. Questo è il cluster Kafka principale associato al cluster di connessione. Questo è anche il cluster di origine che costituisce un'estremità della pipeline del connettore.
Crea un cluster Connect per ospitare il connettore di sink Pub/Sub.
Crea un argomento Pub/Sub.
Crea e configura un argomento Kafka all'interno del cluster di origine. I dati vengono spostati da questo argomento Kafka all'argomento Pub/Sub di destinazione.

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per creare un connettore, chiedi all'amministratore di concederti il ruolo IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare un connettore. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per creare un connettore sono necessarie le seguenti autorizzazioni:

Crea un connettore: managedkafka.connectors.create

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Concedi le autorizzazioni per la pubblicazione nell'argomento Pub/Sub

L'agente di servizio Managed Kafka deve disporre dell'autorizzazione per pubblicare messaggi nell'argomento Pub/Sub. Concedi all'agente di servizio il ruolo Publisher Pub/Sub (roles/pubsub.publisher) nel progetto che contiene l'argomento Pub/Sub.

Se il cluster di connessione si trova in un progetto diverso da quello del cluster Managed Service per Apache Kafka, consulta Crea un cluster di connessione in un progetto diverso.

Come funziona un connettore di sink Pub/Sub

Un connettore di sink Pub/Sub estrae i messaggi da uno o più argomenti Kafka e li pubblica in un argomento Pub/Sub.

Ecco una suddivisione dettagliata di come il connettore di sink Pub/Sub copia i dati:

Il connettore utilizza i messaggi di uno o più argomenti Kafka all'interno del cluster di origine.
Il connettore scrive i messaggi nell'ID argomento Pub/Sub di destinazione specificato utilizzando la proprietà di configurazione cps.topic. Questa è una proprietà obbligatoria.
Il connettore richiede inoltre che il progetto Google Cloud contenente l'argomento Pub/Sub venga specificato utilizzando la proprietà di configurazione cps.project. Questa proprietà è obbligatoria.
Il connettore può anche utilizzare facoltativamente un endpoint Pub/Sub personalizzato specificato utilizzando la proprietà cps.endpoint. L'endpoint predefinito è "pubsub.googleapis.com:443".
Per ottimizzare le prestazioni, il connettore memorizza i messaggi nel buffer prima di pubblicarli in Pub/Sub. Puoi configurare maxBufferSize, maxBufferBytes, maxDelayThresholdMs, maxOutstandingRequestBytes e maxOutstandingMessages per controllare il buffering.
Un record Kafka ha tre componenti: intestazioni, chiavi e valori. Il connettore utilizza convertitori di chiavi e valori per trasformare i dati dei messaggi Kafka nel formato previsto da Pub/Sub. Quando utilizzi schemi di valori struct o mappa, la proprietà messageBodyName specifica il campo o la chiave da utilizzare come corpo del messaggio Pub/Sub.
Il connettore può includere l'argomento, la partizione, l'offset e il timestamp di Kafka come attributi del messaggio utilizzando la proprietà metadata.publish impostata su true.
Il connettore può includere le intestazioni dei messaggi Kafka come attributi dei messaggi Pub/Sub utilizzando la proprietà headers.publish impostata su true.
Il connettore può includere una chiave di ordinamento per i messaggi Pub/Sub utilizzando la proprietà orderingKeySource. Le opzioni per il suo valore includono "none" (predefinito), "key" e "partition".
La proprietà tasks.max controlla il livello di parallelismo per il connettore. L'aumento di tasks.max può migliorare la velocità effettiva, ma il parallelismo effettivo è limitato dal numero di partizioni negli argomenti Kafka.

Proprietà di un connettore sink Pub/Sub

Quando crei un connettore di sink Pub/Sub, devi specificare le seguenti proprietà.

Nome connettore

Un nome univoco per il connettore all'interno del cluster di connessione. Per linee guida sulla denominazione delle risorse, consulta Linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.

Tipo di plug-in del connettore

Seleziona Pub/Sub Sink come tipo di plug-in del connettore. In questo modo viene determinata la direzione del flusso di dati, che va da Kafka a Pub/Sub, e l'implementazione del connettore specifica utilizzata. Se non utilizzi l'interfaccia utente per configurare il connettore, devi specificare anche la classe del connettore.

Argomenti Kafka

Gli argomenti Kafka da cui il connettore utilizza i messaggi. Puoi specificare uno o più argomenti o utilizzare un'espressione regolare per trovare corrispondenze con più argomenti. Ad esempio, topic.* per trovare tutti gli argomenti che iniziano con "topic". Questi argomenti devono esistere all'interno del cluster Managed Service per Apache Kafka associato al tuo cluster Connect.

Argomento Pub/Sub

L'argomento Pub/Sub esistente in cui il connettore pubblica i messaggi. Assicurati che il account di servizio del cluster Connect abbia il ruolo roles/pubsub.publisher nel progetto dell'argomento, come descritto in Prima di iniziare.

Configurazione

Questa sezione consente di specificare proprietà di configurazione aggiuntive specifiche del connettore.

Poiché i dati negli argomenti Kafka possono essere in vari formati, come Avro, JSON o byte non elaborati, una parte fondamentale della configurazione prevede la specifica dei convertitori. I convertitori traducono i dati dal formato utilizzato negli argomenti Kafka al formato interno standardizzato di Kafka Connect. Il connettore Pub/Sub Sink prende quindi questi dati interni e li trasforma nel formato richiesto da Pub/Sub prima di scriverli.

Per informazioni più generali sul ruolo dei convertitori in Kafka Connect, sui tipi di convertitori supportati e sulle opzioni di configurazione comuni, consulta Convertitori.

Di seguito sono riportate alcune configurazioni specifiche per il connettore sink Pub/Sub:

cps.project: specifica l'ID progetto Google Cloud che contiene l'argomento Pub/Sub.
cps.topic: specifica l'argomento Pub/Sub in cui vengono pubblicati i dati.
cps.endpoint: specifica l'endpoint Pub/Sub da utilizzare.

Per un elenco delle proprietà di configurazione disponibili specifiche per questo connettore, consulta le configurazioni del connettore Pub/Sub Sink.

Crea un connettore di sink Pub/Sub

Prima di creare un connettore, consulta la documentazione relativa alle proprietà di un connettore sink Pub/Sub.

Console

Nella console Google Cloud , vai alla pagina Connetti i cluster.

Vai a Cluster connessi
Fai clic sul cluster di connessione per cui vuoi creare il connettore.

Viene visualizzata la pagina Dettagli cluster di connessione.
Fai clic su Crea connettore.

Viene visualizzata la pagina Crea connettore Kafka.
Inserisci una stringa per il nome del connettore.

Per maggiori informazioni su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.
In Plugin connettore, seleziona Pub/Sub Sink.
In Argomenti, scegli Seleziona un elenco di argomenti Kafka o Utilizza un'espressione regolare per gli argomenti. Quindi, seleziona o inserisci gli argomenti Kafka da cui questo connettore utilizza i messaggi. Questi argomenti si trovano nel cluster Kafka associato.
In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento Pub/Sub in cui questo connettore pubblica i messaggi. L'argomento viene visualizzato nel formato del nome completo della risorsa: projects/{project}/topics/{topic}.
(Facoltativo) Configura impostazioni aggiuntive nella sezione Configurazioni. Qui puoi specificare proprietà come tasks.max, key.converter e value.converter, come descritto nella sezione precedente.
Seleziona la Policy di riavvio attività. Per saperne di più, consulta Policy di riavvio delle attività.
Fai clic su Crea.

gcloud

Esegui il comando gcloud managed-kafka connectors create:

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

Sostituisci quanto segue:

CONNECTOR_ID: l'ID o il nome del connettore.
CONNECT_CLUSTER_ID: l'ID del cluster Connect in cui viene creato il connettore.
LOCATION: la località del cluster Connect.
CONFIG_FILE: il percorso di un file di configurazione YAML o JSON locale.

Ecco un esempio di file di configurazione per il connettore sink Pub/Sub:

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"

Sostituisci quanto segue:

CPS_SINK_CONNECTOR_ID: l'ID o il nome del connettore Pub/Sub Sink. Per maggiori informazioni su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Il nome di un connettore è immutabile.
GMK_TOPIC_ID: l'ID dell'argomento Managed Service per Apache Kafka da cui i dati vengono letti dal connettore di sink Pub/Sub.
CPS_TOPIC_ID: l'ID dell'argomento Pub/Sub in cui vengono pubblicati i dati.
GCP_PROJECT_ID: l'ID del progetto Google Cloud in cui si trova l'argomento Pub/Sub.

Terraform

Puoi utilizzare una risorsa Terraform per creare un connettore.

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
}

Per scoprire come applicare o rimuovere una configurazione Terraform, consulta Comandi Terraform di base.

Vai

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Go di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione(ADC). Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.

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

Prima di provare questo esempio, segui le istruzioni di configurazione di Java riportate in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.


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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python di Managed Service per 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}")

Dopo aver creato un connettore, puoi modificarlo, eliminarlo, metterlo in pausa, arrestarlo o riavviarlo.

Passaggi successivi

Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue società affiliate negli Stati Uniti e/o in altri paesi.

Crea un connettore sink Pub/Sub Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.