Crea un connettore Cloud Storage Sink

Un connettore di sink Cloud Storage trasmette i dati da Managed Service per Apache Kafka a Cloud Storage, che ti consente di archiviare ed elaborare grandi volumi di dati in modo scalabile e conveniente.

I casi d'uso tipici di questo connettore includono:

• Importazione dei data lake. Archivia i dati Kafka in un data lake per l'archiviazione a lungo termine e l'elaborazione batch.

• Archivia i dati per soddisfare i requisiti normativi.

Prima di iniziare

Prima di creare un connettore sink Cloud Storage, 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 Cloud Storage.

• Crea un bucket Cloud Storage per archiviare i dati trasmessi in streaming da Kafka.

• Crea e configura un argomento Kafka all'interno del cluster di origine. I dati vengono spostati da questo argomento Kafka al bucket Cloud Storage 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 scrivere nel bucket Cloud Storage

L'agente di servizio Managed Kafka deve disporre delle seguenti autorizzazioni per scrivere messaggi nel bucket Cloud Storage:

• storage.objects.create
• storage.objects.delete

Per concedere queste autorizzazioni, assegna all'agente di servizio il ruolo Storage Object User (roles/storage.objectUser) nel progetto che contiene il bucket Cloud Storage.

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 sink Cloud Storage

Per creare un connettore di sink Cloud Storage, segui questi passaggi.

Console

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

   Vai a Cluster connessi

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

   Viene visualizzata la pagina Dettagli cluster di connessione.

3. Fai clic su Crea connettore.

   Viene visualizzata la pagina Crea connettore Kafka.

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 Plug-in connettore, seleziona Cloud Storage Sink.

6. Specifica gli argomenti da cui puoi trasmettere i dati.

7. Scegli il bucket di archiviazione in cui archiviare i dati.

8. (Facoltativo) Configura impostazioni aggiuntive nella sezione Configurazione.

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 per il connettore sink di Cloud Storage:

connector.class: "io.aiven.kafka.connect.gcs.GcsSinkConnector"
gcs.bucket.name: "BUCKET_NAME"
gcs.credentials.default: "true"
format.output.type: "json"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
topics: "TOPIC_LIST"
tasks.max: "3"

Sostituisci quanto segue:

• BUCKET_NAME: il nome del bucket Cloud Storage in cui scrivere.

• TOPIC_LIST: un elenco separato da virgole di argomenti Kafka da cui leggere. L'elenco può includere anche espressioni regolari.

Terraform

Puoi utilizzare una risorsa Terraform per creare un connettore.

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

  configs = {
    "connector.class"                = "io.aiven.kafka.connect.gcs.GcsSinkConnector"
    "tasks.max"                      = "3"
    "topics"                         = "GMK_TOPIC_ID"
    "gcs.bucket.name"                = "GCS_BUCKET_NAME"
    "gcs.credentials.default"        = "true"
    "format.output.type"             = "json"
    "name"                           = "my-gcs-sink-connector"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "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"
)

// createCloudStorageSinkConnector creates a Cloud Storage Sink connector.
func createCloudStorageSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, gcsBucketName, tasksMax, formatOutputType, valueConverter, valueConverterSchemasEnable, keyConverter, gcsCredentialsDefault 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 := "GCS_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// gcsBucketName := "GCS_BUCKET_NAME"
	// tasksMax := "3"
	// formatOutputType := "json"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// gcsCredentialsDefault := "true"
	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)

	config := map[string]string{
		"connector.class":                "io.aiven.kafka.connect.gcs.GcsSinkConnector",
		"tasks.max":                      tasksMax,
		"topics":                         topics,
		"gcs.bucket.name":                gcsBucketName,
		"gcs.credentials.default":        gcsCredentialsDefault,
		"format.output.type":             formatOutputType,
		"name":                           connectorID,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"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 Cloud Storage 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 CreateCloudStorageSinkConnector {

  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-gcs-sink-connector";
    String bucketName = "my-gcs-bucket";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "io.aiven.kafka.connect.gcs.GcsSinkConnector";
    String maxTasks = "3";
    String gcsCredentialsDefault = "true";
    String formatOutputType = "json";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createCloudStorageSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bucketName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        gcsCredentialsDefault,
        formatOutputType,
        valueConverter,
        valueSchemasEnable,
        keyConverter);
  }

  public static void createCloudStorageSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bucketName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String gcsCredentialsDefault,
      String formatOutputType,
      String valueConverter,
      String valueSchemasEnable,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("tasks.max", maxTasks);
    configMap.put("topics", kafkaTopicName);
    configMap.put("gcs.bucket.name", bucketName);
    configMap.put("gcs.credentials.default", gcsCredentialsDefault);
    configMap.put("format.output.type", formatOutputType);
    configMap.put("name", connectorId);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    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 Cloud Storage 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.

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": "io.aiven.kafka.connect.gcs.GcsSinkConnector",
    "tasks.max": tasks_max,
    "topics": topics,
    "gcs.bucket.name": gcs_bucket_name,
    "gcs.credentials.default": "true",
    "format.output.type": format_output_type,
    "name": connector_id,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "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

Per un elenco completo delle proprietà specifiche di questo connettore, consulta Configurazione nella documentazione del connettore sink di Cloud Storage.

Campi di output

Per impostazione predefinita, il connettore scrive solo il valore del record di ogni record Kafka. Per includere chiavi o intestazioni dei record, imposta la proprietà format.output.fields su un elenco di campi da includere separati da virgole. I campi supportati sono key, value, headers e timestamp. Ad esempio:

format.output.fields=key,value,headers

Formati di output

Il connettore supporta la scrittura di record Kafka in formato CSV, JSON e Parquet. Le sezioni seguenti descrivono come configurare il formato di output.

Formato CSV

Per scrivere file CSV in Cloud Storage, imposta format.output.type=csv. Utilizza i seguenti convertitori:

• Convertitore di chiavi: org.apache.kafka.connect.storage.StringConverter o org.apache.kafka.connect.json.JsonConverter

• Convertitore di valori: org.apache.kafka.connect.converters.ByteArrayConverter

Formato JSON

Per scrivere file in formato JSON in Cloud Storage, imposta format.output.type su uno dei seguenti valori:

• json: ogni file di output contiene un singolo array di oggetti JSON, in cui ogni oggetto JSON rappresenta un record Kafka.

• jsonl: il connettore scrive ogni record Kafka come oggetto JSON su una riga separata.

L'esempio seguente mostra come il connettore scrive un record Kafka in formato JSON:

{"headers":[{"key":"k1","value":"v1"},{"key":"k2","value":"v2"}],"value":"Hello world","key":"my-key"}

Se imposti il convertitore di chiavi o valori su org.apache.kafka.connect.converters.ByteArrayConverter, il connettore scrive il campo come valore codificato in Base64.

Il connettore non include uno schema JSON nell'output, anche se value.converter.schemas.enable è true.

Formato Parquet

Per scrivere file Parquet in Cloud Storage, imposta format.output.type=parquet.

Se il convertitore di chiavi o valori è org.apache.kafka.connect.json.JsonConverter, i dati JSON devono includere uno schema. Imposta la proprietà schemas.enable del convertitore su true.

Per i dati JSON, il connettore scrive i dati come tipo struct nel file Parquet. Ad esempio, se il valore del record Kafka ha un campo denominato userid, il connettore scrive una colonna denominata value.userid. Per appiattire i nomi delle colonne, imposta format.output.envelope=false. Se format.output.envelope è false, format.output.fields deve specificare un singolo campo di output.

Impostazioni dei file di Cloud Storage

Le seguenti impostazioni controllano il modo in cui il connettore scrive i file in Cloud Storage.

• gcs.bucket.name: obbligatorio. Specifica il nome del bucket Cloud Storage in cui il connettore scrive i dati.

• file.compression.type: specifica il tipo di compressione per i file di output. Alcuni esempi sono gzip, snappy, zstd e none. Il valore predefinito è none.

• file.name.prefix: specifica un prefisso da aggiungere al nome di ogni file. Il valore predefinito è vuoto ("").

• file.name.template: specifica un template per il nome file. Il connettore supporta le variabili segnaposto per il template, ad esempio {{topic}} per il nome dell'argomento e {{partition}} per il numero di partizione. Per saperne di più, consulta la documentazione del connettore sink di Cloud Storage.

• file.max.records: il numero massimo di record da scrivere in un singolo file. Se il valore è 0, non esiste un valore massimo specifico. Il valore predefinito è 0.

• file.max.bytes: la dimensione massima del file da scrivere, in byte. Se il valore è 0, non esiste un limite massimo specifico. Il valore predefinito è 0.

Raggruppamento dei record

Quando il connettore di sink Cloud Storage scrive i record Kafka, per impostazione predefinita li raggruppa per argomento e partizione. Ad esempio, i record della partizione 0 dell'argomento my-topic vengono scritti in file denominati my-topic-0-OFFSET, dove OFFSET è l'offset iniziale.

Se il parametro file.name.template include la variabile segnaposto {{key}}, il connettore raggruppa i record per chiave. In questa modalità, il connettore scrive un file separato per ogni chiave di record. Il connettore sovrascrive ogni file con il record più recente per quella chiave.

Credenziali

La proprietà gcs.credentials.default deve essere impostata su true. Questa impostazione consente al connettore di rilevare automaticamente le credenziali Google Cloud .

Passaggi successivi

• Risolvi i problemi di un connettore sink Cloud Storage

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