Crea un connettore Cloud Storage Sink

I connettori di sink Cloud Storage trasmettono i dati da Kafka ai bucket Cloud Storage, consentendoti di archiviare ed elaborare grandi volumi di dati in modo economico e scalabile.

I casi d'uso dei connettori di sink Cloud Storage includono i seguenti:

  • Importazione del data lake. Archivia i dati di 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 di sink Cloud Storage, assicurati di avere quanto segue:

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per creare un connettore, chiedi all'amministratore di concederti il ruolo IAM Editor connettore Kafka gestito (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

Il account di servizio Kafka gestito deve disporre delle seguenti autorizzazioni per scrivere messaggi nel bucket Cloud Storage:

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

Concedi il ruolo Storage Object User (roles/storage.objectUser) al account di servizio nel progetto contenente il bucket Cloud Storage.

Il account di servizio Kafka gestito ha il seguente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, dove PROJECT_NUMBER è il numero del progetto del cluster di connessione.

Se il cluster di connessione si trova in un progetto diverso dal 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 Google Cloud console, vai alla pagina Cluster di connessione.

    Vai a Cluster di connessione

  2. Fai clic sul cluster di connessione per il quale 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. Per il nome del connettore, inserisci una stringa.

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

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

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

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

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

  10. Fai clic su Crea.

gcloud

  1. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

    Nella parte inferiore della Google Cloud console viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già inclusa e installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

  2. Esegui il gcloud managed-kafka connectors create comando:

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

    Sostituisci quanto segue:

    • CONNECTOR_ID: l'ID o 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. Il nome di un connettore non può essere modificato.

    • LOCATION: la località in cui crei il connettore. Deve essere la stessa località in cui hai creato il cluster di connessione.

    • CONNECT_CLUSTER_ID: l'ID del cluster di connessione in cui viene creato il connettore.

    • CONFIG_FILE: il percorso del file di configurazione YAML per il connettore.

    Di seguito è riportato un esempio di file di configurazione per il connettore di sink 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 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 la sezione Configurazione nella documentazione del connettore di sink 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 separato da virgole dei campi da includere. 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 con codifica 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, allora 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. Esempi: 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 modello per il nome file. Il connettore supporta le variabili segnaposto per il modello, ad esempio {{topic}} per il nome dell'argomento e {{partition}} per il numero di partizione. Per saperne di più, consulta la documentazione del connettore di sink Cloud Storage.

  • file.max.records: il numero massimo di record da scrivere in un singolo file. Se il valore è 0, non esiste un 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 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 l'ultimo record per quella chiave.

Credenziali

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

Passaggi successivi

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