Modello Apache Kafka a Cloud Storage

Il modello da Apache Kafka a Cloud Storage è una pipeline di streaming che acquisisce dati di testo da Google Cloud Managed Service per Apache Kafka e restituisce i record a Cloud Storage.

Puoi utilizzare il modello da Apache Kafka a Cloud Storage anche con Kafka autogestito o esterno.

Requisiti della pipeline

• Il bucket Cloud Storage di output deve esistere.
• Il server broker Apache Kafka deve essere in esecuzione ed essere raggiungibile dalle macchine worker di Dataflow.
• Gli argomenti Apache Kafka devono esistere.

Formato dei messaggi Kafka

Questo modello supporta la lettura dei messaggi da Kafka nei seguenti formati:

• JSON
• Codifica binaria Avro
• Avro codificato con Confluent Schema Registry

Formato JSON

Per leggere i messaggi JSON, imposta il parametro del modello messageFormat su "JSON".

Codifica binaria Avro

Per leggere i messaggi Avro binari, imposta i seguenti parametri del modello:

• messageFormat: "AVRO_BINARY_ENCODING".
• binaryAvroSchemaPath: la posizione di un file di schema Avro in Cloud Storage. Esempio: gs://BUCKET_NAME/message-schema.avsc.

Per saperne di più sul formato binario Avro, consulta Codifica binaria nella documentazione di Apache Avro.

Avro codificato con Confluent Schema Registry

Per leggere i messaggi in formato Avro codificato con Confluent Schema Registry, imposta i seguenti parametri del modello:

• messageFormat: "AVRO_CONFLUENT_WIRE_FORMAT".

• schemaFormat: uno dei seguenti valori:
  • "SINGLE_SCHEMA_FILE": lo schema del messaggio è definito in un file di schema Avro. Specifica la posizione del file di schema in Cloud Storage nel confluentAvroSchemaPath parametro.
  • "SCHEMA_REGISTRY": i messaggi vengono codificati utilizzando Confluent Schema Registry. Specifica l'URL dell'istanza di Confluent Schema Registry nel schemaRegistryConnectionUrl parametro e la modalità di autenticazione nel schemaRegistryAuthenticationMode parametro.

Per saperne di più su questo formato, consulta Formato di trasmissione nella documentazione di Confluent.

Formato file di output

Il formato del file di output è lo stesso del messaggio Kafka di input. Ad esempio, se selezioni JSON per il formato dei messaggi Kafka, i file JSON vengono scritti nel bucket Cloud Storage di output.

Autenticazione

Il modello da Apache Kafka a Cloud Storage supporta l'autenticazione SASL/PLAIN per i broker Kafka.

Parametri del modello

Parametri obbligatori

• readBootstrapServerAndTopic: argomento Kafka da cui leggere l'input.
• outputDirectory: il percorso e il prefisso del nome file per la scrittura dei file di output. Deve terminare con una barra. Ad esempio, gs://your-bucket/your-path/.
• kafkaReadAuthenticationMode: la modalità di autenticazione da utilizzare con il cluster Kafka. Utilizza NONE per nessuna autenticazione, SASL_PLAIN per nome utente e password SASL/PLAIN, SASL_SCRAM_512 per l'autenticazione SASL_SCRAM_512 e TLS per l'autenticazione basata su certificato. APPLICATION_DEFAULT_CREDENTIALS deve essere utilizzato solo per il cluster Google Cloud Managed Service per Apache Kafka, in quanto consente l'autenticazione utilizzando le credenziali predefinite dell'applicazione.
• messageFormat: il formato dei messaggi Kafka da leggere. I valori supportati sono AVRO_CONFLUENT_WIRE_FORMAT (Avro codificato con Confluent Schema Registry), AVRO_BINARY_ENCODING (Avro binario semplice) e JSON. Il valore predefinito è AVRO_CONFLUENT_WIRE_FORMAT.
• useBigQueryDLQ: se è true, i messaggi non riusciti verranno scritti in BigQuery con informazioni aggiuntive sugli errori. Il valore predefinito è false.

Parametri facoltativi

• windowDuration: la durata/dimensione della finestra in cui i dati verranno scritti in Cloud Storage. I formati consentiti sono: Ns (per i secondi, ad esempio 5s), Nm (per i minuti, ad esempio 12m), Nh (per le ore, ad esempio 2h). Ad esempio, 5m. Il valore predefinito è 5m.
• outputFilenamePrefix: il prefisso da inserire in ogni file con finestra. Ad esempio, output-. Il valore predefinito è output.
• numShards: il numero massimo di shard di output prodotti durante la scrittura. Un numero maggiore di shard significa una velocità effettiva più elevata per la scrittura in Cloud Storage, ma potenzialmente un costo di aggregazione dei dati più elevato tra gli shard durante l'elaborazione dei file di output di Cloud Storage. Il valore predefinito viene deciso da Dataflow.
• enableCommitOffsets: esegui il commit degli offset dei messaggi elaborati in Kafka. Se abilitata, questa opzione riduce al minimo le lacune o l'elaborazione duplicata dei messaggi durante il riavvio della pipeline. Richiede la specifica dell'ID gruppo di consumer. Il valore predefinito è false.
• consumerGroupId: l'identificatore univoco del gruppo di consumer a cui appartiene questa pipeline. Obbligatorio se l'opzione Esegui il commit degli offset in Kafka è abilitata. Il valore predefinito è vuoto.
• kafkaReadOffset: il punto di partenza per la lettura dei messaggi quando non esistono offset di cui è stato eseguito il commit. Il più recente è quello iniziale, mentre il meno recente viene calcolato partendo dal messaggio più recente. Il valore predefinito è latest.
• kafkaReadUsernameSecretId: l'ID secret di Google Cloud Secret Manager che contiene il nome utente Kafka da utilizzare con l'autenticazione SASL_PLAIN. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. Il valore predefinito è vuoto.
• kafkaReadPasswordSecretId: l'ID secret di Google Cloud Secret Manager che contiene la password Kafka da utilizzare con l'autenticazione SASL_PLAIN. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. Il valore predefinito è vuoto.
• kafkaReadKeystoreLocation: il percorso di Google Cloud Storage del file Java KeyStore (JKS) che contiene il certificato TLS e la chiave privata da utilizzare per l'autenticazione con il cluster Kafka. Ad esempio, gs://your-bucket/keystore.jks.
• kafkaReadTruststoreLocation: il percorso di Google Cloud Storage del file Java TrustStore (JKS) che contiene i certificati attendibili da utilizzare per verificare l'identità del broker Kafka.
• kafkaReadTruststorePasswordSecretId: l'ID secret di Google Cloud Secret Manager che contiene la password da utilizzare per accedere al file Java TrustStore (JKS) per l'autenticazione TLS di Kafka. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeystorePasswordSecretId: l'ID secret di Google Cloud Secret Manager che contiene la password da utilizzare per accedere al file Java KeyStore (JKS) per l'autenticazione TLS di Kafka. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeyPasswordSecretId: l'ID secret di Google Cloud Secret Manager che contiene la password da utilizzare per accedere alla chiave privata all'interno del file Java KeyStore (JKS) per l'autenticazione TLS di Kafka. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramUsernameSecretId: l'ID secret di Google Cloud Secret Manager che contiene il nome utente Kafka da utilizzare con l'autenticazione SASL_SCRAM. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramPasswordSecretId: l'ID secret di Google Cloud Secret Manager che contiene la password Kafka da utilizzare con l'autenticazione SASL_SCRAM. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramTruststoreLocation: il percorso di Google Cloud Storage del file Java TrustStore (JKS) che contiene i certificati attendibili da utilizzare per verificare l'identità del broker Kafka.
• kafkaReadSaslScramTruststorePasswordSecretId: l'ID secret di Google Cloud Secret Manager che contiene la password da utilizzare per accedere al file Java TrustStore (JKS) per l'autenticazione SASL_SCRAM di Kafka. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• schemaFormat: il formato dello schema Kafka. Può essere fornito come SINGLE_SCHEMA_FILE o SCHEMA_REGISTRY. Se viene specificato SINGLE_SCHEMA_FILE, utilizza lo schema menzionato nel file di schema Avro per tutti i messaggi. Se viene specificato SCHEMA_REGISTRY, i messaggi possono avere uno o più schemi. Il valore predefinito è SINGLE_SCHEMA_FILE.
• confluentAvroSchemaPath: il percorso di Google Cloud Storage del singolo file di schema Avro utilizzato per decodificare tutti i messaggi in un argomento. Il valore predefinito è vuoto.
• schemaRegistryConnectionUrl: l'URL dell'istanza di Confluent Schema Registry utilizzata per gestire gli schemi Avro per la decodifica dei messaggi. Il valore predefinito è vuoto.
• binaryAvroSchemaPath: il percorso di Google Cloud Storage del file di schema Avro utilizzato per decodificare i messaggi Avro con codifica binaria. Il valore predefinito è vuoto.
• schemaRegistryAuthenticationMode: la modalità di autenticazione da utilizzare con il registro di schema. Utilizza NONE per nessuna autenticazione, TLS per l'autenticazione basata su certificato e OAUTH per l'autenticazione OAuth2. APPLICATION_DEFAULT_CREDENTIALS deve essere utilizzato solo per il registro di schema Google Cloud Managed Service per Apache Kafka, in quanto consente l'autenticazione utilizzando le credenziali predefinite dell'applicazione.
• schemaRegistryTruststoreLocation: la posizione del certificato SSL in cui sono archiviati gli archivi attendibili per l'autenticazione al registro di schema. Ad esempio, /your-bucket/truststore.jks.
• schemaRegistryTruststorePasswordSecretId: l'ID secret in Secret Manager in cui è archiviata la password per accedere al secret nell'archivio attendibile. Ad esempio, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryKeystoreLocation: la posizione dell'archivio chiavi che contiene il certificato SSL e la chiave privata. Ad esempio, /your-bucket/keystore.jks.
• schemaRegistryKeystorePasswordSecretId: l'ID secret in Secret Manager in cui è archiviata la password per accedere al file dell'archivio chiavi. Ad esempio, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryKeyPasswordSecretId: l'ID secret della password richiesta per accedere alla chiave privata del client archiviata nell'archivio chiavi. Ad esempio, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryOauthClientId: l'ID client utilizzato per autenticare il client del registro di schema in modalità OAUTH. Obbligatorio per il formato dei messaggi AVRO_CONFLUENT_WIRE_FORMAT.
• schemaRegistryOauthClientSecretId: l'ID secret di Google Cloud Secret Manager che contiene il client secret da utilizzare per autenticare il client del registro di schema in modalità OAUTH. Obbligatorio per il formato dei messaggi AVRO_CONFLUENT_WIRE_FORMAT. Ad esempio, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• schemaRegistryOauthScope: l'ambito del token di accesso utilizzato per autenticare il client del registro di schema in modalità OAUTH. Questo campo è facoltativo, in quanto la richiesta può essere effettuata senza un parametro di ambito passato. Ad esempio, openid.
• schemaRegistryOauthTokenEndpointUrl: l'URL basato su HTTP(S) del provider di identità OAuth/OIDC utilizzato per autenticare il client del registro di schema in modalità OAUTH. Obbligatorio per il formato dei messaggi AVRO_CONFLUENT_WIRE_FORMAT.
• outputDeadletterTable: il nome della tabella BigQuery completamente qualificato per i messaggi non riusciti. I messaggi che non sono riusciti a raggiungere la tabella di output per diversi motivi (ad es. schema non corrispondente, JSON non valido) vengono scritti in questa tabella. La tabella verrà creata dal modello. Ad esempio, your-project-id:your-dataset.your-table-name.

Esegui il modello

Console

1. Vai alla pagina Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome job univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

4. Nel menu a discesa Modello Dataflow, seleziona il modello Da Kafka a Cloud Storage.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. (Facoltativo) Per passare dall'elaborazione exactly-once alla modalità di streaming at-least-once, seleziona Almeno una volta.
7. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire flex i modelli, devi avere la versione 284.0.0 o successiva di Google Cloud CLI.

Nella shell o nel terminale, esegui il modello:

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
  

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto Google Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• BOOTSTRAP_SERVER_AND_TOPIC: l'indirizzo del server bootstrap e l'argomento Apache Kafka

  Il formato dell'indirizzo del server bootstrap e dell'argomento dipende dal tipo di cluster:

  • Cluster Managed Service per Apache Kafka: projects/PROJECT_ID/locations/REGION_NAME/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • Cluster Kafka esterno: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME
• STORAGE_BUCKET_NAME: il bucket Cloud Storage in cui viene scritto l'output

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. Per saperne di più sull' API e sui relativi ambiti di autorizzazione, 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",
   }
}
  

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto Google Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• BOOTSTRAP_SERVER_AND_TOPIC: l'indirizzo del server bootstrap e l'argomento Apache Kafka

  Il formato dell'indirizzo del server bootstrap e dell'argomento dipende dal tipo di cluster:

  • Cluster Managed Service per Apache Kafka: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • Cluster Kafka esterno: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME
• STORAGE_BUCKET_NAME: il bucket Cloud Storage in cui viene scritto l'output

Codice sorgente del modello

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

Passaggi successivi

• Scopri di più sui modelli Dataflow.
• Consulta l'elenco dei modelli forniti da Google.