Tabel berikut mencantumkan jenis konektor Kafka Connect yang didukung oleh Managed Service untuk Apache Kafka. Anda dapat menggunakan konektor ini untuk mengintegrasikan Apache Kafka dengan aplikasi Anda dan dengan layanan Google Cloudlainnya.
| Konektor | Deskripsi | Kasus penggunaan |
|---|---|---|
| BigQuery Sink | Melakukan streaming data dari topik Kafka ke tabel BigQuery. | Data warehousing, analisis |
| Sumber Cloud SQL untuk PostgreSQL | Membaca perubahan tingkat baris dari database Cloud SQL untuk PostgreSQL dan menuliskannya ke Kafka. | Integrasi database, memantau dan merespons peristiwa database |
| Sink Cloud Storage | Melakukan streaming data dari topik Kafka ke bucket Cloud Storage. | Penyerapan data lake, pengarsipan data |
| Sumber PostgreSQL Generik | Membaca perubahan tingkat baris dari database PostgreSQL dan menulisnya ke Kafka. | Integrasi database, memantau dan merespons peristiwa database |
| Checkpoint MirrorMaker 2.0 | Menyalin offset konsumen dari satu cluster Kafka ke cluster Kafka lainnya. | Replikasi data, disaster recovery, migrasi data |
| Heartbeat MirrorMaker 2.0 | Membuat pesan heartbeat berkala dan menulisnya ke topik | Memantau kondisi cluster, konektivitas jaringan, atau latensi replikasi end-to-end antar-cluster. |
| Sumber MirrorMaker 2.0 | Mereplikasi topik dan data dari satu cluster Kafka ke cluster Kafka lainnya. | Replikasi data, disaster recovery, migrasi data |
| Sink Pub/Sub | Melakukan streaming data dari topik Kafka ke topik Pub/Sub. | Integrasi layanan, notifikasi real-time |
| Sumber Pub/Sub | Melakukan streaming pesan dari langganan Pub/Sub ke topik Kafka. | Penyerapan data real-time, arsitektur berbasis peristiwa |
Pelaku Konversi
Konverter bertanggung jawab atas serialisasi dan deserialisasi data data Kafka. Converter menerjemahkan antara format byte mentah yang ditemukan di topik Kafka dan representasi data internal terstruktur yang digunakan oleh Kafka Connect.
Untuk konektor sink, konverter mendeserialisasi data dari format kabel topik ke format data internal Kafka Connect, yang digunakan konektor untuk menulis ke sistem target.
Untuk konektor sumber, konverter melakukan serialisasi data dari format data internal Kafka Connect ke format kabel yang ditentukan untuk topik Kafka.
Konverter memastikan bahwa konektor membaca atau menulis data Kafka dalam format yang kompatibel dengan sistem eksternal.
Saat Anda mengonfigurasi konektor, tetapkan properti berikut:
Konverter kunci (
key.converter): Konverter yang akan digunakan saat melakukan serialisasi dan deserialisasi kunci data Kafka.Konverter nilai (
value.converter): Konverter yang akan digunakan saat melakukan serialisasi dan deserialisasi nilai data Kafka.
Jika Anda tidak menentukan pengonversi, jenis pengonversi default adalah
org.apache.kafka.connect.converters.ByteArrayConverter, yang meneruskan
data dalam format byte mentah.
Konverter yang didukung
Managed Service untuk Apache Kafka mendukung konverter bawaan berikut:
| Pengonversi | Format |
|---|---|
io.confluent.connect.avro.AvroConverter |
Apache Avro |
org.apache.kafka.connect.converters.BooleanConverter |
Boolean |
org.apache.kafka.connect.converters.ByteArrayConverter |
Array byte Jenis pengonversi default. Mempertahankan konten pesan yang sama persis di dua sistem. |
org.apache.kafka.connect.converters.DoubleConverter |
Ganda |
org.apache.kafka.connect.converters.FloatConverter |
Float |
org.apache.kafka.connect.converters.IntegerConverter |
Bilangan bulat |
org.apache.kafka.connect.json.JsonConverter |
JSON Untuk data JSON tanpa skema, tetapkan juga |
org.apache.kafka.connect.converters.LongConverter |
Long |
org.apache.kafka.connect.converters.ShortConverter |
Versi singkat |
org.apache.kafka.connect.storage.StringConverter |
String |
Pilihan pengonversi bergantung pada jenis konektor dan data yang Anda simpan di Kafka. Untuk mengetahui informasi selengkapnya, lihat dokumentasi untuk konektor tertentu.
Tasks
Konektor mentransfer data dengan membuat satu atau beberapa tugas yang beroperasi secara
paralel. Untuk menetapkan batas atas jumlah tugas yang dibuat konektor, tetapkan properti konfigurasi tasks.max konektor. Konektor mungkin membuat
tugas yang lebih sedikit dari nilai ini.
Meningkatkan nilai tasks.max dapat meningkatkan throughput, tetapi juga meningkatkan konsumsi resource (CPU dan memori). Nilai optimal bergantung pada workload
dan resource yang dialokasikan ke worker cluster Connect Anda. Untuk konektor sink, jumlah partisi topik Kafka juga dapat memengaruhi paralelisme.
Kebijakan mulai ulang tugas
Anda dapat menetapkan kebijakan mulai ulang tugas konektor, yang menentukan perilaku saat terjadi kegagalan. Konektor mendukung kebijakan berikut:
Jangan pernah memulai ulang. Konektor tidak memulai ulang tugas yang gagal. Kebijakan ini adalah perilaku default. Hal ini berguna untuk proses debug, atau dalam situasi yang memerlukan intervensi manual setelah terjadi error.
Mulai ulang dengan backoff eksponensial. Konektor memulai ulang tugas yang gagal setelah penundaan (disebut periode backoff). Penundaan meningkat secara eksponensial dengan setiap kegagalan berikutnya. Kebijakan ini direkomendasikan untuk sebagian besar workload produksi.
Jika Anda menggunakan kebijakan backoff eksponensial, tetapkan juga nilai untuk backoff minimum dan maksimum. Backoff minimum harus lebih besar dari 60 detik, dan backoff maksimum harus kurang dari 7.200 detik.
Transformasi dan predikat
Managed Service untuk Apache Kafka mendukung transformasi dan predikat Kafka Connect default.
Transformasi memungkinkan Anda mengubah pesan individual sebelum dikirim ke Managed Service untuk Apache Kafka (untuk konektor sumber) atau ke sistem eksternal (untuk konektor sink). Anda dapat menggunakan transformasi untuk menyamarkan data sensitif, menambahkan stempel waktu, atau mengganti nama kolom.
Predikat memungkinkan Anda memfilter data berdasarkan kondisi tertentu, yang menentukan pesan mana yang akan diterapkan transformasi berdasarkan properti pesan.
Misalnya, untuk mengonfigurasi konektor sink agar mengabaikan pesan yang berisi kunci header DoNotProcess, tambahkan konfigurasi berikut:
transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess
Konfigurasi ini melakukan hal berikut:
Mengonfigurasi predikat bernama
hasKeydari jenisorg.apache.kafka.connect.transforms.predicates.HasHeaderKey. Predikat ini cocok dengan semua pesan yang berisi header dengan kunciDoNotProcess.Mengonfigurasi transformasi bernama
dropMessagedari jenisorg.apache.kafka.connect.transforms.Filter. Transformasi ini menghapus semua pesan yang cocok dengan predikat yang dikonfigurasi.Menautkan transformasi ke predikat
hasKey. Hal ini memastikan bahwa hanya pesan dengan kunci headerDoNotProcessyang dihapus oleh transformasi.