Crea un conector de receptor de Cloud Storage

Un conector de receptor de Cloud Storage transmite datos de Managed Service para Apache Kafka a un bucket de Cloud Storage, lo que te permite almacenar y procesar grandes volúmenes de datos de manera rentable y escalable.

Entre los casos de uso típicos de este conector, se incluyen los siguientes:

  • Transferencia de data lakes Almacena datos de Kafka en un data lake para el archivado a largo plazo y el procesamiento por lotes.

  • Archivar datos para cumplir con los requisitos reglamentarios

Antes de comenzar

Antes de crear un conector de receptor de Cloud Storage, asegúrate de tener lo siguiente:

Roles y permisos requeridos

Para obtener los permisos que necesitas para crear un conector, pídele a tu administrador que te otorgue el rol de IAM de editor de conectores de Kafka administrados (roles/managedkafka.connectorEditor) en tu proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear un conector. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para crear un conector:

  • Crea un conector: managedkafka.connectors.create

También puedes obtener estos permisos con roles personalizados o con otros roles predefinidos.

Otorga permisos para escribir en el bucket de Cloud Storage

El agente de servicio de Kafka administrado debe tener los siguientes permisos para escribir mensajes en el bucket de Cloud Storage:

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

Para otorgar estos permisos, asigna al agente de servicio el rol de usuario de objetos de almacenamiento (roles/storage.objectUser) en el proyecto que contiene el bucket de Cloud Storage.

Si tu clúster de Connect está en un proyecto diferente al de tu clúster de Managed Service para Apache Kafka, consulta Crea un clúster de Connect en un proyecto diferente.

Crea un conector de receptor de Cloud Storage

Para crear un conector de receptor de Cloud Storage, sigue estos pasos.

Console

  1. En la consola de Google Cloud , ve a la página Conectar clústeres.

    Ir a Conectar clústeres

  2. Haz clic en el clúster de Connect para el que deseas crear el conector.

    Se muestra la página Detalles de la conexión del clúster.

  3. Haz clic en Crear conector.

    Aparecerá la página Crear conector de Kafka.

  4. Para el nombre del conector, ingresa una cadena.

    Si necesitas ayuda para asignarle un nombre a un conector, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka.

  5. En Complemento del conector, selecciona Receptor de Cloud Storage.

  6. Especifica los temas desde los que puedes transmitir datos.

  7. Elige el bucket de almacenamiento para almacenar los datos.

  8. (Opcional) Establece opciones de configuración adicionales en la sección Configuración.

  9. Selecciona la Política de reinicio de tareas. Para obtener más información, consulta Política de reinicio de tareas.

  10. Haz clic en Crear.

gcloud

  1. En la consola de Google Cloud , activa Cloud Shell.

    Activa Cloud Shell

    En la parte inferior de la consola de Google Cloud , se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.

  2. Ejecuta el comando gcloud managed-kafka connectors create:

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

    Reemplaza lo siguiente:

    • CONNECTOR_ID: Es el ID o el nombre del conector. Si necesitas ayuda para asignarle un nombre a un conector, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka. El nombre de un conector es inmutable.

    • LOCATION: Es la ubicación en la que creas el conector. Debe ser la misma ubicación en la que creaste el clúster de Connect.

    • CONNECT_CLUSTER_ID: Es el ID del clúster de Connect en el que se crea el conector.

    • CONFIG_FILE: Es la ruta de acceso al archivo de configuración YAML del conector.

    A continuación, se muestra un ejemplo de un archivo de configuración para el conector de receptor de 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"

    Reemplaza lo siguiente:

    • BUCKET_NAME: Es el nombre del bucket de Cloud Storage en el que se escribirá.

    • TOPIC_LIST: Es una lista separada por comas de los temas de Kafka desde los que se leerá. La lista también puede incluir expresiones regulares.

Terraform

Puedes usar un recurso de Terraform para crear un conector.

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
}

Si deseas obtener más información para aplicar o quitar una configuración de Terraform, consulta los comandos básicos de Terraform.

Go

Antes de probar esta muestra, sigue las instrucciones de configuración de Go en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Go de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación(ADC). Para obtener más información, consulta Configura ADC para un entorno de desarrollo 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

Antes de probar esta muestra, sigue las instrucciones de configuración de Java en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Java de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo 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

Antes de probar este ejemplo, sigue las instrucciones de configuración para Python que encontrarás en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Python de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo 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}")

Después de crear un conector, puedes editarlo, borrarlo, pausarlo, detenerlo o reiniciarlo.

Configura el conector

Para obtener una lista completa de las propiedades específicas de este conector, consulta Configuración en la documentación del conector de Cloud Storage Sink.

Output fields

De forma predeterminada, el conector escribe solo el valor de cada registro de Kafka. Para incluir claves o encabezados de registros, establece la propiedad format.output.fields en una lista separada por comas de campos que se incluirán. Los campos compatibles son key, value, headers y timestamp. Por ejemplo:

format.output.fields=key,value,headers

Formatos de salida

El conector admite la escritura de registros de Kafka en formato CSV, JSON y Parquet. En las siguientes secciones, se describe cómo configurar el formato de salida.

Formato CSV

Para escribir archivos CSV en Cloud Storage, configura format.output.type=csv. Usa los siguientes convertidores:

  • Convertidor de claves: org.apache.kafka.connect.storage.StringConverter o org.apache.kafka.connect.json.JsonConverter

  • Conversor de valores: org.apache.kafka.connect.converters.ByteArrayConverter

Formato JSON

Para escribir archivos con formato JSON en Cloud Storage, establece format.output.type en uno de los siguientes valores:

  • json: Cada archivo de salida contiene un solo array de objetos JSON, en el que cada objeto JSON representa un registro de Kafka.

  • jsonl: El conector escribe cada registro de Kafka como un objeto JSON en una línea separada.

En el siguiente ejemplo, se muestra cómo el conector escribe un registro de Kafka en formato JSON:

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

Si configuras el convertidor de claves o valores en org.apache.kafka.connect.converters.ByteArrayConverter, el conector escribe ese campo como un valor codificado en base64.

El conector no incluye un esquema JSON en el resultado, incluso si value.converter.schemas.enable es true.

Formato Parquet

Para escribir archivos de Parquet en Cloud Storage, configura format.output.type=parquet.

Si el convertidor de claves o valores es org.apache.kafka.connect.json.JsonConverter, los datos JSON deben incluir un esquema. Establece la propiedad schemas.enable del conversor en true.

Para los datos JSON, el conector escribe los datos como un tipo struct en el archivo Parquet. Por ejemplo, si el valor del registro de Kafka tiene un campo llamado userid, el conector escribe una columna llamada value.userid. Para compactar los nombres de las columnas, establece format.output.envelope=false. Si format.output.envelope es false, format.output.fields debe especificar un solo campo de salida.

Configuración de archivos de Cloud Storage

La siguiente configuración controla cómo el conector escribe archivos en Cloud Storage.

  • gcs.bucket.name: Obligatorio. Especifica el nombre del bucket de Cloud Storage en el que el conector escribe datos.

  • file.compression.type: Especifica el tipo de compresión para los archivos de salida. Algunos ejemplos son gzip, snappy, zstd y none. El valor predeterminado es none.

  • file.name.prefix: Especifica un prefijo que se agregará al nombre de cada archivo. El valor predeterminado está vacío ("").

  • file.name.template: Especifica una plantilla para el nombre del archivo. El conector admite variables de marcador de posición para la plantilla, como {{topic}} para el nombre del tema y {{partition}} para el número de partición. Para obtener más información, consulta la documentación del conector de receptor de Cloud Storage.

  • file.max.records: Es la cantidad máxima de registros que se escribirán en un solo archivo. Si el valor es 0, no hay un máximo específico. El valor predeterminado es 0.

  • file.max.bytes: Es el tamaño máximo del archivo que se escribirá, en bytes. Si el valor es 0, no hay un máximo específico. El valor predeterminado es 0.

Agrupación de registros

Cuando el conector receptor de Cloud Storage escribe registros de Kafka, de forma predeterminada, los agrupa por tema y partición. Por ejemplo, los registros de la partición 0 del tema my-topic se escriben en archivos llamados my-topic-0-OFFSET, en los que OFFSET es el desplazamiento inicial.

Si el parámetro file.name.template incluye la variable de marcador de posición {{key}}, el conector agrupa los registros por clave. En este modo, el conector escribe un archivo independiente para cada clave de registro. El conector reemplaza cada archivo con el registro más reciente de esa clave.

Credenciales

La propiedad gcs.credentials.default debe establecerse en true. Este parámetro de configuración permite que el conector descubra las credenciales de Google Cloud automáticamente.

Próximos pasos

Apache Kafka® es una marca registrada de The Apache Software Foundation o sus afiliados en Estados Unidos y otros países.