Modello da Apache Kafka a Cloud Storage

Il modello da Apache Kafka a Cloud Storage è una pipeline di inserimento flussi che importa dati di testo da Apache Kafka per BigQuery e invia i record a Cloud Storage.

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

Requisiti della pipeline

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

Formato dei messaggi Kafka

Il modello da Apache Kafka a Cloud Storage supporta la lettura di messaggi da Kafka nei seguenti formati: CONFLUENT_AVRO_WIRE_FORMAT e JSON.

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.
• kafkaReadAuthenticationMode : la modalità di autenticazione da utilizzare con il cluster Kafka. Utilizza NONE per nessuna autenticazione e SASL_PLAIN per nome utente e password SASL/PLAIN. Apache Kafka per BigQuery supporta solo la modalità di autenticazione SASL_PLAIN. Il valore predefinito è: SASL_PLAIN.
• 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/.
• messageFormat : il formato dei messaggi Kafka da leggere. I valori supportati sono AVRO_CONFLUENT_WIRE_FORMAT (Avro codificato da Confluent Schema Registry), AVRO_BINARY_ENCODING (Avro binario normale) e JSON. Il valore predefinito è: AVRO_CONFLUENT_WIRE_FORMAT.

Parametri facoltativi

• windowDuration : durata/dimensione della finestra in cui i dati verranno scritti in Cloud Storage. I formati consentiti sono: Ns (per i secondi, ad esempio 5 s), Nm (per i minuti, ad esempio 12 m), Nh (per le ore, ad esempio 2 h). (Esempio: 5 m). Il valore predefinito è 5 min.
• outputFilenamePrefix : il prefisso da inserire in ogni file finestrato. (Esempio: output-). Il valore predefinito è: output.
• numShards : il numero massimo di shard di output prodotti durante la scrittura. Un numero più elevato di shard significa una maggiore velocità effettiva per la scrittura in Cloud Storage, ma un costo di aggregazione dei dati potenzialmente superiore negli shard durante l'elaborazione dei file Cloud Storage di output. Il valore predefinito viene deciso da Dataflow.
• enableCommitOffsets : esegue il commit degli offset dei messaggi elaborati in Kafka. Se abilitata, questa opzione ridurrà al minimo le lacune o l'elaborazione duplicata dei messaggi quando riavvii la pipeline. Richiede la specifica dell'ID gruppo di consumatori. Il valore predefinito è false.
• consumerGroupId : l'identificatore univoco del gruppo di consumatori a cui appartiene questa pipeline. Obbligatorio se è abilitato il Commit Offsets in Kafka. Il campo predefinito è vuoto.
• kafkaReadOffset : il punto di partenza per la lettura dei messaggi quando non esistono offset di commit. La prima parte dell'email parte dall'inizio, la più recente dall'ultimo messaggio. Il valore predefinito è: più recente.
• kafkaReadUsernameSecretId : l'ID secret di Google Cloud Secret Manager che contiene il nome utente Kafka da utilizzare con l'autenticazione SASL_PLAIN. (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. (Esempio: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). Il valore predefinito è vuoto.
• schemaFormat : il formato dello schema Kafka. Può essere fornito nel formato SINGLE_SCHEMA_FILE o SCHEMA_REGISTRY. Se viene specificato SINGLE_SCHEMA_FILE, tutti i messaggi devono avere lo schema indicato nel file di schema avro. Se viene specificato SCHEMA_REGISTRY, i messaggi possono avere un solo schema 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 campo predefinito è vuoto.
• schemaRegistryConnectionUrl : l'URL dell'istanza del Confluent Schema Registry utilizzato per gestire gli schemi Avro per la decodifica dei messaggi. Il campo predefinito è vuoto.
• binaryAvroSchemaPath : il percorso di Google Cloud Storage del file di schema Avro utilizzato per decodificare i messaggi Avro con codifica binaria. Il campo predefinito è vuoto.

Esegui il modello

Console

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

   Per un elenco di regioni in cui è possibile eseguire un job Dataflow, consulta Località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Kafka to Cloud Storage template.
5. Inserisci i valori parametro negli appositi campi.
6. (Facoltativo) Per passare dall'elaborazione "exactly-once" alla modalità flusso di dati Almeno una volta, seleziona Almeno una volta.
7. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire flex modelli, devi disporre di Google Cloud CLI 284.0.0 o successiva.

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_Cloud Storage \
    --parameters \
outputTableSpec=BIGQUERY_TABLE,\
inputTopics=KAFKA_TOPICS,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
bootstrapServers=KAFKA_SERVER_ADDRESSES
  

Sostituisci quanto segue:

• PROJECT_ID: l'ID del 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 padre non con data del 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 è possibile trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione:la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che provocano un errore influiscano sui flussi di lavoro di produzione.
• BIGQUERY_TABLE: il nome della tua tabella Cloud Storage
• KAFKA_TOPICS: elenco degli argomenti di Apache Kakfa. Se vengono forniti più argomenti, le virgole devono essere precedute dal carattere di escape. Vedi gcloud topic escaping.
• PATH_TO_JAVASCRIPT_UDF_FILE: L'URI Cloud Storage del file .js che definisce la funzione JavaScript definita dall'utente che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• JAVASCRIPT_FUNCTION: il nome della funzione definita dall'utente (UDF) JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di funzioni JavaScript definite dall'utente, consulta gli esempi di funzioni definite dall'utente.

• KAFKA_SERVER_ADDRESSES: l'elenco di indirizzi IP del server broker Apache Kafka. Ogni indirizzo IP deve contenere il numero di porta da cui è accessibile il server. Ad esempio: 35.70.252.199:9092. Se vengono forniti più indirizzi, devi utilizzare il carattere di escape per le virgole. Vedi gcloud topic escaping.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni 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": {
          "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",
   }
}
  

Sostituisci quanto segue:

• PROJECT_ID: l'ID del 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 padre non con data del 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 è possibile trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione:la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare i modelli conservati nella cartella principale con data più recente per evitare che queste modifiche che provocano un errore influiscano sui flussi di lavoro di produzione.
• BIGQUERY_TABLE: il nome della tua tabella Cloud Storage
• KAFKA_TOPICS: elenco degli argomenti di Apache Kakfa. Se vengono forniti più argomenti, le virgole devono essere precedute dal carattere di escape. Vedi gcloud topic escaping.
• PATH_TO_JAVASCRIPT_UDF_FILE: L'URI Cloud Storage del file .js che definisce la funzione JavaScript definita dall'utente che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• JAVASCRIPT_FUNCTION: il nome della funzione definita dall'utente (UDF) JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di funzioni JavaScript definite dall'utente, consulta gli esempi di funzioni definite dall'utente.

• KAFKA_SERVER_ADDRESSES: l'elenco di indirizzi IP del server broker Apache Kafka. Ogni indirizzo IP deve contenere il numero di porta da cui è accessibile il server. Ad esempio: 35.70.252.199:9092. Se vengono forniti più indirizzi, devi utilizzare il carattere di escape per le virgole. Vedi gcloud topic escaping.

Per ulteriori informazioni, consulta Scrivere dati da Kafka a Cloud Storage con Dataflow.

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.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.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 {

    // 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.KafkaTopic(
        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 UnsupportedOperationException {

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

    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).build());
    return pipeline.run();
  }

  public static void main(String[] args) {
    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.