Membaca dari Apache Kafka ke Dataflow

Dokumen ini menjelaskan cara membaca data dari Apache Kafka ke Dataflow dan menyertakan tips performa dan praktik terbaik.

Untuk sebagian besar kasus penggunaan, sebaiknya gunakan konektor I/O Terkelola untuk membaca dari Kafka.

Jika Anda memerlukan penyesuaian performa yang lebih canggih, sebaiknya gunakan konektor KafkaIO. Konektor KafkaIO tersedia untuk Java atau dengan menggunakan framework pipeline multi-bahasa untuk Python dan Go.

Keparalelan

Bagian berikut menjelaskan cara mengonfigurasi keparalelan saat membaca dari Kafka.

Ringkasan

Keparalelan dibatasi oleh dua faktor: jumlah maksimum pekerja (max_num_workers) dan jumlah partisi Kafka. Dataflow secara default menggunakan fanout keparalelan 4 x max_num_workers. Namun, fanout dibatasi oleh jumlah partisi. Misalnya, jika 100 vCPU tersedia, tetapi pipeline hanya membaca dari 10 partisi Kafka, keparalelan maksimumnya adalah 10.

Untuk memaksimalkan keparalelan, sebaiknya miliki setidaknya 4 x max_num_workers partisi Kafka. Jika tugas Anda menggunakan Runner Portabel, sebaiknya tetapkan keparalelan yang lebih tinggi. Titik awal yang baik adalah memiliki partisi yang sama dengan dua kali jumlah vCPU pekerja.

Mendistribusikan ulang

Jika tidak dapat meningkatkan jumlah partisi, Anda dapat meningkatkan keparalelan dengan memanggil KafkaIO.Read.withRedistribute. Metode ini menambahkan transformasi Redistribute ke pipeline, yang memberikan petunjuk ke Dataflow untuk mendistribusikan ulang dan memparalelkan data dengan lebih efisien. Sebaiknya tentukan jumlah shard yang optimal dengan memanggil KafkaIO.Read.withRedistributeNumKeys. Penggunaan KafkaIO.Read.withRedistribute saja dapat menghasilkan banyak kunci, yang menyebabkan masalah performa. Untuk mengetahui informasi selengkapnya, lihat Mengidentifikasi tahap dengan keparalelan tinggi. Mendistribusikan ulang data akan menambahkan beberapa overhead tambahan untuk melakukan langkah shuffle. Untuk mengetahui informasi selengkapnya, lihat Mencegah fusi.

Untuk meminimalkan biaya shuffle distribusi ulang, panggil KafkaIO.Read.withOffsetDeduplication. Mode ini meminimalkan jumlah data yang perlu dipertahankan sebagai bagian dari shuffle, sekaligus tetap menyediakan pemrosesan tepat satu kali.

Jika pemrosesan tepat satu kali tidak diperlukan, Anda dapat mengizinkan duplikat dengan memanggil KafkaIO.Read.withAllowDuplicates.

Tabel berikut merangkum opsi distribusi ulang:

Opsi Mode Pemrosesan Apache Beam Konfigurasi
Mendistribusikan ulang input Tepat satu kali v2.60+ KafkaIO.Read.withRedistribute()
Mengizinkan duplikat Minimal satu kali v2.60+ KafkaIO.Read.withRedistribute().withAllowDuplicates()
Penghapusan duplikat offset Tepat satu kali v2.69+ KafkaIO.Read.withRedistribute().withOffsetDeduplication()

Kemiringan beban

Cobalah untuk memastikan bahwa beban antara partisi relatif merata dan tidak miring. Jika beban miring, hal ini dapat menyebabkan pemanfaatan pekerja yang buruk. Pekerja yang membaca dari partisi dengan beban yang lebih ringan mungkin relatif tidak aktif, sedangkan pekerja yang membaca dari partisi dengan beban yang berat mungkin tertinggal. Dataflow menyediakan metrik untuk backlog per partisi.

Jika beban miring, penyeimbangan beban kerja dinamis dapat membantu mendistribusikan pekerjaan. Misalnya, Dataflow dapat mengalokasikan satu pekerja untuk membaca dari beberapa partisi volume rendah, dan mengalokasikan pekerja lain untuk membaca dari satu partisi volume tinggi. Namun, dua pekerja tidak dapat membaca dari partisi yang sama, sehingga partisi yang sangat padat masih dapat menyebabkan pipeline tertinggal.

Praktik terbaik

Bagian ini berisi rekomendasi untuk membaca dari Kafka ke Dataflow.

Topik volume rendah

Skenario umum adalah membaca dari banyak topik volume rendah secara bersamaan — misalnya, satu topik per pelanggan. Membuat tugas Dataflow terpisah untuk setiap topik tidak efisien dari segi biaya, karena setiap tugas memerlukan setidaknya satu pekerja penuh. Sebagai gantinya, pertimbangkan opsi berikut:

  • Menggabungkan topik. Gabungkan topik sebelum diserap ke Dataflow. Menyerap beberapa topik volume tinggi jauh lebih efisien daripada menyerap banyak topik volume rendah. Setiap topik volume tinggi dapat ditangani oleh satu tugas Dataflow yang memanfaatkan sepenuhnya pekerjanya.

  • Membaca beberapa topik. Jika Anda tidak dapat menggabungkan topik sebelum menyerapnya ke Dataflow, sebaiknya buat pipeline yang membaca dari beberapa topik. Pendekatan ini memungkinkan Dataflow menetapkan beberapa topik ke pekerja yang sama. Ada dua cara untuk menerapkan pendekatan ini:

    • Langkah pembacaan tunggal. Buat satu instance konektor KafkaIO dan konfigurasikan untuk membaca beberapa topik. Kemudian, filter berdasarkan nama topik untuk menerapkan logika yang berbeda per topik. Untuk kode contoh, lihat Membaca dari beberapa topik. Pertimbangkan opsi ini jika semua topik Anda berada di cluster yang sama. Salah satu kelemahannya adalah masalah dengan satu sink atau transformasi dapat menyebabkan semua topik mengakumulasi backlog.

      Untuk kasus penggunaan yang lebih canggih, teruskan kumpulan objek KafkaSourceDescriptor yang menentukan topik yang akan dibaca. Dengan menggunakan KafkaSourceDescriptor, Anda dapat memperbarui daftar topik nanti jika diperlukan. Fitur ini memerlukan Java dengan Runner Portabel.

    • Beberapa langkah pembacaan. Untuk membaca dari topik yang berada di cluster yang berbeda, pipeline Anda dapat menyertakan beberapa instance KafkaIO. Saat tugas berjalan, Anda dapat memperbarui sumber individual menggunakan pemetaan transformasi. Menetapkan topik atau cluster baru hanya didukung saat menggunakan Runner Portabel. Observabilitas adalah potensi tantangan dengan pendekatan ini, karena Anda perlu memantau setiap transformasi pembacaan individual, bukan mengandalkan metrik tingkat pipeline.

Melakukan commit kembali ke Kafka

Secara default, konektor KafkaIO tidak menggunakan offset Kafka untuk melacak progres dan tidak melakukan commit kembali ke Kafka. Jika Anda memanggil commitOffsetsInFinalize, konektor akan melakukan upaya terbaik untuk melakukan commit kembali ke Kafka setelah data di-commit di Dataflow. Data yang di-commit di Dataflow mungkin tidak diproses sepenuhnya, jadi jika Anda membatalkan pipeline, offset mungkin di-commit tanpa data pernah diproses sepenuhnya.

Karena menetapkan enable.auto.commit=True akan melakukan commit offset segera setelah dibaca dari Kafka tanpa pemrosesan apa pun oleh Dataflow, sebaiknya jangan gunakan opsi ini. Sebaiknya tetapkan enable.auto.commit=False dan commitOffsetsInFinalize=True. Jika Anda menetapkan enable.auto.commit ke True, data dapat hilang jika pipeline terganggu saat diproses. Data yang sudah di-commit di Kafka mungkin akan dihapus.

Watermark

Secara default, konektor KafkaIO menggunakan waktu pemrosesan saat ini untuk menetapkan watermark output dan waktu peristiwa. Untuk mengubah perilaku ini, panggil withTimestampPolicyFactory dan tetapkan TimestampPolicy. Beam menyediakan implementasi TimestampPolicy yang menghitung watermark berdasarkan waktu penambahan log Kafka atau waktu pembuatan pesan.

Pertimbangan runner

Konektor KafkaIO memiliki dua implementasi dasar untuk pembacaan Kafka, yaitu yang lebih lama ReadFromKafkaViaUnbounded dan yang lebih baru ReadFromKafkaViaSDF. Dataflow secara otomatis memilih implementasi terbaik untuk tugas Anda berdasarkan bahasa SDK dan persyaratan tugas. Hindari meminta runner atau implementasi Kafka secara eksplisit kecuali jika Anda memerlukan fitur tertentu yang hanya tersedia dalam implementasi tersebut. Untuk mengetahui informasi selengkapnya tentang cara memilih runner, lihat Menggunakan Runner Portabel Dataflow.

Jika pipeline Anda menggunakan withTopic atau withTopics, implementasi yang lebih lama akan mengkueri Kafka pada waktu pembuatan pipeline untuk partisi yang tersedia. Mesin yang membuat pipeline harus memiliki izin untuk terhubung ke Kafka. Jika Anda menerima error izin, pastikan Anda memiliki izin untuk terhubung ke Kafka secara lokal. Anda dapat menghindari masalah ini dengan menggunakan withTopicPartitions, yang tidak terhubung ke Kafka pada waktu pembuatan pipeline.

Men-deploy ke produksi

Saat men-deploy solusi Anda dalam produksi, sebaiknya gunakan template Fleksibel. Dengan menggunakan template Fleksibel, pipeline diluncurkan dari lingkungan yang konsisten, yang dapat membantu mengurangi masalah konfigurasi lokal.

Logging dari KafkaIO bisa sangat verbose. Sebaiknya kurangi tingkat logging dalam produksi sebagai berikut:

sdkHarnessLogLevelOverrides='{"org.apache.kafka.clients.consumer.internals.SubscriptionState":"WARN"}'.

Untuk mengetahui informasi selengkapnya, lihat Menetapkan tingkat log pekerja pipeline.

Mengonfigurasi jaringan

Secara default, Dataflow meluncurkan instance dalam jaringan Virtual Private Cloud (VPC) default Anda. Bergantung pada konfigurasi Kafka, Anda mungkin perlu mengonfigurasi jaringan dan subnet yang berbeda untuk Dataflow. Untuk mengetahui informasi selengkapnya, lihat Menentukan jaringan dan subnetwork. Saat mengonfigurasi jaringan, buat aturan firewall yang memungkinkan mesin pekerja Dataflow menjangkau broker Kafka.

Jika Anda menggunakan Kontrol Layanan VPC, tempatkan cluster Kafka dalam perimeter Kontrol Layanan VPC, atau perluas perimeter ke VPN atau Cloud Interconnect yang sah.

Jika cluster Kafka Anda di-deploy di luar Google Cloud, Anda harus membuat koneksi jaringan antara Dataflow dan cluster Kafka. Ada beberapa opsi jaringan dengan berbagai pertimbangan:

Dedicated Interconnect adalah opsi terbaik untuk performa dan keandalan yang dapat diprediksi, tetapi penyiapannya mungkin memerlukan waktu lebih lama karena pihak ketiga harus menyediakan sirkuit baru. Dengan topologi berbasis IP publik, Anda dapat memulai dengan cepat karena hanya sedikit pekerjaan jaringan yang perlu dilakukan.

Dua bagian berikutnya menjelaskan opsi ini secara lebih mendetail.

Ruang alamat RFC 1918 bersama

Dedicated Interconnect dan VPN IPsec memberi Anda akses langsung ke alamat IP RFC 1918 di Virtual Private Cloud (VPC), yang dapat menyederhanakan konfigurasi Kafka. Jika Anda menggunakan topologi berbasis VPN, sebaiknya siapkan VPN throughput tinggi.

Secara default, Dataflow meluncurkan instance di jaringan VPC default Anda. Dalam topologi jaringan pribadi dengan rute yang ditentukan secara eksplisit di Cloud Router yang menghubungkan subnetwork in Google Cloud ke cluster Kafka tersebut, Anda memerlukan lebih banyak kontrol atas lokasi instance Dataflow. Anda dapat menggunakan Dataflow untuk mengonfigurasi network dan subnetwork parameter eksekusi.

Pastikan subnetwork yang sesuai memiliki alamat IP yang cukup untuk diluncurkan Dataflow saat mencoba melakukan penskalaan. Selain itu, saat membuat jaringan terpisah untuk meluncurkan instance Dataflow, pastikan Anda memiliki aturan firewall yang memungkinkan traffic TCP di antara semua mesin virtual dalam project. Jaringan default sudah memiliki aturan firewall ini yang dikonfigurasi.

Ruang alamat IP publik

Arsitektur ini menggunakan Transport Layer Security (TLS) untuk mengamankan traffic antara klien eksternal dan Kafka, serta menggunakan traffic yang tidak dienkripsi untuk komunikasi antar-broker. Saat pemroses Kafka terikat ke antarmuka jaringan yang digunakan untuk komunikasi internal dan eksternal, konfigurasi pemrosesnya akan mudah. Namun, dalam banyak skenario, alamat broker Kafka yang diiklankan secara eksternal dalam cluster berbeda dengan antarmuka jaringan internal yang digunakan Kafka. Dalam skenario seperti itu, Anda dapat menggunakan properti advertised.listeners:

# Configure protocol map
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:SSL

# Use plaintext for inter-broker communication
inter.broker.listener.name=INTERNAL

# Specify that Kafka listeners should bind to all local interfaces
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9093

# Separately, specify externally visible address
advertised.listeners=INTERNAL://kafkabroker-n.mydomain.com:9092,EXTERNAL://kafkabroker-n.mydomain.com:9093

Klien eksternal terhubung menggunakan port 9093 melalui saluran "SSL", dan klien internal terhubung menggunakan port 9092 melalui saluran teks biasa. Saat Anda menentukan alamat di bagian advertised.listeners, gunakan nama DNS (kafkabroker-n.mydomain.com, dalam contoh ini) yang di-resolve ke instance yang sama untuk traffic eksternal dan internal. Penggunaan alamat IP publik mungkin tidak berfungsi karena alamat mungkin gagal di-resolve untuk traffic internal.

Menyesuaikan Kafka

Setelan cluster Kafka dan klien Kafka dapat memberikan dampak besar pada performa. Secara khusus, setelan berikut mungkin terlalu rendah. Bagian ini memberikan beberapa titik awal yang disarankan, tetapi Anda harus bereksperimen dengan nilai ini untuk workload tertentu.

  • unboundedReaderMaxElements. Default-nya adalah 10.000. Nilai yang lebih tinggi seperti 100.000 dapat meningkatkan ukuran paket, yang dapat meningkatkan performa secara signifikan jika pipeline Anda menyertakan agregasi. Namun, nilai yang lebih tinggi juga dapat meningkatkan latensi. Untuk menetapkan nilai, gunakan setUnboundedReaderMaxElements. Setelan ini tidak berlaku untuk Runner Portabel. Untuk Runner Portabel, gunakan opsi layanan Dataflow sdf_checkpoint_after_output_bytes.

  • unboundedReaderMaxReadTimeMs. Default-nya adalah 10.000 md. Nilai yang lebih tinggi seperti 20.000 md dapat meningkatkan ukuran paket, sedangkan nilai yang lebih rendah seperti 5.000 md dapat mengurangi latensi atau backlog. Untuk menetapkan nilai, gunakan setUnboundedReaderMaxReadTimeMs. Setelan ini tidak berlaku untuk Runner Portabel. Untuk Runner Portabel, gunakan opsi layanan Dataflow sdf_checkpoint_after_duration.

  • max.poll.records. Default-nya adalah 500. Nilai yang lebih tinggi mungkin memiliki performa yang lebih baik dengan mengambil lebih banyak data masuk secara bersamaan, terutama saat menggunakan Runner Portabel. Untuk menetapkan nilai, panggil withConsumerConfigUpdates.

  • fetch.max.bytes. Default-nya adalah 1 MB. Nilai yang lebih tinggi dapat meningkatkan throughput dengan mengurangi jumlah permintaan, terutama saat menggunakan Runner Portabel. Namun, menetapkannya terlalu tinggi dapat meningkatkan latensi, meskipun pemrosesan downstream lebih cenderung menjadi hambatan utama. Nilai awal yang direkomendasikan adalah 100 MB. Untuk menetapkan nilai, panggil withConsumerConfigUpdates.

  • max.partition.fetch.bytes. Default-nya adalah 1 MB. Parameter ini menetapkan jumlah data maksimum per partisi yang ditampilkan server. Meningkatkan nilai dapat meningkatkan throughput dengan mengurangi jumlah permintaan, terutama saat menggunakan Runner Portabel. Namun, menetapkannya terlalu tinggi dapat meningkatkan latensi, meskipun pemrosesan downstream lebih cenderung menjadi hambatan utama. Nilai awal yang direkomendasikan adalah 100 MB. Untuk menetapkan nilai, panggil withConsumerConfigUpdates.

  • consumerPollingTimeout. Default-nya adalah 2 detik. Jika klien konsumen mengalami waktu tunggu sebelum dapat membaca data apa pun, coba tetapkan nilai yang lebih tinggi. Setelan ini paling sering relevan saat melakukan pembacaan lintas region atau pembacaan dengan jaringan yang lambat. Untuk menetapkan nilai, panggil withConsumerPollingTimeout.

Pastikan receive.buffer.bytes cukup besar untuk menangani ukuran pesan. Jika nilainya terlalu kecil, log mungkin menunjukkan bahwa konsumen terus dibuat ulang dan mencari offset tertentu.

Contoh

Contoh kode berikut menunjukkan cara membuat pipeline Dataflow yang membaca dari Kafka. Saat menggunakan Kredensial Default Aplikasi bersama dengan Layanan Terkelola Google Cloud untuk Apache Kafka yang menyediakan pengendali callback, diperlukan kafka-clients versi 3.7.0 atau yang lebih tinggi.

Membaca dari satu topik

Contoh ini menggunakan konektor I/O Terkelola . Contoh ini menunjukkan cara membaca dari topik Kafka dan menulis payload pesan ke file teks.

Java

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

import com.google.common.collect.ImmutableMap;
import java.io.UnsupportedEncodingException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class KafkaRead {

  public static Pipeline createPipeline(Options options) {

    // Create configuration parameters for the Managed I/O transform.
    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("bootstrap_servers", options.getBootstrapServer())
        .put("topic", options.getTopic())
        .put("format", "RAW")
        .put("max_read_time_seconds", 15)
        .put("auto_offset_reset_config", "earliest")
        .build();

    // Build the pipeline.
    var pipeline = Pipeline.create(options);
    pipeline
        // Read messages from Kafka.
        .apply(Managed.read(Managed.KAFKA).withConfig(config)).getSinglePCollection()
        // Get the payload of each message and convert to a string.
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via((row -> {
              var bytes = row.getBytes("payload");
              try {
                return new String(bytes, "UTF-8");
              } catch (UnsupportedEncodingException e) {
                throw new RuntimeException(e);
              }
            })))
        // Write the payload to a text file.
        .apply(TextIO
            .write()
            .to(options.getOutputPath())
            .withSuffix(".txt")
            .withNumShards(1));
    return pipeline;
  }
}

Python

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

import argparse

import apache_beam as beam

from apache_beam import window
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


def read_from_kafka() -> None:
    # Parse the pipeline options passed into the application. Example:
    #     --topic=$KAFKA_TOPIC --bootstrap_server=$BOOTSTRAP_SERVER
    #     --output=$CLOUD_STORAGE_BUCKET --streaming
    # For more information, see
    # https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    class MyOptions(PipelineOptions):
        @staticmethod
        def _add_argparse_args(parser: argparse.ArgumentParser) -> None:
            parser.add_argument("--topic")
            parser.add_argument("--bootstrap_server")
            parser.add_argument("--output")

    options = MyOptions()
    with beam.Pipeline(options=options) as pipeline:
        (
            pipeline
            # Read messages from an Apache Kafka topic.
            | beam.managed.Read(
                beam.managed.KAFKA,
                config={
                  "bootstrap_servers": options.bootstrap_server,
                  "topic": options.topic,
                  "data_format": "RAW",
                  "auto_offset_reset_config": "earliest",
                  # The max_read_time_seconds parameter is intended for testing.
                  # Avoid using this parameter in production.
                  "max_read_time_seconds": 5
                }
            )
            # Subdivide the output into fixed 5-second windows.
            | beam.WindowInto(window.FixedWindows(5))
            | WriteToText(
                file_path_prefix=options.output, file_name_suffix=".txt", num_shards=1
            )
        )

Membaca dari beberapa topik

Contoh ini menggunakan konektor KafkaIO. Contoh ini menunjukkan cara membaca dari beberapa topik Kafka dan menerapkan logika pipeline terpisah untuk setiap topik.

Untuk kasus penggunaan yang lebih canggih, teruskan kumpulan objek KafkaSourceDescriptor secara dinamis, sehingga Anda dapat memperbarui daftar topik yang akan dibaca. Pendekatan ini memerlukan Java dengan Runner Portabel.

Java

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.joda.time.Duration;
import org.joda.time.Instant;

public class KafkaReadTopics {

  public static Pipeline createPipeline(Options options) {
    String topic1 = options.getTopic1();
    String topic2 = options.getTopic2();

    // Build the pipeline.
    var pipeline = Pipeline.create(options);
    var allTopics = pipeline
        .apply(KafkaIO.<Long, String>read()
            .withTopics(List.of(topic1, topic2))
            .withBootstrapServers(options.getBootstrapServer())
            .withKeyDeserializer(LongDeserializer.class)
            .withValueDeserializer(StringDeserializer.class)
            .withMaxReadTime(Duration.standardSeconds(10))
            .withStartReadTime(Instant.EPOCH)
        );

    // Create separate pipeline branches for each topic.
    // The first branch filters on topic1.
    allTopics
        .apply(Filter.by(record -> record.getTopic().equals(topic1)))
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via(record -> record.getKV().getValue()))
        .apply(TextIO.write()
            .to(topic1)
            .withSuffix(".txt")
            .withNumShards(1)
        );

    // The second branch filters on topic2.
    allTopics
        .apply(Filter.by(record -> record.getTopic().equals(topic2)))
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via(record -> record.getKV().getValue()))
        .apply(TextIO.write()
            .to(topic2)
            .withSuffix(".txt")
            .withNumShards(1)
        );
    return pipeline;
  }
}

Python

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

import argparse

import apache_beam as beam

from apache_beam.io.kafka import ReadFromKafka
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


def read_from_kafka() -> None:
    # Parse the pipeline options passed into the application. Example:
    #   --bootstrap_server=$BOOTSTRAP_SERVER --output=$STORAGE_BUCKET --streaming
    # For more information, see
    # https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    class MyOptions(PipelineOptions):
        @staticmethod
        def _add_argparse_args(parser: argparse.ArgumentParser) -> None:
            parser.add_argument('--bootstrap_server')
            parser.add_argument('--output')

    options = MyOptions()
    with beam.Pipeline(options=options) as pipeline:
        # Read from two Kafka topics.
        all_topics = pipeline | ReadFromKafka(consumer_config={
                "bootstrap.servers": options.bootstrap_server
            },
            topics=["topic1", "topic2"],
            with_metadata=True,
            max_num_records=10,
            start_read_time=0
        )

        # Filter messages from one topic into one branch of the pipeline.
        (all_topics
            | beam.Filter(lambda message: message.topic == 'topic1')
            | beam.Map(lambda message: message.value.decode('utf-8'))
            | "Write topic1" >> WriteToText(
                file_path_prefix=options.output + '/topic1/output',
                file_name_suffix='.txt',
                num_shards=1))

        # Filter messages from the other topic.
        (all_topics
            | beam.Filter(lambda message: message.topic == 'topic2')
            | beam.Map(lambda message: message.value.decode('utf-8'))
            | "Write topic2" >> WriteToText(
                file_path_prefix=options.output + '/topic2/output',
                file_name_suffix='.txt',
                num_shards=1))

Langkah berikutnya