Créer un connecteur Cloud Storage Sink

Les connecteurs de récepteur Cloud Storage diffusent les données de Kafka vers des buckets Cloud Storage, ce qui vous permet de stocker et de traiter de grands volumes de données de manière économique et évolutive.

Voici quelques cas d'utilisation des connecteurs de récepteur Cloud Storage :

  • Ingestion de lacs de données. Stockez les données Kafka dans un lac de données pour l'archivage à long terme et le traitement par lot.

  • Archivez les données pour répondre aux exigences réglementaires.

Avant de commencer

Avant de créer un connecteur de récepteur Cloud Storage, assurez-vous de disposer des éléments suivants :

Rôles et autorisations requis

Pour obtenir les autorisations nécessaires pour créer un connecteur, demandez à votre administrateur de vous accorder le rôle IAM Éditeur de connecteur Kafka géré (roles/managedkafka.connectorEditor) sur votre projet. Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Ce rôle prédéfini contient les autorisations requises pour créer un connecteur. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Autorisations requises

Les autorisations suivantes sont requises pour créer un connecteur :

  • Créer un connecteur : managedkafka.connectors.create

Vous pouvez également obtenir ces autorisations avec des rôles personnalisés ou d'autres rôles prédéfinis.

Accorder des autorisations d'écriture dans le bucket Cloud Storage

Le compte de service Managed Kafka doit disposer des autorisations suivantes pour écrire des messages dans le bucket Cloud Storage :

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

Attribuez le rôle Utilisateur d'objets Storage (roles/storage.objectUser) au compte de service sur le projet contenant le bucket Cloud Storage.

Le compte de service Managed Kafka a le format suivant : service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, où PROJECT_NUMBER est le numéro de projet du cluster Connect.

Si votre cluster Connect se trouve dans un projet différent de votre cluster Managed Service pour Apache Kafka cluster, consultez Créer un cluster Connect dans un autre projet.

Créer un connecteur de récepteur Cloud Storage

Pour créer un connecteur de récepteur Cloud Storage, procédez comme suit.

Console

  1. Dans la Google Cloud console, accédez à la page Connect Clusters.

    Accéder aux clusters Connect

  2. Cliquez sur le cluster Connect pour lequel vous souhaitez créer le connecteur.

    La page Connect cluster details (Détails du cluster Connect) s'affiche.

  3. Cliquez sur Create connector (Créer un connecteur).

    La page Create Kafka connector (Créer un connecteur Kafka) s'affiche.

  4. Saisissez une chaîne pour le nom du connecteur.

    Pour obtenir des instructions sur la façon de nommer un connecteur, consultez les consignes de dénomination des ressources Managed Service pour Apache Kafka.

  5. Pour Connector plugin (Plug-in de connecteur), sélectionnez Cloud Storage Sink (Récepteur Cloud Storage).

  6. Spécifiez les Topics (Sujets) à partir desquels vous pouvez diffuser des données.

  7. Choisissez le Storage Bucket (Bucket de stockage) dans lequel stocker les données.

  8. (Facultatif) Configurez d'autres paramètres dans la section Configuration.

  9. Sélectionnez la Task restart policy (Règle de redémarrage des tâches). Pour en savoir plus, consultez la section Règle de redémarrage des tâches.

  10. Cliquez sur Create (Créer).

gcloud

  1. Dans la Google Cloud console, activez Cloud Shell.

    Activer Cloud Shell

    En bas de la Google Cloud console, une session Cloud Shell démarre et affiche une invite de ligne de commande. Cloud Shell est un environnement de shell dans lequel Google Cloud CLI est déjà installé, et dans lequel des valeurs sont déjà définies pour votre projet actuel. L'initialisation de la session peut prendre quelques secondes.

  2. Exécutez la gcloud managed-kafka connectors create commande :

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

    Remplacez les éléments suivants :

    • CONNECTOR_ID : ID ou nom du connecteur. Pour obtenir des instructions sur la façon de nommer un connecteur, consultez les consignes de dénomination des ressources Managed Service pour Apache Kafka. Le nom d'un connecteur est immuable.

    • LOCATION: emplacement où vous créez le connecteur. Il doit s'agir du même emplacement que celui où vous avez créé le cluster Connect.

    • CONNECT_CLUSTER_ID: ID du cluster Connect dans lequel le connecteur est créé.

    • CONFIG_FILE: chemin d'accès au fichier de configuration YAML du connecteur.

    Voici un exemple de fichier de configuration pour le connecteur de récepteur 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"

    Remplacez les éléments suivants :

    • BUCKET_NAME: nom du bucket Cloud Storage dans lequel écrire.

    • TOPIC_LIST: liste de sujets Kafka à partir desquels lire, séparés par une virgule. La liste peut également inclure des expressions régulières.

Terraform

Vous pouvez utiliser une ressource Terraform pour créer un connecteur.

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
}

Pour savoir comment appliquer ou supprimer une configuration Terraform, consultez la page Commandes Terraform de base.

Go

Avant d'essayer cet exemple, suivez les instructions de configuration pour Go décrites dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Go de Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les identifiants par défaut de l'application(ADC). Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement local.

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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Java décrites dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Java de Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement local.


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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Python décrites dans Installer les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Python de Managed Service pour Apache Kafka.

Pour vous authentifier auprès de Managed Service pour Apache Kafka, configurez les identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer les ADC pour un environnement de développement local.

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

Une fois que vous avez créé un connecteur, vous pouvez le modifier, le supprimer, le mettre en veille, l'arrêter ou le redémarrer.

Configurer le connecteur

Pour obtenir la liste complète des propriétés spécifiques à ce connecteur, consultez la section Configuration de la documentation du connecteur de récepteur Cloud Storage.

Output fields

Par défaut, le connecteur n'écrit que la valeur d'enregistrement de chaque enregistrement Kafka. Pour inclure des clés ou des en-têtes d'enregistrement, définissez la propriété format.output.fields sur une liste de champs à inclure, séparés par une virgule. Les champs acceptés sont key, value, headers et timestamp. Exemple :

format.output.fields=key,value,headers

Formats de sortie

Le connecteur est compatible avec l'écriture d'enregistrements Kafka aux formats CSV, JSON et Parquet. Les sections suivantes décrivent comment configurer le format de sortie.

Format CSV

Pour écrire des fichiers CSV dans Cloud Storage, définissez format.output.type=csv. Utilisez les convertisseurs suivants :

  • Convertisseur de clés : org.apache.kafka.connect.storage.StringConverter ou org.apache.kafka.connect.json.JsonConverter

  • Convertisseur de valeurs : org.apache.kafka.connect.converters.ByteArrayConverter

Format JSON

Pour écrire des fichiers au format JSON dans Cloud Storage, définissez format.output.type sur l'une des valeurs suivantes :

  • json: chaque fichier de sortie contient un seul tableau d'objets JSON, où chaque objet JSON représente un enregistrement Kafka.

  • jsonl: le connecteur écrit chaque enregistrement Kafka en tant qu'objet JSON sur une ligne distincte.

L'exemple suivant montre comment le connecteur écrit un enregistrement Kafka au format JSON :

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

Si vous définissez le convertisseur de clés ou de valeurs sur org.apache.kafka.connect.converters.ByteArrayConverter, le connecteur écrit ce champ en tant que valeur encodée en base64.

Le connecteur n'inclut pas de schéma JSON dans la sortie, même si value.converter.schemas.enable est défini sur true.

Format Parquet

Pour écrire des fichiers Parquet dans Cloud Storage, définissez format.output.type=parquet.

Si le convertisseur de clés ou de valeurs est org.apache.kafka.connect.json.JsonConverter, les données JSON doivent inclure un schéma. Définissez la propriété schemas.enable du convertisseur sur true.

Pour les données JSON, le connecteur écrit les données en tant que type struct dans le fichier Parquet. Par exemple, si la valeur d'enregistrement Kafka comporte un champ nommé userid, le connecteur écrit une colonne nommée value.userid. Pour aplatir les noms de colonnes, définissez format.output.envelope=false. Si format.output.envelope est false, alors format.output.fields doit spécifier un seul champ de sortie.

Paramètres de fichier Cloud Storage

Les paramètres suivants contrôlent la façon dont le connecteur écrit des fichiers dans Cloud Storage.

  • gcs.bucket.name : valeur obligatoire. Spécifie le nom du bucket Cloud Storage dans lequel le connecteur écrit des données.

  • file.compression.type : spécifie le type de compression des fichiers de sortie. Exemples : gzip, snappy, zstd et none. La valeur par défaut est none.

  • file.name.prefix : spécifie un préfixe à ajouter au nom de chaque fichier. La valeur par défaut est vide ("").

  • file.name.template : spécifie un modèle pour le nom de fichier. Le connecteur est compatible avec les variables d'espace réservé pour le modèle, telles que {{topic}} pour le nom du sujet et {{partition}} pour le numéro de partition. Pour en savoir plus, consultez la documentation du connecteur de récepteur Cloud Storage.

  • file.max.records : nombre maximal d'enregistrements à écrire dans un seul fichier. Si la valeur est 0, il n'y a pas de maximum spécifique. La valeur par défaut est 0.

  • file.max.bytes : taille maximale du fichier à écrire, en octets. Si la valeur est 0, il n'y a pas de maximum spécifique. La valeur par défaut est 0.

Regroupement d'enregistrements

Lorsque le connecteur de récepteur Cloud Storage écrit des enregistrements Kafka, il les regroupe par défaut par sujet et par partition. Par exemple, les enregistrements de la partition 0 du sujet my-topic sont écrits dans des fichiers nommés my-topic-0-OFFSET, où OFFSET est le décalage de début.

Si le paramètre file.name.template inclut la variable d'espace réservé {{key}}, le connecteur regroupe les enregistrements par clé. Dans ce mode, le connecteur écrit un fichier distinct pour chaque clé d'enregistrement. Le connecteur remplace chaque fichier par le dernier enregistrement de cette clé.

Identifiants

La propriété gcs.credentials.default doit être définie sur true. Ce paramètre permet au connecteur de découvrir Google Cloud automatiquement les identifiants.

Étape suivante

Apache Kafka® est une marque déposée d'Apache Software Foundation ou de ses filiales aux États- Unis et/ou dans d'autres pays.