Kuinka lukea tietoja Kafkasta Pythonilla

Kafka on avoimen lähdekoodin hajautettu viestintäjärjestelmä, joka lähettää viestin osioituina ja eri aiheina. Reaaliaikainen tiedon suoratoisto voidaan toteuttaa käyttämällä Kafkaa tietojen vastaanottamiseen sovellusten välillä. Siinä on kolme pääosaa. Nämä ovat tuottaja, kuluttaja ja aiheet. Tuottajaa käytetään viestin lähettämiseen tiettyyn aiheeseen ja jokainen viesti liitetään avaimella. Kuluttajaa käytetään lukemaan tietystä aiheesta viesti osioiden joukosta. Tuottajalta saadut ja osioihin tietyn aiheen perusteella tallennetut tiedot. Pythonissa on monia kirjastoja, jotka luovat tuottajan ja kuluttajan rakentamaan viestijärjestelmän Kafkaa käyttäen. Kuinka Kafkan tiedot voidaan lukea pythonilla, näkyy tässä opetusohjelmassa.

Edellytys

Sinun on asennettava tarvittava python-kirjasto tietojen lukemiseksi Kafkasta. Python3: ta käytetään tässä opetusohjelmassa kuluttajan ja tuottajan komentosarjan kirjoittamiseen. Jos pip-pakettia ei ole asennettu aiemmin Linux-käyttöjärjestelmääsi, sinun on asennettava pip ennen Kafka-kirjaston asentamista pythonille. python3-kafka käytetään tässä opetusohjelmassa tietojen lukemiseen Kafkasta. Asenna kirjasto suorittamalla seuraava komento.

$ pip asenna python3-kafka

Yksinkertaisten tekstitietojen lukeminen Kafkalta

Tuottajalta voidaan lähettää erityyppisiä tietoja tietystä aiheesta, jonka kuluttaja voi lukea. Kuinka tämä yksinkertainen tekstidata voidaan lähettää ja vastaanottaa Kafkalta tuottajan ja kuluttajan avulla, on esitetty tämän opetusohjelman tässä osassa.

Luo tiedosto nimeltä tuottaja 1.py seuraavalla python-komentosarjalla. KafkaTuottaja moduuli tuodaan Kafka-kirjastosta. Välittäjäluettelon on määriteltävä tuottajaobjektin alustushetkellä yhteyden muodostamiseksi Kafka-palvelimeen. Kafkan oletusportti on '9092'. bootstrap_servers-argumenttia käytetään määrittämään isäntänimi portin kanssa. 'Ensimmäinen'on asetettu aiheen nimeksi, jolla tuottaja lähettää tekstiviestin. Seuraavaksi yksinkertainen tekstiviesti, 'Hei Kafka'lähetetään lähettää() menetelmä KafkaTuottaja aiheeseen,Ensimmäinen'.

tuottaja 1.py:

# Tuo KafkaProducer Kafkan kirjastosta
Kafka-tuottajalta KafkaProducer
# Määritä palvelin portilla
bootstrap_servers = ['paikallinen isäntä: 9092']
# Määritä aiheen nimi, missä viesti julkaistaan
topicName = 'First_Topic'
# Alusta tuottajamuuttuja
tuottaja = KafkaProducer (bootstrap_servers = bootstrap_servers)
# Julkaise teksti määritetyssä aiheessa
tuottaja.lähetä (topicName, b'Hello from kafka… ')
# Tulosta viesti
tulosta ("Viesti lähetetty")

Luo tiedosto nimeltä kuluttaja 1.py seuraavalla python-komentosarjalla. KafkaKuluttaja moduuli tuodaan Kafka-kirjastosta lukemaan tietoja Kafkasta. sys moduulia käytetään tässä lopettamaan komentosarja. Tuottajan samaa isäntänimeä ja porttinumeroa käytetään kuluttajan komentosarjassa tietojen lukemiseen Kafkasta. Kuluttajan ja tuottajan aiheen nimen on oltava sama kuinEnsimmäinen aihe'.  Seuraavaksi kuluttajaobjekti alustetaan kolmella argumentilla. Aiheen nimi, ryhmän tunnus ja palvelintiedot. varten silmukkaa käytetään tässä Kafka-tuottajan lähettämän tekstin lukemiseen.

kuluttaja 1.py:

# Tuo KafkaConsumer Kafka-kirjastosta
Kafka-tuonnista KafkaConsumer
# Tuo järjestelmän moduuli
tuoda sys
# Määritä palvelin portilla
bootstrap_servers = ['paikallinen isäntä: 9092']
# Määritä aiheen nimi, josta viesti vastaanotetaan
topicName = 'First_Topic'
# Alusta kuluttajamuuttuja
kuluttaja = KafkaConsumer (aiheenNimi, ryhmän_id = 'ryhmä1', käynnistysvaihe_palvelimet =
bootstrap_servers)
# Lue ja tulosta viesti kuluttajalta
viestin kuluttajalle:
tulosta ("Aiheen nimi =% s, Viesti =% s"% (msg.aihe, viesti.arvo))
# Lopeta komentosarja
sys.poistua ()

Tuotos:

Suorita seuraava komento yhdestä päätelaitteesta suorittaaksesi tuottajakoodin.

$ python3 -tuottaja1.py

Seuraava lähtö näkyy viestin lähettämisen jälkeen.

Suorita seuraava komento toisesta päätelaitteesta suorittaaksesi kuluttajakoodin.

$ python3-kuluttaja1.py

Lähtö näyttää aiheen nimen ja tuottajalta lähetetyn tekstiviestin.

JSON-muotoisten tietojen lukeminen Kafkalta

Kafkan tuottaja voi lähettää JSON-muotoiset tiedot ja lukea Kafkan kuluttajat Json python-moduuli. Kuinka JSON-tiedot voidaan sarjata ja poistaa sarjoista ennen tietojen lähettämistä ja vastaanottamista python-kafka-moduulin avulla, näytetään tämän opetusohjelman tässä osassa.

Luo python-komentosarja nimeltä tuottaja2.py seuraavalla komentosarjalla. Toinen moduuli nimeltä JSON tuodaan KafkaTuottaja moduuli täällä. value_serializer argumenttia käytetään bootstrap_palvelimet argumentti tässä alustaa Kafkan tuottajan kohde. Tämä argumentti osoittaa, että JSON-tiedot koodataan käyttämällä 'utf-8'-merkki lähetyshetkellä. Seuraavaksi JSON-muotoiset tiedot lähetetään nimettyyn aiheeseen JSONtopic.

tuottaja2.py:

# Tuo KafkaProducer Kafkan kirjastosta
Kafka-tuottajalta KafkaProducer
# Tuo JSON-moduuli tietojen sarjalliseksi
tuo json
# Alusta tuottajamuuttuja ja aseta parametri JSON-koodaukselle
tuottaja = KafkaProducer (bootstrap_palvelimet =
['paikallinen isäntä: 9092'], arvo_serialisaattori = lambda v: json.kaatopaikat (v).koodata ('utf-8'))
# Lähetä tiedot JSON-muodossa
tuottaja.lähetä ('JSONtopic', 'name': 'fahmida', 'email': '[email protected]')
 
# Tulosta viesti
tulosta ("Viesti lähetetty JSONtopicille")

Luo python-komentosarja nimeltä kuluttaja2.py seuraavalla komentosarjalla. KafkaKuluttaja, sys ja JSON-moduulit tuodaan tähän komentosarjaan. KafkaKuluttaja moduulia käytetään JSON-muotoisten tietojen lukemiseen Kafkasta. JSON-moduulia käytetään dekoodaamaan Kafka-tuottajalta lähetetyt koodatut JSON-tiedot. Sys moduulia käytetään lopettamaan komentosarja. value_deserializer argumenttia käytetään bootstrap_palvelimet määrittää, miten JSON-tiedot dekoodataan. Seuraava, varten -silmukkaa käytetään kaikkien Kafkasta haettujen kuluttajatietueiden ja JSON-tietojen tulostamiseen.

kuluttaja2.py:

# Tuo KafkaConsumer Kafka-kirjastosta
Kafka-tuonnista KafkaConsumer
# Tuo järjestelmän moduuli
tuoda sys
# Tuo Json-moduuli tietojen sarjalliseksi
tuo json
# Alusta kuluttajamuuttuja ja määritä ominaisuus JSON-dekoodaukselle
kuluttaja = KafkaConsumer ('JSONtopic', bootstrap_servers = ['paikallinen isäntä: 9092'],
value_deserializer = lambda m: json.kuormat (m.purkaa ('utf-8')))
# Lue tiedot kafkalta
viesti kuluttajalle:
tulosta ("Kuluttajarekisterit: \ n")
tulosta (viesti)
tulosta ("\ nLukeminen JSON-tiedoista \ n")
tulosta ("Nimi:", viesti [6] ['nimi'])
tulosta ("Sähköposti:", viesti [6] ['sähköposti'])
# Lopeta komentosarja
sys.poistua ()

Tuotos:

Suorita seuraava komento yhdestä päätelaitteesta suorittaaksesi tuottajakoodin.

$ python3 -tuottaja2.py

Komentosarja tulostaa seuraavan viestin lähetettyään JSON-tiedot.

Suorita seuraava komento toisesta päätelaitteesta suorittaaksesi kuluttajakoodin.

$ python3 kuluttaja2.py

Seuraava tulos tulee näkyviin komentosarjan suorittamisen jälkeen.

Päätelmä:

Tiedot voidaan lähettää ja vastaanottaa eri muodoissa Kafkalta pythonin avulla. Tiedot voidaan myös tallentaa tietokantaan ja noutaa tietokannasta käyttämällä Kafkaa ja Pythonia. Kotona, tämä opetusohjelma auttaa python-käyttäjää aloittamaan työskentelyn Kafkan kanssa.