Kafka Cluster Community 5.5 su RHEL o CentOS

Introduciamo kafka

Apache Kafka è una piattaforma di streaming distribuita.

In Kafka la comunicazione tra client e server avviene con un protocollo TCP agnostico di linguaggio semplice e ad alte prestazioni . Questo protocollo ha una versione e mantiene la compatibilità con le versioni precedenti. E’ presente un client Java per Kafka, ma i client sono disponibili in molti linguaggi.

Ecco una descrizione di alcuni dei più noti casi d’uso di Apache Kafka

Messaggistica: Kafka funziona bene come sostituto di un broker di messaggi più tradizionale. I broker di messaggi vengono utilizzati per una serie di motivi. Rispetto alla maggior parte dei sistemi di messaggistica, Kafka offre un throughput, una partizione, una replica e una tolleranza agli errori migliori che lo rendono una buona soluzione per le applicazioni di elaborazione dei messaggi su larga scala.

Monitoraggio dell’attività del sito Web: Il caso d’uso originale di Kafka era di poter ricostruire una pipeline di tracciamento delle attività degli utenti come un insieme di feed di sottoscrizione e pubblicazione in tempo reale. Ciò significa che l’attività del sito è pubblicata su argomenti centrali con un argomento per tipo di attività.

Metrica: Kafka viene spesso utilizzato per i dati di monitoraggio operativo. Ciò comporta l’aggregazione delle statistiche dalle applicazioni distribuite per produrre feed centralizzati di dati operativi.

Aggregazione di log: Molte persone usano Kafka in sostituzione di una soluzione di aggregazione dei registri. L’aggregazione dei log in genere raccoglie i file di log fisici dai server e li mette in una posizione centrale (un file server o forse HDFS) per l’elaborazione. Kafka estrae i dettagli dei file e fornisce un’astrazione più pulita dei dati di registro o di eventi come flusso di messaggi. Ciò consente un’elaborazione a latenza inferiore e un supporto più semplice per più origini dati e consumo di dati distribuiti.

Elaborazione del flusso: Molti utenti di Kafka elaborano i dati nell’elaborazione di pipeline costituite da più fasi, in cui i dati di input non elaborati vengono consumati dagli argomenti di Kafka e quindi aggregati, arricchiti o altrimenti trasformati in nuovi argomenti per ulteriori consumi o ulteriori elaborazioni. Ad esempio, una pipeline di elaborazione per raccomandare articoli di notizie potrebbe sottoporre a scansione il contenuto dell’articolo dai feed RSS e pubblicarlo su un argomento “articoli”; un’ulteriore elaborazione potrebbe normalizzare o deduplicare questo contenuto e pubblicare il contenuto dell’articolo pulito su un nuovo argomento.

Sourcing di eventi: Il sourcing degli eventi è uno stile di progettazione dell’applicazione in cui le modifiche di stato vengono registrate come una sequenza di record ordinata per tempo. Il supporto di Kafka per dati di registro archiviati molto grandi lo rende un eccellente backend per un’applicazione costruita in questo stile.

Salva registro: Kafka può servire come una sorta di commit-log esterno per un sistema distribuito. Il registro aiuta a replicare i dati tra i nodi e funge da meccanismo di ri-sincronizzazione per i nodi non riusciti per ripristinare i loro dati. La funzione di compattazione dei registri in Kafka aiuta a supportare questo utilizzo.

Una piattaforma di streaming ha tre elementi chiave:

• Pubblica e scrive flussi di record, similmente ad un sistema Enterprise di messaggistica aziendale.
• Archivia flussi di record in modo duraturo e tollerante ai guasti.
• Elabora flussi di record non appena si verificano.

Kafka è generalmente usato per due grandi classi di applicazioni:

• Creazione di pipeline di dati in streaming in tempo reale per ottenere dati affidabili tra sistemi o applicazioni.
• Creazione di applicazioni di streaming in tempo reale che trasformano o reagiscono ai flussi di dati.

Per capire come Kafka fa queste cose, immergiamoci ed esploriamo le capacità di Kafka dal basso verso l’alto.

Innanzitutto alcuni concetti:

• Kafka viene eseguito come un cluster su uno o più server che possono estendersi su più datacenter.
• Il cluster Kafka archivia i flussi di record in categorie denominate argomenti.
• Ogni record è costituito da una chiave, un valore e un timestamp.

Kafka ha cinque API principali:

• API Producer consente a un’applicazione di pubblicare un flusso di record su uno o più argomenti di Kafka.
• API consumer consente a un’applicazione di registrarsi ad uno o più argomenti ed elaborare il flusso di record prodotti.
• API Streams consente a un’applicazione di agire come un processore di flusso, consumando un flusso di input da uno o più argomenti e producendo un flusso di output in uno o più argomenti, trasformando efficacemente i flussi di input in flussi di output.
• API Connector consente di creare ed eseguire produttori o consumatori riutilizzabili che collegano argomenti Kafka ad applicazioni o sistemi di dati esistenti. Ad esempio, un connettore a un database relazionale potrebbe acquisire tutte le modifiche a una tabella.
• Admin API consente di gestire e ispezionare argomenti, broker e altri oggetti Kafka.

Otteniamo il software

Importiamo la chiave

$ sudo rpm --import https://packages.confluent.io/rpm/5.5/archive.key

Creiamo un file con le informazioni della repository /etc/yum.repos.d/confluent.rep e aggiungiamo il testo riportato sotto.

[Confluent.dist]
name=Confluent repository (dist)
baseurl=https://packages.confluent.io/rpm/5.5/7
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/5.5/archive.key
proxy=http://10.48.20.107:80
enabled=1
[Confluent]
name=Confluent repository
baseurl=https://packages.confluent.io/rpm/5.5
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/5.5/archive.key
proxy=http://10.48.20.107:80
enabled=1

Procediamo ad installare il nostro software community di Kafka.

$ sudo yum clean all && sudo yum install confluent-community-2.12

Se l’installazione è andata a buon fine lanciando il comando riportato sotto otterremo la stessa lista di pacchetti.

$ rpm -qa | grep confluent
confluent-rest-utils-5.5.0-1.noarch
confluent-schema-registry-5.5.0-1.noarch
confluent-kafka-2.12-5.5.0-1.noarch
confluent-common-5.5.0-1.noarch
confluent-kafka-connect-storage-common-5.5.0-1.noarch
confluent-kafka-rest-5.5.0-1.noarch
confluent-ksqldb-5.5.0-1.noarch
confluent-kafka-connect-jdbc-5.5.0-1.noarch
confluent-community-2.12-5.5.0-1.noarch
confluent-kafka-connect-s3-5.5.0-1.noarch
confluent-kafka-connect-elasticsearch-5.5.0-1.noarch
confluent-hub-client-5.5.0-1.noarch

Configurazione del cluster

Adesso andiamo a configurare la parte zookeeper editando il file /etc/kafka/zookeeper.properties con le configurazione qui sotto.

tickTime=2000
dataDir=/var/lib/zookeeper/
clientPort=2181
initLimit=5
syncLimit=2
server.1=zoo1:2888:3888
server.2=zoo2:2888:3888
server.3=zoo3:2888:3888
autopurge.snapRetainCount=3
autopurge.purgeInterval=24

Sostituire/Eliminare/Aggiungere, a seconda della nostra installazione i parametri dei server ( zoo1, zoo2, zoo3 ) con i nostri host.

Questa configurazione è per un cluster a tre nodi. Questo file di configurazione dovrebbe essere identico per tutti i nodi cluster. tickTime, dataDire clientPortsono tutti impostati sui valori tipici di un singolo server.  L’initLimit e syncLimit gestiscono quanto tempo seguenti server zookeeper possono prendere per inizializzare con l’attuale leader e per quanto tempo possono essere sincronizzati con il leader. In questa configurazione, un follower può richiedere 10000 ms per l’inizializzazione e può non essere sincronizzato per un massimo di 4000 ms in base tickTime all’impostazione impostata su 2000ms.

I server.*proprietà impostano l’appartenenza all’insieme. Il formato è:

server.<myid>=<hostname>:<leaderport>:<electionport>

In un ambiente di produzione, sono richiesti più broker. Durante l’avvio i broker si registrano in ZooKeeper per diventare membri del cluster.

Passare al file delle proprietà di Apache Kafka (/etc/kafka/server.properties) e personalizzare quanto segue:

• Connettersi allo stesso insieme di ZooKeeper impostando zookeeper.connect in tutti i nodi sullo stesso valore. Sostituisci tutte le istanze localhost con il nome host o FQDN (nome di dominio completo) del tuo nodo. Ad esempio, se il nome host è zookeeper:

zookeeper.connect=zookeeper:2181

• Configurare gli ID broker per ciascun nodo nel cluster utilizzando uno di questi metodi.
  • Genera dinamicamente gli ID broker: aggiungi broker.id.generation.enable=true e commenta broker.id. Per esempio:
  • Imposta manualmente gli ID broker: imposta un valore univoco per broker.idogni nodo.

############################# Server Basics #############################

# The ID of the broker. This must be set to a unique integer for each broker.
#broker.id=0
broker.id.generation.enable=true

Configurare il modo in cui altri broker e client comunicano con il broker utilizzando listeners e facoltativamente advertised.listeners.

• listeners: Elenco separato da virgole di URI e nomi di ascoltatori su cui ascoltare.
• advertised.listeners: Elenco separato da virgole di URL e nomi di listener da utilizzare per altri broker e client. Il advertised.listenersparametro garantisce che il broker pubblicizzi un indirizzo accessibile da host sia locali che esterni.

Passare al file delle proprietà del proxy REST Confluent (/etc/kafka-rest/kafka-rest.properties) e personalizzare quanto segue:

• Facoltativamente configurare zookeeper.connect. La connettività ZooKeeper è necessaria per gli endpoint precedenti / v1 / consumer.  Passare localhostal nome host o al nome di dominio completo (nome di dominio completo) del nodo. Ad esempio, se il nome host è zookeeper:

zookeeper.connect=zookeeper:2181

Passare al file delle proprietà dello Schema Registry (/etc/schema-registry/schema-registry.properties) e specificare le seguenti proprietà:

# Specify the address the socket server listens on, e.g. listeners = PLAINTEXT://your.host.name:9092
listeners=http://0.0.0.0:8081

# The host name advertised in ZooKeeper. This must be specified if your running Schema Registry
# with multiple nodes.
host.name=192.168.50.1

# List of Kafka brokers to connect to, e.g. PLAINTEXT://hostname:9092,SSL://hostname2:9092
kafkastore.bootstrap.servers=PLAINTEXT://hostname:9092,SSL://hostname2:9092

Questa configurazione è per un cluster multi-nodo a tre nodi. Per ulteriori informazioni, consultare Running Schema Registry in Production.

Eseguire lo start dei servizi confluent

Lo start dei servizi, va eseguito in ordine:

• confluent-zookeeper.service
• confluent-kafka.service
• confluent-kafka-connect.service
• confluent-schema-registry.service
• confluent-kafka-rest.service
• confluent-ksqldb.service

Disinstallare kafka

Eseguire questo comando per rimuovere Confluent Platform, dove si trova confluent-platform-2.12 (Confluent Platform) o confluent-community-2.12 (Confluent Platform utilizzando solo i componenti Confluent Community).

$ sudo yum autoremove <component-name>

Ad esempio, eseguire questo comando per rimuovere Confluent Community:

$ sudo yum autoremove confluent-community-2.12

L’output dovrebbe essere simile a:

Loaded plugins: fastestmirror, langpacks
Resolving Dependencies
--> Running transaction check
---> Package confluent-community-2.12.noarch 0:5.5.0-0.1.cp2 will be erased
...
Removed:
confluent-community-2.12.noarch 0:5.5.0-0.1.cp2

Dependency Removed:
confluent-common.noarch 0:5.5.0-0.1.cp2 confluent-control-center.noarch 0:5.5.0-0.1.cp2
confluent-kafka-2.12.noarch 0:5.5.0-0.1.cp2 confluent-kafka-connect-elasticsearch.noarch 0:5.5.0-0.1.cp2
confluent-kafka-connect-jdbc.noarch 0:5.5.0-0.1.cp2 confluent-kafka-connect-jms.noarch 0:5.5.0-0.1.cp2
confluent-kafka-connect-replicator.noarch 0:5.5.0-0.1.cp2 confluent-kafka-connect-s3.noarch 0:5.5.0-0.1.cp2
confluent-kafka-connect-storage-common.noarch 0:5.5.0-0.1.cp2 confluent-kafka-mqtt.noarch 0:5.5.0-0.1.cp2
confluent-kafka-rest.noarch 0:5.5.0-0.1.cp2 confluent-ksql.noarch 0:5.5.0-0.1.cp2
confluent-rebalancer.noarch 0:5.5.0-0.1.cp2 confluent-rest-utils.noarch 0:5.5.0-0.1.cp2
confluent-schema-registry.noarch 0:5.5.0-0.1.cp2

Complete!

Ogni tool in sé ha un mondo dietro, l’unico modo per saperne sempre di più è formarti, formarti e formarti costantemente. Il mondo si evolve velocemente, non farti lasciare indietro!