Kafka Cluster Quick­start

Übersicht

In diesem Blogpost vermitteln wir die Kernkonzepte von Apache Kafka anhand eines lokal aufgesetzten Kafka Clusters, bestehend aus drei Kafka Brokern. Bei Oepfelbaum IT Management AG haben wir ausgewiesene Kafka-Experten, die tiefgreifendes Wissen und langjährige Erfahrung mit dieser Technologie besitzen. Ob bei der Integration in bestehende Systeme, der Skalierung von Datenströmen oder der Optimierung von Kafka-Setups – unser Team ist bestens vertraut mit den vielseitigen Einsatzmöglichkeiten und Herausforderungen, die Kafka bietet.

Einführung

Apache Kafka ist nicht nur eine verteilte Streaming-Plattform, sondern eine Schlüsseltechnologie, die es Unternehmen ermöglicht, enorme Datenströme in Echtzeit zu verarbeiten. Besonders in datenintensiven Branchen wie der Finanzwelt, wo Schnelligkeit und Genauigkeit entscheidend sind, hat Kafka einen festen Platz. Banken setzen Kafka ein, um Prozesse wie Transaktionsverarbeitung, Betrugserkennung und Datenanalyse zu beschleunigen und gleichzeitig eine hohe Verfügbarkeit und Zuverlässigkeit zu gewährleisten. Dank seiner Skalierbarkeit und der Fähigkeit, grosse Datenmengen effizient zu verwalten, ist Kafka ideal geeignet, um zeitkritische Anwendungen und datengetriebene Entscheidungen zu unterstützen.

Voraussetzungen

• Docker installation

Kafka Cluster starten

3 Zookeeper Container starten:

docker run -d \
   --net=host \
   --name=zk-1 \
   -e ZOOKEEPER_SERVER_ID=1 \
   -e ZOOKEEPER_CLIENT_PORT=22181 \
   -e ZOOKEEPER_TICK_TIME=2000 \
   -e ZOOKEEPER_INIT_LIMIT=5 \
   -e ZOOKEEPER_SYNC_LIMIT=2 \
   -e ZOOKEEPER_SERVERS="localhost:22888:23888;localhost:32888:33888;localhost:42888:43888" \
   confluentinc/cp-zookeeper:7.5.0

docker run -d \
   --net=host \
   --name=zk-2 \
   -e ZOOKEEPER_SERVER_ID=2 \
   -e ZOOKEEPER_CLIENT_PORT=32181 \
   -e ZOOKEEPER_TICK_TIME=2000 \
   -e ZOOKEEPER_INIT_LIMIT=5 \
   -e ZOOKEEPER_SYNC_LIMIT=2 \
   -e ZOOKEEPER_SERVERS="localhost:22888:23888;localhost:32888:33888;localhost:42888:43888" \
   confluentinc/cp-zookeeper:7.5.0

docker run -d \
   --net=host \
   --name=zk-3 \
   -e ZOOKEEPER_SERVER_ID=3 \
   -e ZOOKEEPER_CLIENT_PORT=42181 \
   -e ZOOKEEPER_TICK_TIME=2000 \
   -e ZOOKEEPER_INIT_LIMIT=5 \
   -e ZOOKEEPER_SYNC_LIMIT=2 \
   -e ZOOKEEPER_SERVERS="localhost:22888:23888;localhost:32888:33888;localhost:42888:43888" \
   confluentinc/cp-zookeeper:7.5.0

3 Kafka Container starten:

docker run -d \
    --net=host \
    --name=kafka-1 \
    -e KAFKA_BROKER_ID=1 \
    -e KAFKA_ZOOKEEPER_CONNECT=localhost:22181,localhost:32181,localhost:42181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:19092 \
    confluentinc/cp-kafka:7.5.0

docker run -d \
    --net=host \
    --name=kafka-2 \
    -e KAFKA_BROKER_ID=2 \
    -e KAFKA_ZOOKEEPER_CONNECT=localhost:22181,localhost:32181,localhost:42181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:29092 \
    confluentinc/cp-kafka:7.5.0

 docker run -d \
     --net=host \
     --name=kafka-3 \
     -e KAFKA_BROKER_ID=3 \
     -e KAFKA_ZOOKEEPER_CONNECT=localhost:22181,localhost:32181,localhost:42181 \
     -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:39092 \
     confluentinc/cp-kafka:7.5.0

Die neuesten Versionen der Zookeeper und Kafka Images sind verfügbar auf DockerHub.

Kafka CLI starten

docker run -it --net=host --rm  confluentinc/cp-kafka:7.5.0 /bin/bash

Topic erstellen

Führe den folgenden Befehl innerhalb des Kafka-CLI aus, um ein Topic zu erstellen:

/bin/kafka-topics --bootstrap-server localhost:19092 --create --topic sensor --partitions 2 --replication-factor 3 --if-not-exists

Überprüfe mit dem Befehl „list“, ob das Topic vorhanden ist:

/bin/kafka-topics --bootstrap-server localhost:19092 --list

Überprüfe Topic Partition Leaders und In-sync-replicas mit dem Befehl “describe“:

/bin/kafka-topics --bootstrap-server localhost:19092 --describe --topic sensor

Partitioning & Replication

Partitioning teilt Topics in mehrere Partitionen auf, wodurch Messages parallel verteilt und verarbeitet werden können.

Replication sorgt dafür, dass Messages redundante Kopien haben, indem Partitionen auf mehrere Broker repliziert werden. Dies gewährleistet die Datenverfügbarkeit bei Ausfall einzelner Broker.

Das oben erstellte Topic ist aufgeteilt in 2 Partitionen, welche auf 3 Broker repliziert werden.

Messages produzieren

Führe den folgenden Befehl innerhalb des Kafka-CLI aus, um Kafka Messages auf das Topic zu schicken:

seq 30 | sed 's/\([0-9]\+\)/\1:\1/g' | /bin/kafka-console-producer --broker-list localhost:19092 --topic sensor  --property parse.key=true --property key.separator=:

Generiert eine Reihe von 30 Messages im Format key:value: 0:0, 1:1, 2:2, ...

Messages konsumieren

Nutze den Kafka Console Consumer, um die Messages zu konsumieren:

/bin/kafka-console-consumer --bootstrap-server localhost:19092,localhost:29092,localhost:39092 --from-beginning --property print.key=true --property key.separator=":" --topic sensor

Messages parallel konsumieren

Starte 2 separate Kafka-CLI, indem du den folgenden Befehl in zwei separaten Terminals ausführst:

docker run -it --net=host --rm  confluentinc/cp-kafka:7.5.0 /bin/bash

Führe den folgenden Befehl innerhalb der beiden Kafka-CLI aus, um 2 Consumer zu starten. Verwende die “--group” Option, damit beide Consumer in derselben Consumer Group starten:

/bin/kafka-console-consumer --bootstrap-server localhost:19092,localhost:29092,localhost:39092 --property print.key=true --property key.separator=":" --topic sensor --group myGroup

Beobachte die beiden Consumer, während du neue Messages produzierst.

Starte dafür ein drittes Kafka-CLI in einem separaten Terminal:

docker run -it --net=host --rm  confluentinc/cp-kafka:7.5.0 /bin/bash

Führe den folgenden Befehl aus, um Messages auf das Topic zu schicken:

seq 30 | sed 's/\([0-9]\+\)/\1:\1/g' | /bin/kafka-console-producer --broker-list localhost:19092 --topic sensor  --property parse.key=true --property key.separator=:

Generiert eine Reihe von 30 Messages im Format key:value: 0:0, 1:1, 2:2, ...

Offsets und Consumer Lag

Überprüfe die Current Offsets und Lags der beiden Consumer:

/bin/kafka-consumer-groups --bootstrap-server localhost:19092 --describe --group myGroup

Consumer Groups

Um ein Topic mit mehreren Consumern parallel zu konsumieren, müssen diese in derselben Consumer Group sein:

1. Ein Consumer konsumiert alle Partitions im Topic
2. Mehrere Consumer in derselben Consumer Group teilen sich die Partitions untereinander auf. Das Topic wird parallel konsumiert.
3. Eine Partition kann von maximal einem Consumer in derselben Consumer Group konsumiert werden.

Retention und Log Compaction

Retention und Log Compaction sind zwei Möglichkeiten, um alte Messages automatisch aus Kafka zu löschen.

Standardmässig löscht Kafka alle Messages nach 7 Tagen.

Retention

Kafka löscht Messages automatisch nach einer bestimmten Zeit oder nachdem eine bestimmte Menge Diskspace genutzt wurde.

Kafka stellt dafür die beiden Topic Configs “retention.ms” und “retention.bytes” zur Verfügung.

Log Compaction

Log Compaction ist ein Hintergrundprozess, der alte Messages aus dem Topic löscht und jeweils pro Message Key nur die neueste Message behält.

Log Compaction garantiert jedoch nicht, dass pro Key nur eine Message im Topic ist. Es kann weiterhin sein, dass mehrere Messages mit demselben Key auf dem Topic erhalten bleiben, da Compaction periodisch ausgeführt wird und nur geschlossene Log Segments von Compaction bereinigt werden.

Fazit

Dieser Blogpost hat die wichtigsten Konzepte eines Kafka Clusters vermittelt und gezeigt, wie man mit wenigen Schritten ein funktionierendes Setup lokal aufsetzt. Durch Partitioning und Replikation wird Kafka zu einer leistungsfähigen Plattform, die hohe Ausfallsicherheit und parallele Verarbeitung gewährleistet. Diese Architektur erlaubt es, Daten auch im Falle eines Broker-Ausfalls verfügbar zu halten. Zudem bietet Kafka mit Funktionen wie Retention und Log Compaction flexible Mechanismen, um die langfristige Speicherung von Nachrichten effizient zu verwalten. In Kombination machen diese Features Kafka zu einer unverzichtbaren Lösung für moderne Datenströme in Echtzeit.