Plantilla de Apache Kafka a Cloud Storage

La plantilla de Apache Kafka a Cloud Storage es una canalización de transmisión que transfiere datos de texto de Google Cloud Managed Service para Apache Kafka y envía los registros a Cloud Storage.

También puedes usar la plantilla de Apache Kafka a BigQuery con Kafka externa o autoadministrada.

Requisitos de la canalización

El bucket de salida de Cloud Storage debe existir.
El servidor del agente de Apache Kafka debe estar en ejecución y se debe poder acceder a él desde las máquinas de trabajador de Dataflow.
Los temas de Apache Kafka deben existir.

Formato del mensaje de Kafka

La plantilla de Apache Kafka a Cloud Storage admite la lectura de mensajes de Kafka en los siguientes formatos: CONFLUENT_AVRO_WIRE_FORMAT y JSON.

Formato de archivos de salida

El formato de archivo de salida tiene el mismo formato que el mensaje de entrada de Kafka. Por ejemplo, si eliges JSON para el formato de mensaje de Kafka, los archivos JSON se escriben en el bucket de salida de Cloud Storage.

Autenticación

La plantilla de Apache Kafka a Cloud Storage admite la autenticación SASL/PLAIN para los agentes de Kafka.

Parámetros de la plantilla

Parámetros obligatorios

readBootstrapServerAndTopic: El tema de Kafka desde el que se lee la entrada.
kafkaReadAuthenticationMode : El modo de autenticación que se usará con el clúster de Kafka. Usa NONE para la no autenticación y SASL_PLAIN para nombre de usuario y contraseña de SASL/PLAIN. Apache Kafka para BigQuery solo es compatible con el modo de autenticación SASL_PLAIN. La configuración predeterminada es: SASL_PLAIN.
outputDirectory: La ruta de acceso y el prefijo del nombre de archivo para escribir los archivos de salida. Debe terminar con una barra. (Ejemplo: gs://your-bucket/your-path/).
messageFormat : El formato de los mensajes de Kafka que se leerán. Los valores admitidos son AVRO_CONFLUENT_WIRE_FORMAT (Avro codificado del Registro de esquema de Confluent), AVRO_BINARY_ENCODING (Avro binario sin formato) y JSON. La configuración predeterminada es: AVRO_CONFLUENT_WIRE_FORMAT.

Parámetros opcionales

windowDuration: La duración/tamaño de la ventana en la que se escribirán los datos en Cloud Storage. Los formatos permitidos son: Ns (para los segundos, por ejemplo, 5 s), Nm (para los minutos, por ejemplo, 12 m) y Nh (para las horas, por ejemplo, 2 h). (Ejemplo: 5 m). La configuración predeterminada es de 5 m.
outputFilenamePrefix: El prefijo para colocar en cada archivo con ventanas. (Ejemplo: output-). La configuración predeterminada es “output”.
numShards: La cantidad máxima de fragmentos de salida que se produce con la escritura. Una mayor cantidad de fragmentos implica una mayor capacidad de procesamiento para la escritura en Cloud Storage, pero, también, un mayor costo de agregación de datos entre fragmentos cuando se procesan archivos de salida de Cloud Storage. Dataflow decide el valor predeterminado.
enableCommitOffsets : confirma desplazamientos de mensajes procesados en Kafka. Si se habilita, esto minimizará las brechas o el procesamiento duplicado de los mensajes cuando se reinicie la canalización. Requiere especificar el ID del grupo de consumidores. La configuración predeterminada es "false".
consumerGroupId : El identificador único del grupo de consumidores al que pertenece esta canalización. Obligatorio si la confirmación de desplazamientos a Kafka está habilitada. La configuración predeterminada es vacía.
kafkaReadOffset : El punto de partida para leer mensajes cuando no existen compensaciones confirmadas. El primero comienza desde el principio y el más reciente desde el mensaje más reciente. La configuración predeterminada es: la más reciente.
kafkaReadUsernameSecretId : El ID del secreto de Google Cloud Secret Manager que contiene el nombre de usuario de Kafka que se usará con la autenticación SASL_PLAIN. (Ejemplo: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). El valor predeterminado es vacío.
kafkaReadPasswordSecretId : El ID del secreto de Google Cloud Secret Manager que contiene la contraseña de Kafka que se usará con la autenticación SASL_PLAIN. (Ejemplo: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). El valor predeterminado es vacío.
schemaFormat : el formato del esquema de Kafka. Se puede proporcionar como SINGLE_SCHEMA_FILE o SCHEMA_REGISTRY. Si se especifica SINGLE_Schema_FILE, todos los mensajes deben tener el esquema mencionado en el archivo de esquema avro. Si se especifica SCHEMA_REGISTRY, los mensajes pueden tener un solo esquema o varios. La configuración predeterminada es: SINGLE_SCHEMA_FILE.
confluentAvroSchemaPath : la ruta de acceso de Google Cloud Storage al único archivo de esquema de Avro que se usa para decodificar todos los mensajes de un tema. La configuración predeterminada es vacía.
schemaRegistryConnectionUrl : La URL de la instancia del registro de esquemas de Confluent que se usa para administrar esquemas de Avro para la decodificación de mensajes. La configuración predeterminada es vacía.
binaryAvroSchemaPath : Es la ruta de acceso de Google Cloud Storage al archivo de esquema de Avro que se usa para decodificar mensajes de Avro con codificación binaria. La configuración predeterminada es vacía.

Ejecuta la plantilla

Console

Ve a la página Crear un trabajo a partir de una plantilla de Dataflow.

Ir a Crear un trabajo a partir de una plantilla

En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
Opcional: Para Extremo regional, selecciona un valor del menú desplegable. La región predeterminada es us-central1.
Para obtener una lista de regiones en las que puedes ejecutar un trabajo de Dataflow, consulta Ubicaciones de Dataflow.
En el menú desplegable Plantilla de Dataflow, selecciona the Kafka to Cloud Storage template.
En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
Opcional: Para cambiar del procesamiento “exactamente una vez” al modo de transmisión al menos una vez, selecciona Al menos una vez.
Haz clic en Ejecutar trabajo.

gcloud

En tu shell o terminal, ejecuta la plantilla:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Kafka_to_Cloud Storage \
    --parameters \
outputTableSpec=BIGQUERY_TABLE,\
inputTopics=KAFKA_TOPICS,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
bootstrapServers=KAFKA_SERVER_ADDRESSES

Reemplaza lo siguiente:

PROJECT_ID: El ID del proyecto de Google Cloud en el que deseas ejecutar el trabajo de Dataflow.
JOB_NAME: Es el nombre del trabajo que elijas
REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
VERSION: Es la versión de la plantilla que deseas usar.
Puedes usar los siguientes valores:
- latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
- el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
BIGQUERY_TABLE: Es el nombre de la tabla de Cloud Storage
KAFKA_TOPICS: Es la lista de temas de Apache Kkafa. Si se proporcionan varios temas, debes escapar las comas. Consulta gcloud topic escaping.
PATH_TO_JAVASCRIPT_UDF_FILE: El URI de Cloud Storage de .js archivo que define la función definida por el usuario (UDF) de JavaScript que deseas usar, por ejemplo:gs://my-bucket/my-udfs/my_file.js
JAVASCRIPT_FUNCTION es el nombre de la función definida por el usuario (UDF) de JavaScript que deseas usar.
Por ejemplo, si el código de tu función de JavaScript es myTransform(inJson) { /*...do stuff...*/ }, el nombre de la función es myTransform. Para ver ejemplos de UDF de JavaScript, consulta Ejemplos de UDF.
KAFKA_SERVER_ADDRESSES: La lista de direcciones IP del servidor del agente de Apache Kafka. Cada dirección IP debe tener el número de puerto desde el que se puede acceder al servidor. Por ejemplo: 35.70.252.199:9092. Si se proporcionan varias direcciones, debes escapar las comas. Consulta gcloud topic escaping.

API

Para ejecutar la plantilla con la API de REST, envía una solicitud POST HTTP. Para obtener más información de la API y sus permisos de autorización, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "outputTableSpec": "BIGQUERY_TABLE",
          "inputTopics": "KAFKA_TOPICS",
          "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
          "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
          "bootstrapServers": "KAFKA_SERVER_ADDRESSES"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Kafka_to_Cloud Storage",
   }
}

Reemplaza lo siguiente:

PROJECT_ID: El ID del proyecto de Google Cloud en el que deseas ejecutar el trabajo de Dataflow.
JOB_NAME: Es el nombre del trabajo que elijas
LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
VERSION: Es la versión de la plantilla que deseas usar.
Puedes usar los siguientes valores:
- latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
- el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
BIGQUERY_TABLE: Es el nombre de la tabla de Cloud Storage
KAFKA_TOPICS: Es la lista de temas de Apache Kkafa. Si se proporcionan varios temas, debes escapar las comas. Consulta gcloud topic escaping.
PATH_TO_JAVASCRIPT_UDF_FILE: El URI de Cloud Storage de .js archivo que define la función definida por el usuario (UDF) de JavaScript que deseas usar, por ejemplo:gs://my-bucket/my-udfs/my_file.js
JAVASCRIPT_FUNCTION es el nombre de la función definida por el usuario (UDF) de JavaScript que deseas usar.
Por ejemplo, si el código de tu función de JavaScript es myTransform(inJson) { /*...do stuff...*/ }, el nombre de la función es myTransform. Para ver ejemplos de UDF de JavaScript, consulta Ejemplos de UDF.
KAFKA_SERVER_ADDRESSES: La lista de direcciones IP del servidor del agente de Apache Kafka. Cada dirección IP debe tener el número de puerto desde el que se puede acceder al servidor. Por ejemplo: 35.70.252.199:9092. Si se proporcionan varias direcciones, debes escapar las comas. Consulta gcloud topic escaping.

Para obtener más información, consulta Escribe datos de Kafka en Cloud Storage con Dataflow.

Código fuente de la plantilla

Java

/*
 * Copyright (C) 2024 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.v2.kafka.dlq.BigQueryDeadLetterQueue;
import com.google.cloud.teleport.v2.kafka.dlq.BigQueryDeadLetterQueueOptions;
import com.google.cloud.teleport.v2.kafka.options.KafkaReadOptions;
import com.google.cloud.teleport.v2.kafka.options.SchemaRegistryOptions;
import com.google.cloud.teleport.v2.kafka.transforms.KafkaTransform;
import com.google.cloud.teleport.v2.kafka.utils.KafkaConfig;
import com.google.cloud.teleport.v2.kafka.utils.KafkaTopicUtils;
import com.google.cloud.teleport.v2.transforms.WriteTransform;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.io.kafka.KafkaRecord;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.errorhandling.BadRecord;
import org.apache.beam.sdk.transforms.errorhandling.BadRecordRouter;
import org.apache.beam.sdk.transforms.errorhandling.ErrorHandler;
import org.apache.beam.sdk.values.PCollection;

@Template(
    name = "Kafka_to_Gcs_Flex",
    category = TemplateCategory.STREAMING,
    displayName = "Kafka to Cloud Storage",
    description =
        "A streaming pipeline which ingests data from Kafka and writes to a pre-existing Cloud"
            + " Storage bucket with a variety of file types.",
    optionsClass = KafkaToGcsFlex.KafkaToGcsOptions.class,
    flexContainerName = "kafka-to-gcs-flex",
    contactInformation = "https://cloud.google.com/support",
    requirements = {"The output Google Cloud Storage directory must exist."})
public class KafkaToGcsFlex {

  public interface KafkaToGcsOptions
      extends PipelineOptions,
          DataflowPipelineOptions,
          KafkaReadOptions,
          SchemaRegistryOptions,
          BigQueryDeadLetterQueueOptions {

    // This is a duplicate option that already exist in KafkaReadOptions but keeping it here
    // so the KafkaTopic appears above the authentication enum on the Templates UI.
    @TemplateParameter.KafkaReadTopic(
        order = 1,
        name = "readBootstrapServerAndTopic",
        groupName = "Source",
        description = "Source Kafka Topic",
        helpText = "Kafka Topic to read the input from.")
    String getReadBootstrapServerAndTopic();

    void setReadBootstrapServerAndTopic(String value);

    @TemplateParameter.Duration(
        order = 20,
        optional = true,
        groupName = "Destination",
        description = "Window duration",
        helpText =
            "The window duration/size in which data will be written to Cloud Storage. Allowed formats are: Ns (for "
                + "seconds, example: 5s), Nm (for minutes, example: 12m), Nh (for hours, example: 2h).",
        example = "5m")
    @Default.String("5m")
    String getWindowDuration();

    void setWindowDuration(String windowDuration);

    @TemplateParameter.GcsWriteFolder(
        order = 21,
        groupName = "Destination",
        description = "Output file directory in Cloud Storage",
        helpText = "The path and filename prefix for writing output files. Must end with a slash.",
        example = "gs://your-bucket/your-path/")
    String getOutputDirectory();

    void setOutputDirectory(String outputDirectory);

    @TemplateParameter.Text(
        order = 22,
        optional = true,
        groupName = "Destination",
        description = "Output filename prefix of the files to write",
        helpText = "The prefix to place on each windowed file.",
        example = "output-")
    @Default.String("output")
    String getOutputFilenamePrefix();

    void setOutputFilenamePrefix(String outputFilenamePrefix);

    @TemplateParameter.Integer(
        order = 23,
        optional = true,
        description = "Maximum output shards",
        groupName = "Destination",
        helpText =
            "The maximum number of output shards produced when writing. A higher number of "
                + "shards means higher throughput for writing to Cloud Storage, but potentially higher "
                + "data aggregation cost across shards when processing output Cloud Storage files. "
                + "Default value is decided by Dataflow.")
    @Default.Integer(0)
    Integer getNumShards();

    void setNumShards(Integer numShards);
  }

  public static PipelineResult run(KafkaToGcsOptions options) throws Exception {
    // Create the Pipeline
    Pipeline pipeline = Pipeline.create(options);
    String bootstrapServes;
    List<String> topicsList;
    if (options.getReadBootstrapServerAndTopic() != null) {
      List<String> bootstrapServerAndTopicList =
          KafkaTopicUtils.getBootstrapServerAndTopic(
              options.getReadBootstrapServerAndTopic(), options.getProject());
      topicsList = List.of(bootstrapServerAndTopicList.get(1));
      bootstrapServes = bootstrapServerAndTopicList.get(0);
    } else {
      throw new IllegalArgumentException(
          "Please provide a valid bootstrap server which matches `[,:a-zA-Z0-9._-]+` and a topic which matches `[,a-zA-Z0-9._-]+`");
    }

    options.setStreaming(true);

    Map<String, Object> kafkaConfig = new HashMap<>(KafkaConfig.fromReadOptions(options));

    // Configure dead letter queue params
    ErrorHandler<BadRecord, ?> errorHandler = new ErrorHandler.DefaultErrorHandler<>();
    // Throwing Router throws the error instead of sending it to the DLQ. This will be the case
    // when no DLQ is configured and the pipeline will retry the failed error.
    BadRecordRouter badRecordRouter = BadRecordRouter.THROWING_ROUTER;

    if (options.getUseBigQueryDLQ()) {
      if (options.getOutputDeadletterTable() == null
          || options.getOutputDeadletterTable().isBlank()) {
        throw new IllegalArgumentException(
            "Please provide a Fully Qualified BigQuery table name when BigQuery Dead Letter"
                + "Queue is enabled");
      }
      badRecordRouter = BadRecordRouter.RECORDING_ROUTER;
      errorHandler =
          pipeline.registerBadRecordErrorHandler(
              BigQueryDeadLetterQueue.newBuilder()
                  .setTableName(options.getOutputDeadletterTable())
                  .build());
    }

    PCollection<KafkaRecord<byte[], byte[]>> kafkaRecord;
    // Step 1: Read from Kafka as bytes.
    KafkaIO.Read<byte[], byte[]> kafkaTransform =
        KafkaTransform.readBytesFromKafka(
            bootstrapServes, topicsList, kafkaConfig, options.getEnableCommitOffsets());
    kafkaRecord = pipeline.apply(kafkaTransform);

    kafkaRecord.apply(
        WriteTransform.newBuilder()
            .setOptions(options)
            .setBadRecordErrorHandler(errorHandler)
            .setBadRecordRouter(badRecordRouter)
            .build());
    if (options.getUseBigQueryDLQ()) {
      errorHandler.close();
    }
    return pipeline.run();
  }

  public static void main(String[] args) throws Exception {
    KafkaToGcsOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(KafkaToGcsOptions.class);

    run(options);
  }
}

¿Qué sigue?

Obtén información sobre las plantillas de Dataflow.
Consulta la lista de plantillas que proporciona Google.