Bermula dengan Apache Kafka dan Python

Dalam pelajaran ini, kita akan melihat bagaimana kita dapat menggunakan Apache Kafka dengan Python dan membuat contoh aplikasi menggunakan klien Python untuk Apache Kafka.

Untuk menyelesaikan pelajaran ini, anda mesti mempunyai pemasangan aktif untuk Kafka pada mesin anda. Baca Pasang Apache Kafka di Ubuntu untuk mengetahui cara melakukannya.

Memasang klien Python untuk Apache Kafka

Sebelum kita dapat mula bekerja dengan Apache Kafka dalam program Python, kita perlu memasang klien Python untuk Apache Kafka. Ini boleh dilakukan dengan menggunakan pip (Indeks pakej Python). Berikut adalah arahan untuk mencapainya:

pip3 memasang kafka-python

Ini akan menjadi pemasangan cepat di terminal:

Pemasangan Pelanggan Python Kafka menggunakan PIP

Sekarang kita mempunyai pemasangan aktif untuk Apache Kafka dan kita juga telah memasang klien Python Kafka, kita sudah bersedia untuk memulakan pengekodan.

Menjadikan Pengeluar

Perkara pertama yang perlu diterbitkan dalam Kafka ialah aplikasi pengeluar yang boleh menghantar mesej ke topik di Kafka.

Perhatikan bahawa pengeluar Kafka adalah pengeluar mesej tidak segerak. Ini bermakna operasi yang dilakukan semasa mesej diterbitkan di partisi Kafka Topik tidak disekat. Untuk memastikan semuanya mudah, kami akan menulis penerbit JSON ringkas untuk pelajaran ini.

Untuk memulakan, buat contoh untuk Kafka Producer:

dari kafka import KafkaProducer
import json
import kertas
pengeluar = KafkaProducer (
bootstrap_servers = 'localhost: 9092',
value_serializer = lambda v: json.lambakan (v).mengekod ('utf-8'))

Atribut bootstrap_servers memberi maklumat mengenai host & port untuk pelayan Kafka. Atribut value_serializer hanya untuk tujuan penyirian JSON Nilai JSON yang dihadapi.

Untuk bermain dengan Kafka Producer, mari cuba mencetak metrik yang berkaitan dengan pengeluar dan kluster Kafka:

sukatan = pengeluar.sukatan ()
kertas cetak.cetak (metrik)

Kami akan melihat perkara berikut sekarang:

Mafik Kafka

Sekarang, mari kita cuba menghantar beberapa mesej ke Kafka Queue. Objek JSON ringkas akan menjadi contoh yang baik:

penerbit.hantar ('linuxhint', 'topic': 'kafka')

The linuxhint adalah partisi topik di mana Objek JSON akan dihantar. Semasa anda menjalankan skrip, anda tidak akan mendapat output kerana mesej hanya dihantar ke partisi topik. Sudah tiba masanya untuk menulis pengguna supaya kita dapat menguji aplikasi kita.

Menjadikan Pengguna

Sekarang, kami bersedia untuk membuat sambungan baru sebagai aplikasi Pengguna dan mendapatkan mesej dari Topik Kafka. Mulakan dengan membuat contoh baru untuk Pengguna:

dari kafka import KafkaConsumer
dari kafka import TopicPartition
cetak ('Membuat sambungan.')
pengguna = KafkaConsumer (bootstrap_servers = 'localhost: 9092')

Sekarang, tetapkan topik untuk hubungan ini dan kemungkinan nilai ofset juga.

cetak ('Menetapkan Topik.')
pengguna.menetapkan ([TopicPartition ('linuxhint', 2)])

Akhirnya, kami bersedia untuk mencetak mssage:

cetak ('Mendapatkan mesej.')
untuk pesanan dalam pengguna:
cetak ("OFFSET:" + str (message [0]) + "\ t MSG:" + str (message))

Melalui ini, kita akan mendapat senarai semua mesej yang diterbitkan di Partisi Topik Pengguna Kafka. Hasil untuk program ini adalah:

Pengguna Kafka

Sebagai rujukan ringkas, berikut adalah skrip Producer yang lengkap:

dari kafka import KafkaProducer
import json
import kertas
pengeluar = KafkaProducer (
bootstrap_servers = 'localhost: 9092',
value_serializer = lambda v: json.lambakan (v).mengekod ('utf-8'))
penerbit.hantar ('linuxhint', 'topic': 'kafka')
# metrik = pengeluar.sukatan ()
# cetakan.cetak (metrik)

Dan inilah program Pengguna lengkap yang kami gunakan:

dari kafka import KafkaConsumer
dari kafka import TopicPartition
cetak ('Membuat sambungan.')
pengguna = KafkaConsumer (bootstrap_servers = 'localhost: 9092')
cetak ('Menetapkan Topik.')
pengguna.menetapkan ([TopicPartition ('linuxhint', 2)])
cetak ('Mendapatkan mesej.')
untuk pesanan dalam pengguna:
cetak ("OFFSET:" + str (message [0]) + "\ t MSG:" + str (message))

Kesimpulannya

Dalam pelajaran ini, kami melihat bagaimana kami dapat memasang dan mula menggunakan Apache Kafka dalam program Python kami. Kami menunjukkan betapa mudahnya melakukan tugas-tugas sederhana yang berkaitan dengan Kafka di Python dengan Pelanggan Kafka yang ditunjukkan untuk Python.