Crea un connettore di origine Pub/Sub

Un connettore di origine Pub/Sub trasmette i messaggi da Pub/Sub a Managed Service per Apache Kafka, che ti consente di integrare Pub/Sub con le tue applicazioni e pipeline di dati basate su Kafka. I casi d'uso tipici includono l'importazione dati in tempo reale per l'elaborazione dei flussi e la creazione di architetture basate su eventi.

Panoramica

Questo connettore legge i messaggi da una sottoscrizione Pub/Sub, converte ogni messaggio in un record Kafka e scrive i record in un argomento Kafka. Per impostazione predefinita, il connettore crea i record Kafka nel seguente modo:

• La chiave del record Kafka è null.
• Il valore del record Kafka corrisponde ai dati del messaggio Pub/Sub sotto forma di byte.
• Le intestazioni dei record Kafka sono vuote.

Tuttavia, puoi configurare questo comportamento. Per saperne di più, consulta Configura il connettore.

Prima di iniziare

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

• Un argomento Pub/Sub con una sottoscrizione.

• Un cluster Managed Service per Apache Kafka.

• Un argomento Kafka all'interno del cluster Kafka.

• Un cluster di connessione. Quando crei il cluster Connect, imposta il cluster Managed Service per Apache Kafka come cluster Kafka principale.

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 leggere da Pub/Sub

L'agente di servizio Managed Kafka deve disporre dell'autorizzazione per leggere i messaggi dalla sottoscrizione Pub/Sub. Concedi all'agente di servizio i seguenti ruoli IAM nel progetto che contiene la sottoscrizione:

• Pub/Sub Subscriber (roles/pubsub.subscriber)
• Pub/Sub Viewer (roles/pubsub.viewer)

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.

Crea un connettore di origine Pub/Sub

Console

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

   Vai a Cluster connessi

2. Fai clic sul cluster di connessione in cui vuoi creare il connettore.

3. Fai clic su Crea connettore.

4. 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.

5. In Plugin connettore, seleziona Origine Pub/Sub.

6. Nell'elenco Sottoscrizione Cloud Pub/Sub, seleziona una sottoscrizione Pub/Sub. Il connettore esegue il pull dei messaggi da questa sottoscrizione. La sottoscrizione viene visualizzata come nome completo della risorsa: projects/{project}/subscriptions/{subscription}.

7. Nell'elenco Argomento Kafka, seleziona l'argomento Kafka in cui vengono scritti i messaggi.

8. (Facoltativo) Nella casella Configurazioni, aggiungi le proprietà di configurazione o modifica le proprietà predefinite. Per saperne di più, consulta Configura il connettore.

9. Seleziona la Policy di riavvio attività. Per saperne di più, consulta Policy di riavvio delle attività.

10. 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:

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"

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Google Cloud in cui si trova la sottoscrizione Pub/Sub.

• PUBSUB_SUBSCRIPTION_ID: l'ID della sottoscrizione Pub/Sub da cui estrarre i dati.

• KAFKA_TOPIC_ID: l'ID dell'argomento Kafka in cui vengono scritti i dati.

Le proprietà di configurazione cps.project, cps.subscription e kafka.topic sono obbligatorie. Per ulteriori opzioni di configurazione, consulta Configura il connettore.

Terraform

Puoi utilizzare una risorsa Terraform per creare un connettore.

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
}

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

// 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

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

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.

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.

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

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

Configura il connettore

Questa sezione descrive alcune proprietà di configurazione che puoi impostare sul connettore.

Per un elenco completo delle proprietà specifiche di questo connettore, consulta Configurazioni del connettore di origine Pub/Sub.

Modalità pull

La modalità di pull specifica il modo in cui il connettore recupera i messaggi Pub/Sub. Sono supportate le seguenti modalità:

• Modalità pull (predefinita). I messaggi vengono estratti in batch. Per attivare questa modalità, imposta cps.streamingPull.enabled=false.. Per configurare le dimensioni del batch, imposta la proprietà cps.maxBatchSize.

  Per saperne di più sulla modalità pull, consulta API Pull.

• Modalità pull in streaming. Consente di ottenere la massima velocità effettiva e la latenza più bassa durante il recupero dei messaggi da Pub/Sub. Per attivare questa modalità, imposta cps.streamingPull.enabled=true.

  Per saperne di più sulla modalità di pull in modalità flusso, consulta API StreamingPull.

  Se il pull in modalità flusso è abilitato, puoi ottimizzare le prestazioni impostando le seguenti proprietà di configurazione:

  • cps.streamingPull.flowControlBytes: il numero massimo di byte di messaggi in sospeso per attività.
  • cps.streamingPull.flowControlMessages: il numero massimo di messaggi in sospeso per attività.
  • cps.streamingPull.maxAckExtensionMs: il tempo massimo per cui il connettore estende la scadenza della sottoscrizione, in millisecondi.
  • cps.streamingPull.maxMsPerAckExtension: il tempo massimo per cui il connettore estende la scadenza della sottoscrizione per estensione, in millisecondi.
  • cps.streamingPull.parallelStreams: il numero di stream per eseguire il pull dei messaggi dalla sottoscrizione.

Endpoint Pub/Sub

Per impostazione predefinita, il connettore utilizza l'endpoint Pub/Sub globale. Per specificare un endpoint, imposta la proprietà cps.endpoint sull'indirizzo dell'endpoint. Per saperne di più sugli endpoint, consulta Endpoint Pub/Sub.

Partizioni Kafka

Per impostazione predefinita, il connettore scrive in una singola partizione dell'argomento. Per specificare il numero di partizioni in cui il connettore scrive, imposta la proprietà kafka.partition.count. Il valore non deve superare il numero di partizioni dell'argomento.

Per specificare il modo in cui il connettore assegna i messaggi alle partizioni, imposta la proprietà kafka.partition.scheme. Per saperne di più, consulta Configurazioni del connettore di origine Pub/Sub.

Utenti che hanno completato una conversione

Imposta il convertitore di chiavi su org.apache.kafka.connect.storage.StringConverter.

A seconda della configurazione del connettore, imposta il convertitore di valori su uno dei seguenti:

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

Per saperne di più, consulta Valore del record.

Conversione messaggi

Il connettore di origine Pub/Sub converte i messaggi Pub/Sub in record Kafka. Le sezioni seguenti descrivono il processo di conversione.

Chiave di registrazione

Il convertitore di chiavi deve essere org.apache.kafka.connect.storage.StringConverter.

• Per impostazione predefinita, le chiavi dei record sono null.

• Per utilizzare un attributo del messaggio Pub/Sub come chiave, imposta kafka.key.attribute sul nome dell'attributo. Ad esempio kafka.key.attribute=username.

• Per utilizzare la chiave di ordinamento Pub/Sub come chiave, imposta kafka.key.attribute=orderingKey.

Intestazioni dei record

Per impostazione predefinita, le intestazioni dei record sono vuote.

Se kafka.record.headers è true, gli attributi dei messaggi Pub/Sub vengono scritti come intestazioni dei record. Per includere la chiave di ordinamento, imposta cps.makeOrderingKeyAttribute=true.

Valore del record

I valori dei record vengono scritti come array di byte o come tipi struct.

Valori dei record di array di byte

Se kafka.record.headers è true o il messaggio Pub/Sub non ha attributi personalizzati, il connettore scrive i dati del messaggio come array di byte. Imposta il convertitore di valori su org.apache.kafka.connect.converters.ByteArrayConverter.

Valori dei record Struct

Se kafka.record.headers è false e il messaggio ha almeno un attributo personalizzato, il connettore scrive il valore del record come struct. Imposta il convertitore di valori su org.apache.kafka.connect.json.JsonConverter.

struct contiene i seguenti campi:

• message: i dati del messaggio Pub/Sub, in byte.

• Un campo per ogni attributo del messaggio Pub/Sub. Per includere la chiave di ordinamento, imposta cps.makeOrderingKeyAttribute=true.

Ad esempio, se il messaggio ha un attributo username, il valore del record è simile al seguente:

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

Se value.converter.schemas.enable è true, struct include sia il payload che lo schema:

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

Passaggi successivi

• Replica i messaggi Pub/Sub in Kafka

• Risolvi i problemi di un connettore Pub/Sub

• Metti in pausa, riprendi, arresta e riavvia un connettore

• Elenca tutti i connettori

• Visualizza i dettagli di un connettore

• Aggiorna un connettore

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