Plantilla de Apache Kafka a Cloud Storage

La plantilla de Apache Kafka a Cloud Storage es una canalización de streaming que ingiere datos de texto del servicio gestionado de Google Cloud para Apache Kafka y envía los registros a Cloud Storage.

También puedes usar la plantilla de Apache Kafka a Cloud Storage con Kafka autogestionado o externo.

Requisitos del flujo de procesamiento

• El segmento de Cloud Storage de salida debe existir.
• El servidor de broker de Apache Kafka debe estar en funcionamiento y se debe poder acceder a él desde las máquinas de trabajador de Dataflow.
• Los temas de Apache Kafka deben existir.

Formato de mensaje de Kafka

Esta plantilla admite la lectura de mensajes de Kafka en los siguientes formatos:

• JSON
• Codificación binaria Avro
• Avro codificado con Confluent Schema Registry

Formato JSON

Para leer mensajes JSON, asigna el valor "JSON" al parámetro de plantilla messageFormat.

Codificación binaria Avro

Para leer mensajes Avro binarios, define los siguientes parámetros de plantilla:

• messageFormat: "AVRO_BINARY_ENCODING".
• binaryAvroSchemaPath: la ubicación de un archivo de esquema Avro en Cloud Storage. Ejemplo: gs://BUCKET_NAME/message-schema.avsc

Para obtener más información sobre el formato binario Avro, consulta Codificación binaria en la documentación de Apache Avro.

Avro codificado con Confluent Schema Registry

Para leer mensajes en Avro codificados en Confluent Schema Registry, define los siguientes parámetros de plantilla:

• messageFormat: "AVRO_CONFLUENT_WIRE_FORMAT".

• schemaFormat: puede ser uno de los siguientes valores:
  • "SINGLE_SCHEMA_FILE": el esquema del mensaje se define en un archivo de esquema Avro. Especifica la ubicación del archivo de esquema en Cloud Storage en el parámetro confluentAvroSchemaPath.
  • "SCHEMA_REGISTRY": los mensajes se codifican mediante Confluent Schema Registry. Especifique la URL de la instancia de Confluent Schema Registry en el parámetro schemaRegistryConnectionUrl y el modo de autenticación en el parámetro schemaRegistryAuthenticationMode.

Para obtener más información sobre este formato, consulta Wire format (Formato de cable) en la documentación de Confluent.

Formato de archivo de salida

El formato del archivo de salida es el mismo que el del mensaje de Kafka de entrada. Por ejemplo, si seleccionas JSON como formato de mensaje de Kafka, los archivos JSON se escribirán en el segmento de Cloud Storage de salida.

Autenticación

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

Parámetros de plantilla

Parámetros obligatorios

• readBootstrapServerAndTopic tema de Kafka del que se leerán los datos de entrada.
• outputDirectory prefijo de ruta y de nombre de archivo para escribir archivos de salida. Debe terminar con una barra diagonal. Por ejemplo, gs://your-bucket/your-path/.
• kafkaReadAuthenticationMode el modo de autenticación que se va a usar con el clúster de Kafka. Usa KafkaAuthenticationMethod.NONE para no usar autenticación, KafkaAuthenticationMethod.SASL_PLAIN para usar el nombre de usuario y la contraseña de SASL/PLAIN, KafkaAuthenticationMethod.SASL_SCRAM_512 para usar la autenticación SASL_SCRAM_512 y KafkaAuthenticationMethod.TLS para usar la autenticación basada en certificados. KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS solo se debe usar en el clúster de Google Cloud Apache Kafka para BigQuery. Permite autenticarte mediante las credenciales predeterminadas de la aplicación.
• messageFormat: formato de los mensajes de Kafka que se van a leer. Los valores admitidos son AVRO_CONFLUENT_WIRE_FORMAT (Avro codificado en Confluent Schema Registry), AVRO_BINARY_ENCODING (Avro binario sin formato) y JSON. Valor predeterminado: AVRO_CONFLUENT_WIRE_FORMAT.
• useBigQueryDLQ si es true, los mensajes fallidos se escribirán en BigQuery con información adicional sobre el error. Valor predeterminado: false.

Parámetros opcionales

• windowDuration la duración o el tamaño de la ventana en la que se escribirán los datos en Cloud Storage. Los formatos permitidos son: Ns (para segundos, por ejemplo, 5s), Nm (para minutos, por ejemplo, 12m) y Nh (para horas, por ejemplo, 2h). Por ejemplo, 5m. El valor predeterminado es 5 minutos.
• outputFilenamePrefix el prefijo que se colocará en cada archivo de ventana. Por ejemplo, output-. El valor predeterminado es "output".
• numShards: número máximo de fragmentos de salida que se producen al escribir. Un mayor número de particiones implica un mayor rendimiento para escribir en Cloud Storage, pero puede suponer un mayor coste de agregación de datos en las particiones al procesar los archivos de Cloud Storage de salida. Dataflow decide el valor predeterminado.
• enableCommitOffsets confirma los desplazamientos de los mensajes procesados en Kafka. Si se habilita, se minimizarán los huecos o el procesamiento duplicado de mensajes al reiniciar la canalización. Requiere que se especifique el ID de grupo de consumidores. Valor predeterminado: false.
• consumerGroupId identificador único del grupo de consumidores al que pertenece esta canalización. Obligatorio si la opción Confirmar desfases en Kafka está habilitada. El valor predeterminado es una cadena vacía.
• kafkaReadOffset el punto de partida para leer mensajes cuando no hay offsets confirmados. La más antigua empieza desde el principio y la más reciente, desde el mensaje más reciente. El valor predeterminado es latest.
• kafkaReadUsernameSecretId el ID del secreto de Google Cloud Secret Manager que contiene el nombre de usuario de Kafka que se va a usar con la autenticación SASL_PLAIN. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. El valor predeterminado es una cadena vacía.
• kafkaReadPasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña de Kafka que se va a usar con la autenticación SASL_PLAIN. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. El valor predeterminado es una cadena vacía.
• kafkaReadKeystoreLocation ruta de Google Cloud Storage al archivo Java KeyStore (JKS) que contiene el certificado TLS y la clave privada que se usarán para autenticarte con el clúster de Kafka. Por ejemplo, gs://your-bucket/keystore.jks.
• kafkaReadTruststoreLocation la ruta de Google Cloud Storage al archivo Java TrustStore (JKS) que contiene los certificados de confianza que se usarán para verificar la identidad del broker de Kafka.
• kafkaReadTruststorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java TrustStore (JKS) para la autenticación TLS de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeystorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java KeyStore (JKS) para la autenticación TLS de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeyPasswordSecretId el ID de secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder a la clave privada del archivo Java KeyStore (JKS) para la autenticación TLS de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramUsernameSecretId el ID del secreto de Google Cloud Secret Manager que contiene el nombre de usuario de Kafka que se va a usar con la autenticación SASL_SCRAM. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramPasswordSecretId el ID del secreto de Secret Manager de Google Cloud que contiene la contraseña de Kafka que se va a usar con la autenticación SASL_SCRAM. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramTruststoreLocation la ruta de Google Cloud Storage al archivo Java TrustStore (JKS) que contiene los certificados de confianza que se usarán para verificar la identidad del broker de Kafka.
• kafkaReadSaslScramTruststorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java TrustStore (JKS) para la autenticación SASL_SCRAM de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• schemaFormat: el formato del esquema de Kafka. Se puede proporcionar como SINGLE_SCHEMA_FILE o SCHEMA_REGISTRY. Si se especifica SINGLE_SCHEMA_FILE, usa el esquema mencionado en el archivo de esquema de Avro para todos los mensajes. Si se especifica SCHEMA_REGISTRY, los mensajes pueden tener un solo esquema o varios. Valor predeterminado: SINGLE_SCHEMA_FILE.
• confluentAvroSchemaPath la ruta de Google Cloud Storage al archivo de esquema de Avro único que se usa para decodificar todos los mensajes de un tema. El valor predeterminado es una cadena vacía.
• schemaRegistryConnectionUrl la URL de la instancia de Confluent Schema Registry que se usa para gestionar los esquemas de Avro para decodificar mensajes. El valor predeterminado es una cadena vacía.
• binaryAvroSchemaPath: la ruta de Google Cloud Storage al archivo de esquema de Avro que se usa para decodificar mensajes de Avro codificados en formato binario. El valor predeterminado es una cadena vacía.
• schemaRegistryAuthenticationMode modo de autenticación del registro de esquemas. Puede ser NONE, TLS u OAUTH. El valor predeterminado es NONE.
• schemaRegistryTruststoreLocation ubicación del certificado SSL en el que se almacena el almacén de confianza para la autenticación en Schema Registry. Por ejemplo, /your-bucket/truststore.jks.
• schemaRegistryTruststorePasswordSecretId: SecretId en Secret Manager donde se almacena la contraseña para acceder al secreto en el almacén de confianza. Por ejemplo, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryKeystoreLocation ubicación del almacén de claves que contiene el certificado SSL y la clave privada. Por ejemplo, /your-bucket/keystore.jks.
• schemaRegistryKeystorePasswordSecretId SecretId en Secret Manager donde se encuentra la contraseña para acceder al archivo del almacén de claves. Por ejemplo, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryKeyPasswordSecretId SecretId de la contraseña necesaria para acceder a la clave privada del cliente almacenada en el almacén de claves. Por ejemplo, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryOauthClientId ID de cliente usado para autenticar el cliente de Schema Registry en el modo OAuth. Obligatorio para el formato de mensaje AVRO_CONFLUENT_WIRE_FORMAT.
• schemaRegistryOauthClientSecretId el ID de secreto de Google Cloud Secret Manager que contiene el secreto de cliente que se usará para autenticar el cliente de Schema Registry en el modo OAUTH. Obligatorio para el formato de mensaje AVRO_CONFLUENT_WIRE_FORMAT. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• schemaRegistryOauthScope el ámbito del token de acceso que se usa para autenticar el cliente de Schema Registry en el modo OAUTH. Este campo es opcional, ya que la solicitud se puede hacer sin pasar un parámetro de ámbito. Por ejemplo, openid.
• schemaRegistryOauthTokenEndpointUrl la URL basada en HTTP(S) del proveedor de identidades de OAuth u OIDC que se usa para autenticar el cliente de Schema Registry en el modo OAUTH. Obligatorio para el formato de mensaje AVRO_CONFLUENT_WIRE_FORMAT.
• outputDeadletterTable nombre de tabla de BigQuery completo de los mensajes fallidos. Los mensajes que no se han podido enviar a la tabla de salida por diferentes motivos (por ejemplo, un esquema que no coincide o un JSON con formato incorrecto) se escriben en esta tabla. La plantilla creará la tabla. Por ejemplo, your-project-id:your-dataset.your-table-name.

Ejecutar la plantilla

Consola

1. Ve a la página Crear tarea a partir de plantilla de Dataflow.
Ir a Crear tarea a partir de plantilla
2. En el campo Nombre de la tarea, introduce un nombre único.
3. Opcional: En Endpoint regional, seleccione un valor en el menú desplegable. La región predeterminada es us-central1.

   Para ver una lista de las regiones en las que puedes ejecutar una tarea de Dataflow, consulta Ubicaciones de Dataflow.

4. En el menú desplegable Plantilla de flujo de datos, seleccione the Kafka to Cloud Storage template.
5. En los campos de parámetros proporcionados, introduzca los valores de los parámetros.
6. Opcional: Para cambiar del procesamiento una sola vez al modo de streaming al menos una vez, selecciona Al menos una vez.
7. Haz clic en Ejecutar trabajo.

gcloud

Nota: Para usar la CLI de Google Cloud y ejecutar plantillas de flex, debes tener la versión CLI de Google Cloud 284.0.0 o una posterior.

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_Gcs_Flex \
    --parameters \
readBootstrapServerAndTopic=BOOTSTRAP_SERVER_AND_TOPIC,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
outputDirectory=gs://STORAGE_BUCKET_NAME,\
useBigQueryDLQ=false
  

Haz los cambios siguientes:

• PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
• JOB_NAME: un nombre de trabajo único que elijas
• REGION_NAME: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• VERSION: la versión de la plantilla que quieres usar

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: 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 encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• BOOTSTRAP_SERVER_AND_TOPIC: la dirección del servidor de arranque y el tema de Apache Kafka

  El formato de la dirección del servidor de arranque y del tema depende del tipo de clúster:

  • Clúster de Managed Service para Apache Kafka: projects/PROJECT_ID/locations/REGION_NAME/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • Clúster de Kafka externo: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME
• STORAGE_BUCKET_NAME: el segmento de Cloud Storage en el que se escribe la salida.

API

Para ejecutar la plantilla mediante la API REST, envía una solicitud HTTP POST. Para obtener más información sobre la API y sus ámbitos 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": {
          "readBootstrapServerAndTopic": "BOOTSTRAP_SERVER_AND_TOPIC",
          "kafkaReadAuthenticationMode": "APPLICATION_DEFAULT_CREDENTIALS",
          "messageFormat": "JSON",
          "outputDirectory": "gs://STORAGE_BUCKET_NAME",
          "useBigQueryDLQ": "false"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Kafka_to_Gcs_Flex",
   }
}
  

Haz los cambios siguientes:

• PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
• JOB_NAME: un nombre de trabajo único que elijas
• LOCATION: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• VERSION: la versión de la plantilla que quieres usar

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: 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 encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• BOOTSTRAP_SERVER_AND_TOPIC: la dirección del servidor de arranque y el tema de Apache Kafka

  El formato de la dirección del servidor de arranque y del tema depende del tipo de clúster:

  • Clúster de Managed Service para Apache Kafka: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • Clúster de Kafka externo: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME
• STORAGE_BUCKET_NAME: el segmento de Cloud Storage en el que se escribe la salida.

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

Siguientes pasos

• Consulta información sobre las plantillas de Dataflow.
• Consulta la lista de plantillas proporcionadas por Google.