Halaman ini diterjemahkan oleh Cloud Translation API.

Template Apache Kafka ke Cloud Storage

Template Apache Kafka ke Cloud Storage adalah pipeline streaming yang menyerap data teks dari Layanan yang Dikelola Google Cloud untuk Apache Kafka dan menghasilkan kumpulan data ke Cloud Storage.

Anda juga dapat menggunakan template Apache Kafka to BigQuery dengan Kafka yang dikelola sendiri atau eksternal.

Persyaratan pipeline

Bucket Cloud Storage output harus ada.
Server broker Apache Kafka harus berjalan dan dapat dijangkau dari mesin pekerja Dataflow.
Topik Apache Kafka harus ada.

Format pesan Kafka

Template Apache Kafka ke Cloud Storage mendukung pembacaan pesan dari Kafka dalam format berikut: CONFLUENT_AVRO_WIRE_FORMAT dan JSON.

Format file output

Format file output memiliki format yang sama dengan pesan Kafka input. Misalnya, jika Anda memilih JSON untuk format pesan Kafka, file JSON akan ditulis ke bucket Cloud Storage output.

Autentikasi

Template Apache Kafka ke Cloud Storage mendukung autentikasi SASL/PLAIN untuk broker Kafka.

Parameter template

Parameter yang diperlukan

readBootstrapServerAndTopic : Topik Kafka untuk membaca input.
kafkaReadAuthenticationMode : Mode autentikasi yang akan digunakan dengan cluster Kafka. Gunakan NONE jika tidak ada autentikasi dan SASL_PLAIN untuk nama pengguna dan sandi SASL/PLAIN. Apache Kafka untuk BigQuery hanya mendukung mode autentikasi SASL_PLAIN. Default-nya adalah: SASL_PLAIN.
outputDirectory : Jalur dan awalan nama file untuk menulis file output. Harus diakhiri dengan garis miring. (Contoh: gs://your-bucket/your-path/).
messageFormat : Format pesan Kafka yang akan dibaca. Nilai yang didukung adalah AVRO_CONFLUENT_WIRE_FORMAT (Confluent Schema Registry yang dienkode Avro), AVRO_BINARY_ENCODING (Biner polos Avro), dan JSON. Default-nya adalah: AVRO_CONFLUENT_WIRE_FORMAT.

Parameter opsional

windowDuration : Durasi/ukuran periode tempat data akan ditulis ke Cloud Storage. Format yang diizinkan adalah: Ns (untuk detik, misalnya: 5d), Nm (untuk menit, misalnya: 12m), Nh (untuk jam, misalnya: 2j). (Contoh: 5 m). Default-nya adalah: 5m.
outputFilenamePrefix : Awalan yang akan ditempatkan pada setiap file yang memiliki jendela. (Contoh: output-). Default-nya adalah: output.
numShards : Jumlah maksimum shard output yang dihasilkan saat menulis. Jumlah shard yang lebih tinggi berarti throughput yang lebih tinggi untuk menulis ke Cloud Storage, tetapi berpotensi menyebabkan biaya agregasi data yang lebih tinggi di seluruh shard saat memproses file Cloud Storage output. Nilai default ditentukan oleh Dataflow.
enableCommitOffsets : Meng-commit offset pesan yang diproses ke Kafka. Jika diaktifkan, tindakan ini akan meminimalkan kesenjangan atau pemrosesan duplikat pesan saat memulai ulang pipeline. Memerlukan ID Grup Konsumen. Default-nya adalah: false.
consumerGroupId : ID unik untuk grup konsumen tempat pipeline ini berada. Wajib jika Commit Offsets to Kafka diaktifkan. Nilai defaultnya adalah kosong.
kafkaReadOffset : Titik awal untuk membaca pesan saat tidak ada offset yang di-commit. Pesan paling awal dimulai dari awal, terbaru dari pesan terbaru. Default-nya adalah: terbaru.
kafkaReadUsernameSecretId : ID rahasia Google Cloud Secret Manager yang berisi nama pengguna Kafka yang akan digunakan dengan autentikasi SASL_PLAIN. (Contoh: projects/<PROJECT_ID>/secrets/<SECRET_ID>/version/<SECRET_VERSION>). Nilai defaultnya adalah kosong.
kafkaReadPasswordSecretId : ID rahasia Google Cloud Secret Manager yang berisi sandi Kafka untuk digunakan dengan autentikasi SASL_PLAIN. (Contoh: projects/<PROJECT_ID>/secrets/<SECRET_ID>/version/<SECRET_VERSION>). Nilai defaultnya adalah kosong.
schemaFormat : Format skema Kafka. Dapat disediakan sebagai SINGLE_SCHEMA_FILE atau SCHEMA_REGISTRY. Jika SINGLE_SCHEMA_FILE ditentukan, semua pesan harus memiliki skema yang disebutkan dalam file skema avro. Jika SCHEMA_REGISTRY ditentukan, pesan dapat memiliki satu skema atau beberapa skema. Default-nya adalah: SINGLE_SCHEMA_FILE.
confluentAvroSchemaPath : Jalur Google Cloud Storage ke file skema Avro tunggal yang digunakan untuk mendekode semua pesan dalam topik. Nilai defaultnya adalah kosong.
schemaRegistryConnectionUrl : URL untuk instance Confluent Schema Registry yang digunakan untuk mengelola skema Avro untuk decoding pesan. Nilai defaultnya adalah kosong.
binaryAvroSchemaPath : Jalur Google Cloud Storage ke file skema Avro yang digunakan untuk mendekode pesan Avro yang dienkode dengan biner. Nilai defaultnya adalah kosong.

Menjalankan template

Konsol

Buka halaman Create job from template Dataflow.

Buka Buat tugas dari template

Di kolom Job name, masukkan nama tugas yang unik.
Opsional: Untuk Regional endpoint, pilih nilai dari menu drop-down. Default region adalah us-central1.
Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.
Dari menu drop-down Dataflow template, pilih the Kafka to Cloud Storage template.
Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
Opsional: Untuk beralih dari pemrosesan tepat satu kali ke mode streaming minimal satu kali, pilih Setidaknya Sekali.
Klik Run job.

gcloud

Di shell atau terminal Anda, jalankan template:

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

Ganti kode berikut:

PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
JOB_NAME: nama pekerjaan unik pilihan Anda
REGION_NAME: wilayah tempat Anda ingin men-deploy tugas Dataflow Anda—misalnya, us-central1
VERSION: versi {i>template<i} yang ingin Anda gunakan.
Anda dapat menggunakan nilai berikut:
- latest untuk menggunakan template versi terbaru, yang tersedia di Folder induk non-dated di bucket— gs://dataflow-templates-REGION_NAME/latest/
- nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi tertentu dari , yang dapat ditemukan disarangkan di masing-masing folder induk bertanggal dalam bucket— gs://dataflow-templates-REGION_NAME/
Perhatian: Versi template terbaru mungkin diupdate dengan menyebabkan gangguan perubahan. Lingkungan produksi Anda harus menggunakan template yang disimpan dalam versi terbaru dated untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi di alur kerja produksi.
BIGQUERY_TABLE: nama tabel Cloud Storage Anda
KAFKA_TOPICS: daftar topik Apache Kakfa. Jika beberapa topik disediakan, Anda harus meng-escape koma. Lihat gcloud topic escaping.
PATH_TO_JAVASCRIPT_UDF_FILE: URI Cloud Storage dari file .js yang menentukan JavaScript yang ditentukan pengguna (UDF) yang ingin Anda gunakan—misalnya, gs://my-bucket/my-udfs/my_file.js
JAVASCRIPT_FUNCTION: nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan
Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, maka nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.
KAFKA_SERVER_ADDRESSES: daftar alamat IP server broker Apache Kafka. Setiap alamat IP harus memiliki nomor port yang dapat digunakan untuk mengakses server. Contoh: 35.70.252.199:9092. Jika ada beberapa alamat, Anda harus meng-escape koma. Lihat gcloud topic escaping.

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk informasi selengkapnya tentang API dan cakupan otorisasinya, lihat 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",
   }
}

Ganti kode berikut:

PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
JOB_NAME: nama pekerjaan unik pilihan Anda
LOCATION: wilayah tempat Anda ingin men-deploy tugas Dataflow Anda—misalnya, us-central1
VERSION: versi {i>template<i} yang ingin Anda gunakan.
Anda dapat menggunakan nilai berikut:
- latest untuk menggunakan template versi terbaru, yang tersedia di Folder induk non-dated di bucket— gs://dataflow-templates-REGION_NAME/latest/
- nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi tertentu dari , yang dapat ditemukan disarangkan di masing-masing folder induk bertanggal dalam bucket— gs://dataflow-templates-REGION_NAME/
Perhatian: Versi template terbaru mungkin diupdate dengan menyebabkan gangguan perubahan. Lingkungan produksi Anda harus menggunakan template yang disimpan dalam versi terbaru dated untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi di alur kerja produksi.
BIGQUERY_TABLE: nama tabel Cloud Storage Anda
KAFKA_TOPICS: daftar topik Apache Kakfa. Jika beberapa topik disediakan, Anda harus meng-escape koma. Lihat gcloud topic escaping.
PATH_TO_JAVASCRIPT_UDF_FILE: URI Cloud Storage dari file .js yang menentukan JavaScript yang ditentukan pengguna (UDF) yang ingin Anda gunakan—misalnya, gs://my-bucket/my-udfs/my_file.js
JAVASCRIPT_FUNCTION: nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan
Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, maka nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.
KAFKA_SERVER_ADDRESSES: daftar alamat IP server broker Apache Kafka. Setiap alamat IP harus memiliki nomor port yang dapat digunakan untuk mengakses server. Contoh: 35.70.252.199:9092. Jika ada beberapa alamat, Anda harus meng-escape koma. Lihat gcloud topic escaping.

Untuk informasi selengkapnya, lihat Menulis data dari Kafka ke Cloud Storage dengan Dataflow.

Kode sumber template

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

Langkah selanjutnya

Pelajari template Dataflow.
Lihat daftar template yang disediakan Google.