Dokumen ini menjelaskan cara membaca data dari Apache Kafka ke Dataflow dan mencakup tips performa serta praktik terbaik.
Untuk sebagian besar kasus penggunaan, pertimbangkan untuk menggunakan konektor Managed I/O untuk membaca dari Kafka.
Jika Anda memerlukan penyesuaian performa yang lebih canggih, pertimbangkan untuk menggunakan konektor KafkaIO
. Konektor KafkaIO
tersedia untuk
Java
atau dengan menggunakan
framework pipeline multi-bahasa
untuk Python
dan Go.
Keparalelan
Paralelisme dibatasi oleh dua faktor: jumlah maksimum pekerja
(max_num_workers
) dan jumlah partisi Kafka. Dataflow
secara default menggunakan fanout paralelisme 4 x max_num_workers
. Namun, fanout dibatasi oleh jumlah partisi. Misalnya, jika 100 vCPU tersedia,
tetapi pipeline hanya membaca dari 10 partisi Kafka, paralelisme maksimum adalah
10.
Untuk memaksimalkan paralelisme, sebaiknya miliki minimal 4 x
partisi Kafka max_num_workers
. Jika tugas Anda menggunakan
Runner v2, pertimbangkan untuk menyetel paralelisme yang lebih tinggi.
Titik awal yang baik adalah memiliki partisi yang sama dengan dua kali jumlah vCPU pekerja.
Jika Anda tidak dapat meningkatkan jumlah partisi, Anda dapat meningkatkan paralelisme dengan
memanggil KafkaIO.Read.withRedistribute
. Metode ini menambahkan
transformasi Redistribute
ke pipeline, yang memberikan petunjuk kepada
Dataflow untuk mendistribusikan ulang dan memparalelkan data secara lebih
efisien. Anda juga dapat menentukan jumlah optimal shard yang akan digunakan dalam langkah redistribusi, dengan memanggil KafkaIO.Read.withRedistributeNumKeys
.
Dataflow memperlakukan nilai ini sebagai petunjuk pengoptimalan.
Mendistribusikan ulang data akan menambah beberapa overhead tambahan untuk melakukan langkah pengacakan. Untuk mengetahui informasi selengkapnya, lihat
Mencegah penggabungan.
Coba pastikan beban antarpartisi relatif merata dan tidak miring. Jika beban tidak seimbang, hal ini dapat menyebabkan pemanfaatan pekerja yang buruk. Worker yang membaca dari partisi dengan beban yang lebih ringan mungkin relatif tidak aktif, sedangkan worker yang membaca dari partisi dengan beban berat mungkin tertinggal. Dataflow menyediakan metrik untuk backlog per partisi.
Jika beban tidak seimbang, penyeimbangan kerja dinamis dapat membantu mendistribusikan pekerjaan. Misalnya, Dataflow dapat mengalokasikan satu worker untuk membaca dari beberapa partisi bervolume rendah, dan mengalokasikan worker lain untuk membaca dari satu partisi bervolume 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 dengan volume rendah
Skenario umum adalah membaca dari banyak topik bervolume 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:
Gabungkan topik. Gabungkan topik sebelum diserap ke dalam Dataflow. Menyerap beberapa topik bervolume tinggi jauh lebih efisien daripada menyerap banyak topik bervolume rendah. Setiap topik bervolume tinggi dapat ditangani oleh satu tugas Dataflow yang sepenuhnya memanfaatkan pekerjanya.
Membaca beberapa topik. Jika Anda tidak dapat menggabungkan topik sebelum menyerapnya ke Dataflow, pertimbangkan untuk membuat pipeline yang membaca dari beberapa topik. Pendekatan ini memungkinkan Dataflow menetapkan beberapa topik ke worker 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 menurut nama topik untuk menerapkan logika yang berbeda per topik. Untuk kode contoh, lihat Membaca dari beberapa topik. Pertimbangkan opsi ini jika semua topik Anda ditempatkan bersama dalam cluster yang sama. Salah satu kekurangannya adalah masalah pada satu tujuan atau transformasi dapat menyebabkan semua topik menumpuk backlog.Untuk kasus penggunaan yang lebih lanjut, teruskan sekumpulan objek
KafkaSourceDescriptor
yang menentukan topik yang akan dibaca. MenggunakanKafkaSourceDescriptor
memungkinkan Anda memperbarui daftar topik nanti jika diperlukan. Fitur ini memerlukan Java dengan Runner v2.Beberapa langkah membaca. Untuk membaca dari topik yang berada di cluster yang berbeda, pipeline Anda dapat menyertakan beberapa instance
KafkaIO
. Saat tugas sedang berjalan, Anda dapat memperbarui setiap sumber menggunakan pemetaan transformasi. Menetapkan topik atau cluster baru hanya didukung saat menggunakan Runner v2. Kemampuan pengamatan adalah potensi tantangan dengan pendekatan ini, karena Anda perlu memantau setiap transformasi bacaan satu per satu, bukan mengandalkan metrik tingkat pipeline.
Melakukan kembali ke Kafka
Secara default, konektor KafkaIO
tidak menggunakan offset Kafka untuk melacak progres dan tidak melakukan kembali ke Kafka. Jika Anda memanggil
commitOffsetsInFinalize
, konektor akan berupaya
melakukan commit kembali ke Kafka setelah record di-commit di
Dataflow. Rekaman yang di-commit di Dataflow mungkin tidak
diproses sepenuhnya, jadi jika Anda
membatalkan pipeline, offset
mungkin di-commit tanpa rekaman diproses sepenuhnya.
Karena setelan enable.auto.commit=True
melakukan offset segera setelah dibaca dari
Kafka tanpa pemrosesan apa pun oleh Dataflow, sebaiknya jangan gunakan opsi ini.
Sebaiknya setel enable.auto.commit=False
dan
commitOffsetsInFinalize=True
. Jika Anda menyetel
enable.auto.commit
ke True
, data dapat hilang jika pipeline terganggu
saat diproses. Data yang sudah di-commit di Kafka mungkin 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 pelari
Konektor KafkaIO
memiliki dua implementasi dasar untuk pembacaan Kafka, yaitu
ReadFromKafkaViaUnbounded
yang lebih lama dan
ReadFromKafkaViaSDF
yang lebih baru. Dataflow secara otomatis memilih penerapan terbaik untuk tugas Anda berdasarkan bahasa SDK dan persyaratan tugas Anda. Hindari meminta implementasi runner atau 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 Dataflow v2.
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 dalam produksi, sebaiknya gunakan template Flex. Dengan menggunakan template Flex, pipeline diluncurkan dari lingkungan yang konsisten, yang dapat membantu mengurangi masalah konfigurasi lokal.
Logging dari KafkaIO
bisa sangat panjang. Pertimbangkan untuk mengurangi tingkat logging
dalam produksi sebagai berikut:
sdkHarnessLogLevelOverrides='{"org.apache.kafka.clients.consumer.internals.SubscriptionState":"WARN"}'.
Untuk mengetahui informasi selengkapnya, lihat Menetapkan level 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 diberi otorisasi.
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 kelebihan dan kekurangan:
- Menghubungkan menggunakan ruang alamat RFC 1918 bersama, dengan menggunakan salah satu opsi berikut:
- Akses cluster Kafka yang dihosting secara eksternal melalui alamat IP publik, dengan
menggunakan salah satu cara berikut:
- Internet publik
- Peering langsung
- Peering operator
Dedicated Interconnect adalah opsi terbaik untuk performa dan keandalan yang dapat diprediksi, tetapi penyiapannya dapat memakan 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 IPsec VPN memberi Anda akses langsung ke alamat IP RFC 1918 di Virtual Private Cloud (VPC), yang dapat menyederhanakan konfigurasi Kafka Anda. Jika Anda menggunakan topologi berbasis VPN, pertimbangkan untuk menyiapkan 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 di Google Cloud ke cluster Kafka tersebut, Anda memerlukan
kontrol lebih besar atas lokasi instance Dataflow Anda. Anda
dapat menggunakan Dataflow untuk mengonfigurasi network
dan subnetwork
parameter eksekusi.
Pastikan subnetwork yang sesuai memiliki alamat IP yang cukup tersedia agar Dataflow dapat meluncurkan instance saat mencoba melakukan penskalaan. Selain itu, saat membuat jaringan terpisah untuk meluncurkan instance Dataflow, pastikan Anda memiliki aturan firewall yang mengaktifkan traffic TCP di antara semua virtual machine dalam project. Jaringan default sudah mengonfigurasi aturan firewall ini.
Ruang alamat IP publik
Arsitektur ini menggunakan Transport Layer Security
(TLS) untuk mengamankan lalu lintas
antara klien eksternal dan Kafka, serta menggunakan lalu lintas yang tidak terenkripsi untuk komunikasi
antar-broker. Jika pendengar Kafka terikat ke antarmuka jaringan yang digunakan
untuk komunikasi internal dan eksternal, konfigurasi pendengar akan
sederhana. Namun, dalam banyak skenario, alamat broker Kafka yang diiklankan secara eksternal dalam cluster berbeda dengan antarmuka jaringan internal yang digunakan Kafka. Dalam skenario tersebut, 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 tersebut mungkin gagal diselesaikan untuk traffic internal.
Menyesuaikan Kafka
Setelan cluster Kafka dan klien Kafka dapat berdampak 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
. Defaultnya adalah 10.000. Nilai yang lebih tinggi seperti 100.000 dapat meningkatkan ukuran bundle, yang dapat meningkatkan performa secara signifikan jika pipeline Anda menyertakan penggabungan. Namun, nilai yang lebih tinggi juga dapat meningkatkan latensi. Untuk menetapkan nilai, gunakansetUnboundedReaderMaxElements
. Setelan ini tidak berlaku untuk Runner v2.unboundedReaderMaxReadTimeMs
. Defaultnya 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, gunakansetUnboundedReaderMaxReadTimeMs
. Setelan ini tidak berlaku untuk Runner v2.max.poll.records
. Defaultnya adalah 500. Nilai yang lebih tinggi mungkin berperforma lebih baik dengan mengambil lebih banyak catatan masuk secara bersamaan, terutama saat menggunakan Runner v2. Untuk menetapkan nilai, panggilwithConsumerConfigUpdates
.fetch.max.bytes
. Defaultnya adalah 1 MB. Nilai yang lebih tinggi dapat meningkatkan throughput dengan mengurangi jumlah permintaan, terutama saat menggunakan Runner v2. Namun, menyetelnya terlalu tinggi dapat meningkatkan latensi, meskipun pemrosesan hilir lebih cenderung menjadi hambatan utama. Nilai awal yang direkomendasikan adalah 100 MB. Untuk menetapkan nilai, panggilwithConsumerConfigUpdates
.max.partition.fetch.bytes
. Defaultnya 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 v2. Namun, menyetelnya terlalu tinggi dapat meningkatkan latensi, meskipun pemrosesan di hilir lebih cenderung menjadi hambatan utama. Nilai awal yang direkomendasikan adalah 100 MB. Untuk menetapkan nilai, panggilwithConsumerConfigUpdates
.consumerPollingTimeout
. Defaultnya 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, panggilwithConsumerPollingTimeout
.
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 callback handler yang disediakan oleh
Google Cloud Managed Service for Apache Kafka, 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.
Python
Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.
Membaca dari beberapa topik
Contoh ini menggunakan konektor KafkaIO
. Menunjukkan cara membaca dari beberapa topik Kafka dan menerapkan logika pipeline terpisah untuk setiap topik.
Untuk kasus penggunaan yang lebih canggih, teruskan secara dinamis serangkaian objek
KafkaSourceDescriptor
, sehingga Anda dapat memperbarui
daftar topik yang akan dibaca. Pendekatan ini memerlukan Java dengan Runner v2.
Java
Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.
Python
Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.
Langkah berikutnya
- Menulis ke Apache Kafka.
- Pelajari lebih lanjut I/O Terkelola.