Phenquestions
Untuk menyelesaikan pelajaran ini, Anda harus memiliki instalasi aktif untuk Kafka di mesin Anda. Baca Instal Apache Kafka di Ubuntu untuk mengetahui cara melakukan ini.
Menginstal klien Python untuk Apache Kafka
Sebelum kita dapat mulai bekerja dengan Apache Kafka dalam program Python, kita perlu menginstal klien Python untuk Apache Kafka. Ini dapat dilakukan dengan menggunakan pip (Indeks paket Python). Berikut adalah perintah untuk mencapai ini:
pip3 instal kafka-pythonIni akan menjadi instalasi cepat di terminal:
Instalasi Klien Python Kafka menggunakan PIP
Sekarang kami memiliki instalasi aktif untuk Apache Kafka dan kami juga telah menginstal klien Python Kafka, kami siap untuk memulai pengkodean.
Menjadi Produser
Hal pertama yang harus memublikasikan pesan di Kafka adalah aplikasi produser yang dapat mengirim pesan ke topik di Kafka.
Perhatikan bahwa produsen Kafka adalah produsen pesan asinkron. Ini berarti bahwa operasi yang dilakukan saat pesan dipublikasikan di partisi Topik Kafka adalah non-blocking. Untuk mempermudah, kami akan menulis penerbit JSON sederhana untuk pelajaran ini.
Untuk memulai, buat instance untuk Kafka Producer:
dari kafka import KafkaProducerimpor json
impor cetak
produser = KafkaProduser(
bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.pembuangan (v).enkode('utf-8'))
Atribut bootstrap_servers menginformasikan tentang host & port untuk server Kafka. Atribut value_serializer hanya untuk tujuan serialisasi JSON dari Nilai JSON yang ditemui.
Untuk bermain dengan Kafka Producer, mari kita coba mencetak metrik yang terkait dengan Producer dan cluster Kafka:
metrik = produser.metrik()cetak.cetak(metrik)
Kita akan melihat yang berikut sekarang:
Metrik Kafka
Sekarang, mari kita coba mengirim beberapa pesan ke Antrian Kafka. Objek JSON sederhana akan menjadi contoh yang baik:
produsen.kirim('linuxhint', 'topik': 'kafka')Itu linuxhint adalah partisi topik tempat Objek JSON akan dikirim. Saat Anda menjalankan skrip, Anda tidak akan mendapatkan output apa pun karena pesan baru saja dikirim ke partisi topik. Saatnya menulis konsumen sehingga kami dapat menguji aplikasi kami.
Menjadikan Konsumen
Sekarang, kami siap membuat koneksi baru sebagai aplikasi Konsumen dan menerima pesan dari Topik Kafka. Mulailah dengan membuat instance baru untuk Konsumen:
dari kafka impor KafkaConsumerdari kafka import TopicPartition
print('Membuat koneksi.')
konsumen = KafkaConsumer(bootstrap_servers='localhost:9092')
Sekarang, tetapkan topik untuk koneksi ini dan kemungkinan nilai offset juga.
print('Menetapkan Topik.')konsumen.assign([TopicPartition('linuxhint', 2)])
Akhirnya, kami siap untuk mencetak pesan:
print('Mendapatkan pesan.')untuk pesan di konsumen:
print("OFFSET: " + str(pesan[0])+ "\t MSG: " + str(pesan))
Melalui ini, kami akan mendapatkan daftar semua pesan yang dipublikasikan di Partisi Topik Konsumen Kafka. Output untuk program ini akan menjadi:
Konsumen Kafka
Sekedar referensi singkat, berikut adalah skrip Produser lengkap:
dari kafka import KafkaProducerimpor json
impor cetak
produser = KafkaProduser(
bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.pembuangan (v).enkode('utf-8'))
produsen.kirim('linuxhint', 'topik': 'kafka')
# metrik = produser.metrik()
# cetak.cetak(metrik)
Dan berikut adalah program Konsumen lengkap yang kami gunakan:
dari kafka impor KafkaConsumerdari kafka import TopicPartition
print('Membuat koneksi.')
konsumen = KafkaConsumer(bootstrap_servers='localhost:9092')
print('Menetapkan Topik.')
konsumen.assign([TopicPartition('linuxhint', 2)])
print('Mendapatkan pesan.')
untuk pesan di konsumen:
print("OFFSET: " + str(pesan[0])+ "\t MSG: " + str(pesan))
Kesimpulan
Dalam pelajaran ini, kita melihat bagaimana kita dapat menginstal dan mulai menggunakan Apache Kafka dalam program Python kita. Kami menunjukkan betapa mudahnya melakukan tugas-tugas sederhana yang terkait dengan Kafka dengan Python dengan Klien Kafka yang didemonstrasikan untuk Python.
