Crea un conector de receptor de Cloud Storage

Los conectores de receptor de Cloud Storage transmiten datos de Kafka a buckets de Cloud Storage, lo que te permite almacenar y procesar grandes volúmenes de datos de manera rentable y escalable.

Los casos de uso de los conectores de receptor de Cloud Storage incluyen los siguientes:

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

  • Archiva 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 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:

  • Crear un conector: managedkafka.connectors.create

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

Otorga permisos para escribir en el bucket de Cloud Storage

La cuenta de servicio de Managed Kafka debe tener los siguientes permisos para escribir mensajes en el bucket de Cloud Storage:

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

Otorga el rol de Usuario de objetos de almacenamiento (roles/storage.objectUser) a la cuenta de servicio en el proyecto que contiene el bucket de Cloud Storage.

La cuenta de servicio de Managed Kafka tiene el siguiente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, donde PROJECT_NUMBER es el número de proyecto del clúster de Connect.

Si tu clúster de Connect está en un proyecto diferente de tu clúster de Managed Service para Apache Kafka cluster, 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 Google Cloud consola de, ve a la página Clústeres de Connect.

    Ir a los clústeres de Connect

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

    Se muestra la página Detalles del clúster de Connect.

  3. Haz clic en Crear conector.

    Se muestra 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 Connector plugin (Plug-in del conector), selecciona Cloud Storage Sink.

  6. Especifica los Topics (Temas) desde los que puedes transmitir datos.

  7. Elige el Storage Bucket (Bucket de almacenamiento) para almacenar los datos.

  8. (Opcional) Configura parámetros adicionales en la sección Configuration (Configuración).

  9. Selecciona la Task restart policy (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 Google Cloud consola de, activa Cloud Shell.

    Activa Cloud Shell

    En la parte inferior de la Google Cloud consola de 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 gcloud managed-kafka connectors create comando:

    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 debe 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 este ejemplo, sigue las instrucciones de configuración para 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 este ejemplo, sigue las instrucciones de configuración para 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 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 receptor de Cloud Storage.

Output fields

De forma predeterminada, el conector escribe solo el valor del registro de cada registro de Kafka. Para incluir claves o encabezados de registro, establece la propiedad format.output.fields en una lista separada por comas de los campos que se incluirán. Los campos admitidos 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, establece format.output.type=csv. Usa los siguientes convertidores:

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

  • Convertidor 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 estableces el convertidor de claves o valores en org.apache.kafka.connect.converters.ByteArrayConverter, el conector escribirá ese campo como un valor codificado en base64.

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

Formato Parquet

Para escribir archivos Parquet en Cloud Storage, establece 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 convertidor 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 aplanar los nombres de las columnas, establece format.output.envelope=false. Si format.output.envelope es false, entonces 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 de 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 de receptor de Cloud Storage escribe registros de Kafka, de forma predeterminada, agrupa los registros 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 separado para cada clave de registro. El conector sobrescribe 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 Google Cloud las credenciales automáticamente.

Próximos pasos

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