Vorlage „Apache Kafka für Cloud Storage“

Die Vorlage „Apache Kafka für Cloud Storage“ ist eine Streamingpipeline, die Textdaten aus Google Cloud Managed Service for Apache Kafka aufnimmt und die Datensätze in Cloud Storage ausgibt.

Sie können die Vorlage „Apache Kafka für BigQuery“ auch mit selbstverwalteter oder externer Kafka verwenden.

Pipelineanforderungen

• Der Cloud Storage-Ausgabe-Bucket muss vorhanden sein.
• Der Apache Kafka-Broker-Server muss ausgeführt werden und über die Dataflow-Worker-Maschinen erreichbar sein.
• Die Apache Kafka-Themen müssen vorhanden sein.

Kafka-Nachrichtenformat

Diese Vorlage unterstützt das Lesen von Nachrichten aus Kafka in den folgenden Formaten:

• JSON
• Binärcodierung in Avro
• Confluent Schema Registry-codiertes Avro

JSON-Format

Wenn Sie JSON-Nachrichten lesen möchten, setzen Sie den Vorlagenparameter messageFormat auf "JSON".

Binärcodierung in Avro

Wenn Sie binäre Avro-Nachrichten lesen möchten, legen Sie die folgenden Vorlagenparameter fest:

• messageFormat: "AVRO_BINARY_ENCODING".
• binaryAvroSchemaPath: Der Speicherort einer Avro-Schemadatei in Cloud Storage. Beispiel: gs://BUCKET_NAME/message-schema.avsc.

Weitere Informationen zum binären Avro-Format finden Sie in der Apache Avro-Dokumentation unter Binary encoding.

Confluent Schema Registry-codiertes Avro

Wenn Sie Nachrichten im in der Confluent Schema Registry codierten Avro lesen möchten, legen Sie die folgenden Vorlagenparameter fest:

• messageFormat: "AVRO_CONFLUENT_WIRE_FORMAT".

• schemaFormat: Einer der folgenden Werte:
  • "SINGLE_SCHEMA_FILE": Das Nachrichtenschema ist in einer Avro-Schemadatei definiert. Geben Sie den Cloud Storage-Speicherort der Schemadatei im Parameter confluentAvroSchemaPath an.
  • "SCHEMA_REGISTRY": Die Nachrichten werden mit der Confluent Schema Registry codiert. Geben Sie die URL der Confluent Schema Registry-Instanz im Parameter schemaRegistryConnectionUrl und den Authentifizierungsmodus im Parameter schemaRegistryAuthenticationMode an.

Weitere Informationen zu diesem Format finden Sie in der Confluent-Dokumentation unter Wire format.

Format der Ausgabedatei

Das Ausgabedateiformat entspricht dem Format der eingehenden Kafka-Nachricht. Wenn Sie beispielsweise JSON für das Kafka-Nachrichtenformat auswählen, werden JSON-Dateien in den Cloud Storage-Ausgabe-Bucket geschrieben.

Authentifizierung

Die Vorlage „Apache Kafka zu Cloud Storage“ unterstützt die SASL/PLAIN-Authentifizierung bei Kafka-Brokern.

Vorlagenparameter

Erforderliche Parameter

• readBootstrapServerAndTopic: Kafka-Thema, aus dem die Eingabe gelesen werden soll.
• outputDirectory: Das Pfad- und Dateinamenpräfix zum Schreiben von Ausgabedateien. Muss mit einem Schrägstrich enden. Beispiel: gs://your-bucket/your-path/.
• kafkaReadAuthenticationMode: Der Authentifizierungsmodus zur Verwendung mit dem Kafka-Cluster. Verwenden Sie KafkaAuthenticationMethod.NONE für keine Authentifizierung, KafkaAuthenticationMethod.SASL_PLAIN für SASL/PLAIN-Nutzername und -Passwort, KafkaAuthenticationMethod.SASL_SCRAM_512 für die SASL_SCRAM_512-Authentifizierung und KafkaAuthenticationMethod.TLS für die zertifikatbasierte Authentifizierung. KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS sollte nur für Google Cloud Apache Kafka für BigQuery-Cluster verwendet werden. Damit ist die Authentifizierung mit Standardanmeldedaten für Anwendungen möglich.
• messageFormat: Das Format der zu lesenden Kafka-Nachrichten. Die unterstützten Werte sind AVRO_CONFLUENT_WIRE_FORMAT (Confluent Schema Registry-codierter Avro), AVRO_BINARY_ENCODING (einfaches binäres Avro) und JSON. Die Standardeinstellung ist AVRO_CONFLUENT_WIRE_FORMAT.
• useBigQueryDLQ: Wenn dieser Wert „true“ ist, werden fehlgeschlagene Meldungen mit zusätzlichen Fehlerinformationen in BigQuery geschrieben. Die Standardeinstellung ist "false".

Optionale Parameter

• windowDuration: Die Fensterdauer/Größe, in der Daten in Cloud Storage geschrieben werden. Zulässige Formate sind: Ns (für Sekunden, Beispiel: 5s), Nm (für Minuten, Beispiel: 12m), Nh (für Stunden, Beispiel: 2h). Beispiel: 5m. Die Standardeinstellung ist "5m".
• outputFilenamePrefix: Das Präfix für die Namen der einzelnen Dateien im Fenstermodus. Beispiel: output-. Die Standardeinstellung ist "output".
• numShards: Die maximale Anzahl von Ausgabe-Shards, die beim Schreiben erzeugt werden. Eine höhere Anzahl von Shards erhöht den Durchsatz für das Schreiben in Cloud Storage, aber möglicherweise auch höhere Kosten für die Datenaggregation über Shards bei der Verarbeitung von Cloud Storage-Ausgabedateien. Der Standardwert wird von Dataflow festgelegt.
• enableCommitOffsets: Commit-Offsets verarbeiteter Nachrichten an Kafka. Wenn diese Option aktiviert ist, werden dadurch die Lücken oder doppelte Verarbeitung von Nachrichten beim Neustart der Pipeline minimiert. Angabe der Nutzergruppen-ID erforderlich. Die Standardeinstellung ist "false".
• consumerGroupId: Die eindeutige Kennung für die Nutzergruppe, zu der diese Pipeline gehört. Erforderlich, wenn Commit-Offsets für Kafka aktiviert sind. Die Standardeinstellung ist leer.
• kafkaReadOffset: Der Ausgangspunkt für das Lesen von Nachrichten, wenn keine festgeschriebenen Offsets vorhanden sind. Die früheste beginnt am Anfang, die neueste aus der neuesten Nachricht. Die Standardeinstellung ist "latest".
• kafkaReadUsernameSecretId: Die Secret-ID von Google Cloud Secret Manager, die den Kafka-Nutzernamen enthält, der für die SASL_PLAIN-Authentifizierung verwendet werden soll. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION> Die Standardeinstellung ist leer.
• kafkaReadPasswordSecretId: Die geheime ID von Google Cloud Secret Manager, die das Kafka-Passwort enthält, das für die SASL_PLAIN-Authentifizierung verwendet werden soll. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION> Die Standardeinstellung ist leer.
• kafkaReadKeystoreLocation: Der Google Cloud Storage-Pfad zur Java KeyStore-Datei (JKS), die das zur Authentifizierung beim Kafka-Cluster zu verwendende TLS-Zertifikat und den privaten Schlüssel enthält. Beispiel: gs://your-bucket/keystore.jks.
• kafkaReadTruststoreLocation: Der Google Cloud Storage-Pfad zur Java TrustStore-Datei (JKS), die die vertrauenswürdigen Zertifikate enthält, mit denen die Identität des Kafka-Brokers geprüft werden soll.
• kafkaReadTruststorePasswordSecretId: Die geheime ID von Google Cloud Secret Manager, die das Passwort enthält, das für den Zugriff auf die Java TrustStore-Datei (JKS) für die Kafka-TLS-Authentifizierung verwendet werden soll, z. B. projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeystorePasswordSecretId: Die geheime ID von Google Cloud Secret Manager, die das Passwort enthält, das für den Zugriff auf die Java KeyStore-Datei (JKS) für die Kafka-TLS-Authentifizierung verwendet werden soll. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeyPasswordSecretId: Die geheime ID von Google Cloud Secret Manager, die das Passwort enthält, das für den Zugriff auf den privaten Schlüssel in der Java KeyStore-Datei (JKS) für die Kafka-TLS-Authentifizierung verwendet werden soll. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramUsernameSecretId: Die Secret-ID von Google Cloud Secret Manager, die den Kafka-Nutzernamen enthält, der für die SASL_SCRAM-Authentifizierung verwendet werden soll. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>
• kafkaReadSaslScramPasswordSecretId: Die Secret-ID von Google Cloud Secret Manager, die das Kafka-Passwort enthält, das für die SASL_SCRAM-Authentifizierung verwendet werden soll. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>
• kafkaReadSaslScramTruststoreLocation: Der Google Cloud Storage-Pfad zur Java TrustStore-Datei (JKS), die die vertrauenswürdigen Zertifikate enthält, mit denen die Identität des Kafka-Brokers geprüft werden soll.
• kafkaReadSaslScramTruststorePasswordSecretId: Die Secret-ID von Google Cloud Secret Manager, die das Passwort für den Zugriff auf die Java TrustStore-Datei (JKS) für die Kafka-SASL_SCRAM-Authentifizierung enthält, z. B. projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• schemaFormat: Das Kafka-Schemaformat. Kann als SINGLE_SCHEMA_FILE oder SCHEMA_REGISTRY angegeben werden. Wenn SINGLE_SCHEMA_FILE angegeben ist, verwenden Sie das Schema, das in der Avro-Schemadatei für alle Nachrichten erwähnt wird. Wenn SCHEMA_REGISTRY angegeben ist, können die Nachrichten entweder ein einzelnes Schema oder mehrere Schemas haben. Die Standardeinstellung ist SINGLE_SCHEMA_FILE.
• confluentAvroSchemaPath: Der Google Cloud Storage-Pfad zu der einzelnen Avro-Schemadatei, die zur Decodierung aller Nachrichten in einem Thema verwendet wird. Die Standardeinstellung ist leer.
• schemaRegistryConnectionUrl: Die URL für die Confluent Schema Registry-Instanz, die zum Verwalten von Avro-Schemas für die Nachrichtendecodierung verwendet wird. Die Standardeinstellung ist leer.
• binaryAvroSchemaPath: Der Google Cloud Storage-Pfad zur Avro-Datei, die zum Decodieren binärcodierter Avro-Nachrichten verwendet wird. Die Standardeinstellung ist leer.
• schemaRegistryAuthenticationMode: Authentifizierungsmodus für die Schema-Registry. Kann NONE, TLS oder OAUTH sein. Die Standardeinstellung ist: NONE.
• schemaRegistryTruststoreLocation: Speicherort des SSL-Zertifikats, in dem der Trust Store für die Authentifizierung bei der Schema Registry gespeichert ist. Beispiel: /your-bucket/truststore.jks.
• schemaRegistryTruststorePasswordSecretId: SecretId im Secret Manager, in dem das Passwort für den Zugriff auf das Secret im Truststore gespeichert ist. Beispiel: projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryKeystoreLocation: Speicherort des Keystores, der das SSL-Zertifikat und den privaten Schlüssel enthält. Beispiel: /your-bucket/keystore.jks.
• schemaRegistryKeystorePasswordSecretId: SecretId im Secret Manager, in dem das Passwort für den Zugriff auf die Keystore-Datei gespeichert ist, z. B. projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryKeyPasswordSecretId: SecretId des Passworts, das für den Zugriff auf den im Keystore gespeicherten privaten Schlüssel des Kunden erforderlich ist, z. B. projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• schemaRegistryOauthClientId: Client-ID, die zum Authentifizieren des Schema Registry-Clients im OAUTH-Modus verwendet wird. Erforderlich für das Nachrichtenformat AVRO_CONFLUENT_WIRE_FORMAT.
• schemaRegistryOauthClientSecretId: Die Secret-ID von Google Cloud Secret Manager, die den Clientschlüssel enthält, der zum Authentifizieren des Schema Registry-Clients im OAUTH-Modus verwendet werden soll. Erforderlich für das Nachrichtenformat AVRO_CONFLUENT_WIRE_FORMAT. Beispiel: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• schemaRegistryOauthScope: Der Zugriffstokenbereich, der zum Authentifizieren des Schema Registry-Clients im OAUTH-Modus verwendet wird. Dieses Feld ist optional, da die Anfrage auch ohne Übergabe eines Bereichsparameters gestellt werden kann. Beispiel: openid.
• schemaRegistryOauthTokenEndpointUrl: Die HTTP(S)-basierte URL für den OAuth-/OIDC-Identitätsanbieter, der zum Authentifizieren des Schema Registry-Clients im OAUTH-Modus verwendet wird. Erforderlich für das Nachrichtenformat AVRO_CONFLUENT_WIRE_FORMAT.
• outputDeadletterTable: Voll qualifizierter BigQuery-Tabellenname für fehlgeschlagene Nachrichten. Nachrichten, die die Ausgabetabelle aus verschiedenen Gründen nicht erreicht haben (z.B. nicht übereinstimmendes Schema, fehlerhaft formatierte JSON-Datei), werden in diese Tabelle geschrieben.Die Tabelle wird von der Vorlage erstellt. Beispiel: your-project-id:your-dataset.your-table-name.

Führen Sie die Vorlage aus.

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
  

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

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