Come leggere i dati da Kafka con Python

Kafka è un sistema di messaggistica distribuito open source per inviare il messaggio in argomenti partizionati e diversi. Lo streaming di dati in tempo reale può essere implementato utilizzando Kafka per ricevere dati tra le applicazioni. Ha tre parti principali. Questi sono il produttore, il consumatore e gli argomenti. Il produttore viene utilizzato per inviare un messaggio a un particolare argomento e ogni messaggio è allegato con una chiave. Il consumatore è abituato a leggere un messaggio su un particolare argomento dall'insieme di partizioni. I dati ricevuti dal produttore e memorizzati sulle partizioni in base a un particolare argomento. Esistono molte librerie in Python per creare producer e consumer per costruire un sistema di messaggistica usando Kafka. In questo tutorial viene mostrato come è possibile leggere i dati di Kafka usando Python.

Prerequisito

Devi installare la libreria Python necessaria per leggere i dati da Kafka. Python3 viene utilizzato in questo tutorial per scrivere lo script di consumer e producer. Se il pacchetto pip non è stato installato prima nel tuo sistema operativo Linux, devi installare pip prima di installare la libreria Kafka per python. python3-kafka è usato in questo tutorial per leggere i dati da Kafka. Esegui il seguente comando per installare la libreria.

$ pip installa python3-kafka

Lettura di semplici dati di testo da Kafka

Diversi tipi di dati possono essere inviati dal produttore su un particolare argomento che può essere letto dal consumatore. In questa parte di questo tutorial viene mostrato come inviare e ricevere dati di testo semplici da Kafka utilizzando produttore e consumatore.

Crea un file chiamato produttore1.pi con il seguente script Python. KafkaProduttore il modulo è importato dalla libreria Kafka. L'elenco dei broker deve essere definito al momento dell'inizializzazione dell'oggetto produttore per connettersi con il server Kafka. La porta predefinita di Kafka è '9092'. L'argomento bootstrap_servers viene utilizzato per definire il nome host con la porta. 'Primo_argomento'è impostato come nome dell'argomento con il quale verrà inviato il messaggio di testo dal produttore. Successivamente, un semplice messaggio di testo, 'Ciao da Kafka'viene inviato utilizzando Spedire() metodo di KafkaProduttore all'argomento, 'Primo_argomento'.

produttore1.pi:

# Importa KafkaProducer dalla libreria Kafka
da kafka import KafkaProducer
# Definisci il server con la porta
bootstrap_servers = ['localhost:9092']
# Definisci il nome dell'argomento in cui verrà pubblicato il messaggio
topicName = 'First_Topic'
# Inizializza variabile produttore producer
producer = KafkaProducer(bootstrap_servers = bootstrap_servers)
# Pubblica testo in un argomento definito
produttore.send(topicName, b'Hello from kafka… ')
# Stampa messaggio
print("Messaggio inviato")

Crea un file chiamato consumatore1.pi con il seguente script Python. KafkaConsumatore il modulo viene importato dalla libreria Kafka per leggere i dati da Kafka. sistema modulo è usato qui per terminare lo script. Lo stesso nome host e numero di porta del produttore vengono utilizzati nello script del consumatore per leggere i dati da Kafka. Il nome dell'argomento del consumatore e del produttore deve essere lo stesso cioè 'Primo_argomento'.  Successivamente, l'oggetto consumatore viene inizializzato con i tre argomenti. Nome argomento, ID gruppo e informazioni sul server. per loop è usato qui per leggere il testo inviato dal produttore di Kafka.

consumatore1.pi:

# Importa KafkaConsumer dalla libreria Kafka
da kafka import KafkaConsumer
# Importa modulo sys
sistema di importazione
# Definisci il server con la porta
bootstrap_servers = ['localhost:9092']
# Definisci il nome dell'argomento da cui riceverà il messaggio
topicName = 'First_Topic'
# Inizializza la variabile consumatore
consumatore = KafkaConsumer (topicName, group_id ='group1',bootstrap_servers =
bootstrap_server)
# Leggi e stampa il messaggio del consumatore
per msg in consumer:
print("Nome argomento=%s,Messaggio=%s"%(msg.argomento, messaggio.valore))
# Termina lo script
sistema.Uscita()

Produzione:

Esegui il seguente comando da un terminale per eseguire lo script del produttore.

$ python3 produttore1.pi

Dopo l'invio del messaggio apparirà il seguente output.

Esegui il seguente comando da un altro terminale per eseguire lo script consumer.

$ python3 consumatore1.pi

L'output mostra il nome dell'argomento e il messaggio di testo inviato dal produttore.

Lettura di dati in formato JSON da Kafka

I dati in formato JSON possono essere inviati dal produttore Kafka e letti dal consumatore Kafka utilizzando il json modulo di Python. In questa parte di questo tutorial viene mostrato come i dati JSON possono essere serializzati e de-serializzati prima di inviare e ricevere i dati utilizzando il modulo python-kafka.

Crea uno script Python chiamato produttore2.pi con il seguente script. Un altro modulo chiamato JSON viene importato con KafkaProduttore modulo qui. value_serializer l'argomento è usato con bootstrap_servers argomento qui per inizializzare l'oggetto del produttore di Kafka. Questo argomento indica che i dati JSON verranno codificati utilizzando 'utf-8' set di caratteri al momento dell'invio. Successivamente, i dati in formato JSON vengono inviati all'argomento denominato JSONtopic.

produttore2.pi:

# Importa KafkaProducer dalla libreria Kafka
da kafka import KafkaProducer
# Importa il modulo JSON per serializzare i dati
import json
# Inizializza la variabile producer e imposta il parametro per la codifica JSON
producer = KafkaProducer(bootstrap_servers =
['localhost:9092'],value_serializer=lambda v: json.discariche (v).codifica('utf-8'))
# Invia dati in formato JSON
produttore.send('JSONtopic', 'name': 'fahmida','email':'[email protected]')
 
# Stampa messaggio
print("Messaggio inviato a JSONtopic")

Crea uno script Python chiamato consumatore2.pi con il seguente script. KafkaConsumatore, sistema e i moduli JSON vengono importati in questo script. KafkaConsumatore il modulo viene utilizzato per leggere i dati formattati JSON da Kafka. Il modulo JSON viene utilizzato per decodificare i dati JSON codificati inviati dal produttore Kafka. sistema il modulo viene utilizzato per terminare lo script. value_deserializzatore l'argomento è usato con bootstrap_servers per definire come verranno decodificati i dati JSON. Il prossimo, per loop viene utilizzato per stampare tutti i record dei consumatori e i dati JSON recuperati da Kafka.

consumatore2.pi:

# Importa KafkaConsumer dalla libreria Kafka
da kafka import KafkaConsumer
# Importa modulo sys
sistema di importazione
# Importa il modulo json per serializzare i dati
import json
# Inizializza la variabile consumer e imposta la proprietà per la decodifica JSON
consumatore = KafkaConsumer ('JSONtopic',bootstrap_servers = ['localhost:9092'],
value_deserializzatore=lambda m: json.carichi (m.decode('utf-8')))
# Leggi i dati da kafka
per il messaggio nel consumatore:
print("Registrazioni consumatori:\n")
stampa (messaggio)
print("\nLettura da dati JSON\n")
print("Nome:",messaggio[6]['nome'])
print("Email:",messaggio[6]['email'])
# Termina lo script
sistema.Uscita()

Produzione:

Esegui il seguente comando da un terminale per eseguire lo script del produttore.

$ python3 produttore2.pi

Lo script stamperà il seguente messaggio dopo aver inviato i dati JSON.

Esegui il seguente comando da un altro terminale per eseguire lo script consumer.

$ python3 consumatore2.pi

Il seguente output apparirà dopo aver eseguito lo script.

Conclusione:

I dati possono essere inviati e ricevuti in diversi formati da Kafka utilizzando python. I dati possono anche essere archiviati nel database e recuperati dal database utilizzando Kafka e python. A casa, questo tutorial aiuterà l'utente Python a iniziare a lavorare con Kafka.