So lesen Sie Daten aus Kafka mit Python

Kafka ist ein verteiltes Open-Source-Messaging-System zum Senden von Nachrichten in partitionierten und verschiedenen Themen. Echtzeit-Datenstreaming kann implementiert werden, indem Kafka verwendet wird, um Daten zwischen den Anwendungen zu empfangen. Es hat drei Hauptteile. Das sind Produzent, Konsument und Themen. Der Produzent wird verwendet, um eine Nachricht zu einem bestimmten Thema zu senden, und jede Nachricht ist mit einem Schlüssel versehen. Der Consumer wird verwendet, um eine Nachricht zu einem bestimmten Thema aus der Menge der Partitionen zu lesen. Die vom Hersteller empfangenen und auf den Partitionen gespeicherten Daten basierend auf einem bestimmten Thema. In Python gibt es viele Bibliotheken, um Producer und Consumer zu erstellen, um ein Messaging-System mit Kafka . aufzubauen. Wie die Daten von Kafka mit Python ausgelesen werden können, zeigt dieses Tutorial.

Voraussetzung

Sie müssen die erforderliche Python-Bibliothek installieren, um Daten von Kafka . zu lesen. Python3 wird in diesem Tutorial verwendet, um das Skript von Consumer und Producer zu schreiben. Wenn das pip-Paket noch nicht in Ihrem Linux-Betriebssystem installiert ist, müssen Sie pip installieren, bevor Sie die Kafka-Bibliothek für Python installieren. python3-kafka wird in diesem Tutorial verwendet, um Daten aus Kafka . auszulesen. Führen Sie den folgenden Befehl aus, um die Bibliothek zu installieren.

$ pip install python3-kafka

Lesen einfacher Textdaten aus Kafka

Vom Hersteller können verschiedene Arten von Daten zu einem bestimmten Thema gesendet werden, die vom Verbraucher gelesen werden können. Wie einfache Textdaten von Kafka mit Producer und Consumer gesendet und empfangen werden können, wird in diesem Teil dieses Tutorials gezeigt.

Erstellen Sie eine Datei mit dem Namen Produzent1.py mit dem folgenden Python-Skript. KafkaProduzent Modul wird aus der Kafka-Bibliothek importiert. Die Brokerliste muss zum Zeitpunkt der Initialisierung des Producer-Objekts definieren, um sich mit dem Kafka-Server zu verbinden. Der Standardport von Kafka ist '9092'. Das Argument bootstrap_servers wird verwendet, um den Hostnamen mit dem Port zu definieren. 'Erstes_Thema' wird als Themenname festgelegt, mit dem eine Textnachricht vom Produzenten gesendet wird. Als nächstes eine einfache Textnachricht, 'Hallo aus Kafka' wird gesendet mit senden() Methode von KafkaProduzent zum Thema, 'Erstes_Thema'.

Produzent1.py:

# Importieren Sie KafkaProducer aus der Kafka-Bibliothek
aus Kafka-Import KafkaProduzent
# Server mit Port definieren
bootstrap_servers = ['localhost:9092']
# Definieren Sie den Themennamen, in dem die Nachricht veröffentlicht wird
TopicName = 'First_Topic'
# Produzentenvariable initialisieren
Produzent = KafkaProducer(bootstrap_servers = bootstrap_servers)
# Text in einem definierten Thema veröffentlichen
Produzent.send(topicName, b'Hallo von kafka… ')
# Nachricht drucken
print("Nachricht gesendet")

Erstellen Sie eine Datei mit dem Namen Verbraucher1.py mit dem folgenden Python-Skript. KafkaVerbraucher Modul wird aus der Kafka-Bibliothek importiert, um Daten aus Kafka zu lesen. sys Modul wird hier verwendet, um das Skript zu beenden. Derselbe Hostname und Portnummer des Producers werden im Skript des Consumers verwendet, um Daten von Kafka . zu lesen. Der Themenname des Consumers und des Produzenten muss identisch sein, dh 'Erstes_Thema'.  Als nächstes wird das Consumer-Objekt mit den drei Argumenten initialisiert. Themenname, Gruppen-ID und Serverinformationen. zum loop wird hier verwendet, um den vom Kafka-Produzenten gesendeten Text zu lesen.

Verbraucher1.py:

# KafkaConsumer aus der Kafka-Bibliothek importieren
von kafka import KafkaConsumer
# sys-Modul importieren
Importsystem
# Server mit Port definieren
bootstrap_servers = ['localhost:9092']
# Definieren Sie den Themennamen, von dem die Nachricht empfangen wird
TopicName = 'First_Topic'
# Verbrauchervariable initialisieren
Consumer = KafkaConsumer (topicName, group_id ='group1',bootstrap_servers =
Bootstrap-Server)
# Nachricht vom Verbraucher lesen und drucken
für msg im Consumer:
print("Themenname=%s,Nachricht=%s"%(msg.Thema, Nachricht.Wert))
# Beende das Skript
sys.Ausfahrt()

Ausgabe:

Führen Sie den folgenden Befehl von einem Terminal aus, um das Producer-Skript auszuführen.

$ python3-Produzent1.py

Die folgende Ausgabe erscheint nach dem Senden der Nachricht.

Führen Sie den folgenden Befehl von einem anderen Terminal aus, um das Consumer-Skript auszuführen.

$ python3 Verbraucher1.py

Die Ausgabe zeigt den Themennamen und die vom Produzenten gesendete Textnachricht.

Lesen von JSON-formatierten Daten aus Kafka

JSON-formatierte Daten können vom Kafka-Produzenten gesendet und vom Kafka-Konsumenten gelesen werden mit der json Python-Modul. In diesem Teil dieses Tutorials wird gezeigt, wie JSON-Daten vor dem Senden und Empfangen der Daten mit dem python-kafka-Modul serialisiert und deserialisiert werden können.

Erstellen Sie ein Python-Skript namens Produzent2.py mit folgendem Skript. Ein weiteres Modul namens JSON wird importiert mit KafkaProduzent Modul hier. value_serializer Argument wird verwendet mit Bootstrap-Server Argument hier, um das Objekt des Kafka-Produzenten zu initialisieren. Dieses Argument gibt an, dass JSON-Daten mit 'utf-8' Zeichensatz zum Zeitpunkt des Sendens. Als Nächstes werden JSON-formatierte Daten an das Thema namens gesendet JSONthema.

Produzent2.py:

# KafkaProducer aus der Kafka-Bibliothek importieren
aus Kafka-Import KafkaProduzent
# JSON-Modul importieren, um Daten zu serialisieren
json importieren
# Produzentenvariable initialisieren und Parameter für die JSON-Kodierung festlegen
Produzent = KafkaProducer(bootstrap_servers =
['localhost:9092'],value_serializer=lambda v: json.Deponien (v).codieren('utf-8'))
# Daten im JSON-Format senden
Produzent.send('JSONtopic', 'name': 'fahmida','email':'[email protected]')
 
# Nachricht drucken
print("Nachricht an JSONtopic gesendet")

Erstellen Sie ein Python-Skript namens Verbraucher2.py mit folgendem Skript. KafkaVerbraucher, sys und JSON-Module werden in dieses Skript importiert. KafkaVerbraucher -Modul wird verwendet, um JSON-formatierte Daten aus dem Kafka . zu lesen. Das JSON-Modul wird verwendet, um die codierten JSON-Daten zu entschlüsseln, die vom Kafka-Produzenten gesendet werden. Sys Modul wird verwendet, um das Skript zu beenden. value_deserializer Argument wird verwendet mit Bootstrap-Server um zu definieren, wie JSON-Daten dekodiert werden. Nächster, zum Schleife wird verwendet, um alle von Kafka J abgerufenen Verbraucherdatensätze und JSON-Daten zu drucken.

Verbraucher2.py:

# Importieren Sie KafkaConsumer aus der Kafka-Bibliothek
von kafka import KafkaConsumer
# sys-Modul importieren
Importsystem
# Json-Modul importieren, um Daten zu serialisieren
json importieren
# Consumer-Variable initialisieren und Eigenschaft für die JSON-Decodierung festlegen
Consumer = KafkaConsumer ('JSONtopic',bootstrap_servers = ['localhost:9092'],
value_deserializer=lambda m: json.Lasten (m.dekodieren('utf-8')))
# Daten von kafka . lesen
für Nachricht im Consumer:
print("Verbraucherdatensätze:\n")
drucken (Nachricht)
print("\nLesen von JSON-Daten\n")
print("Name:",Nachricht[6]['Name'])
print("E-Mail:",Nachricht[6]['E-Mail'])
# Beende das Skript
sys.Ausfahrt()

Ausgabe:

Führen Sie den folgenden Befehl von einem Terminal aus, um das Producer-Skript auszuführen.

$ python3-Produzent2.py

Das Skript druckt nach dem Senden der JSON-Daten die folgende Nachricht.

Führen Sie den folgenden Befehl von einem anderen Terminal aus, um das Consumer-Skript auszuführen.

$ python3 Verbraucher2.py

Die folgende Ausgabe erscheint nach dem Ausführen des Skripts.

Fazit:

Die Daten können mit Python in verschiedenen Formaten von Kafka gesendet und empfangen werden. Die Daten können auch in der Datenbank gespeichert und mit Kafka und Python aus der Datenbank abgerufen werden. I home, dieses Tutorial wird dem Python-Benutzer helfen, mit Kafka zu arbeiten.