Kafka Streams und ksqlDB beherrschen

Kapitel 1. Eine schnelle Einführung in Kafka

Diese Arbeit wurde mithilfe von KI übersetzt. Wir freuen uns über dein Feedback und deine Kommentare: [e-mail]

Die Datenmenge auf der Welt wächst exponentiell und laut dem Weltwirtschaftsforum übersteigt die Anzahl der gespeicherten Bytes bereits jetzt bei weitem die Anzahl der Sterne im beobachtbaren Universum.

Wenn du an diese Daten denkst, denkst du vielleicht an Haufen von Bytes, die in Data Warehouses, in relationalen Datenbanken oder auf verteilten Dateisystemen liegen. Systeme wie diese haben uns beigebracht, Daten in ihrem Ruhezustand zu sehen. Mit anderen Worten: Die Daten liegen irgendwo und ruhen, und wenn du sie verarbeiten musst, führst du eine Abfrage oder einen Auftrag gegen den Haufen Bytes aus.

Diese Sicht auf die Welt ist die traditionellere Art, über Daten nachzudenken. Daten können sich zwar an bestimmten Stellen anhäufen, aber meistens sind sie in Bewegung. Viele Systeme erzeugen kontinuierliche Datenströme, z. B. IoT-Sensoren, medizinische Sensoren, Finanzsysteme, Benutzer- und Kundenanalysesoftware, Anwendungs- und Serverprotokolle und vieles mehr. Selbst Daten, die schließlich einen schönen Platz zum Ausruhen finden, wandern wahrscheinlich irgendwann durch das Netzwerk, bevor sie ihr endgültiges Zuhause finden.

Wenn wir Daten in Echtzeit verarbeiten wollen, während sie sich bewegen, können wir nicht einfach warten, bis sie sich irgendwo stapeln, und dann eine Abfrage oder einen Auftrag in einem von uns gewählten Intervall ausführen. Dieser Ansatz eignet sich für einige geschäftliche Anwendungsfälle, aber viele wichtige Anwendungsfälle erfordern, dass wir Daten schrittweise verarbeiten, anreichern, umwandeln und auf sie reagieren, sobald sie verfügbar sind. Deshalb brauchen wir etwas, das eine ganz andere Weltsicht auf Daten hat: eine Technologie, die uns Zugang zu Daten in ihrem fließenden Zustand verschafft und die es uns ermöglicht, schnell und effizient mit diesen kontinuierlichen und unbegrenzten Datenströmen zu arbeiten. An dieser Stelle kommt Apache Kafka ins Spiel.

Apache Kafka (oder einfach Kafka) ist eine Streaming-Plattform zum Aufnehmen, Speichern, Abrufen und Verarbeiten von Datenströmen. Obwohl die gesamte Plattform sehr interessant ist, konzentriert sich dieses Buch auf den Teil von Kafka, der meiner Meinung nach am interessantesten ist: die Stream-Verarbeitungsschicht. Um Kafka Streams und ksqlDB zu verstehen (die beide auf dieser Ebene arbeiten, wobei letztere auch auf der Stream-Ingestion-Ebene arbeitet), musst du wissen, wie Kafka alsPlattform funktioniert.

Deshalb werden in diesem Kapitel einige wichtige Konzepte und Begriffe vorgestellt, die du für den Rest des Buches brauchst. Wenn du dich bereits mit Kafka auskennst, kannst du dieses Kapitel getrost überspringen. Ansonsten lies einfach weiter.

Einige der Fragen, die wir in diesem Kapitel beantworten werden, sind:

• Wie vereinfacht Kafka die Kommunikation zwischen Systemen?

• Was sind die wichtigsten Komponenten in der Architektur von Kafka?

• Welche Abstraktion der Speicherung bildet Streams am ehesten ab?

• Wie speichert Kafka Daten auf eine fehlertolerante und dauerhafte Weise?

• Wie werden Hochverfügbarkeit und Fehlertoleranz in denDatenverarbeitungsschichten erreicht?

Wir schließen dieses Kapitel mit einem Tutorial ab, das zeigt, wie man Kafka installiert und ausführt. Aber zuerst wollen wir uns das Kommunikationsmodell von Kafka ansehen.

Kommunikationsmodell

Das vielleicht häufigste Kommunikationsmuster zwischen Systemen ist das synchrone Client-Server-Modell. Wenn wir in diesem Zusammenhang von Systemen sprechen, meinen wir Anwendungen, Microservices, Datenbanken und alles andere, das Daten über ein Netzwerk liest und schreibt. Das Client-Server-Modell ist zunächst einfach und beinhaltet eine direkte Kommunikation zwischen Systemen, wie in Abbildung 1-1 dargestellt.

Abbildung 1-1. Punkt-zu-Punkt-Kommunikation ist bei einer kleinen Anzahl von Systemen einfach zu pflegen und zu verstehen

Du könntest zum Beispiel eine Anwendung haben, die synchron eine Datenbank nach bestimmten Daten abfragt, oder eine Sammlung von Microservices, die direkt miteinander kommunizieren.

Wenn jedoch mehr Systeme miteinander kommunizieren müssen, lässt sich die Punkt-zu-Punkt-Kommunikation nur noch schwer skalieren. Das Ergebnis ist ein komplexes Netz von Kommunikationswegen, das schwer zu durchschauen und zu pflegen ist. Abbildung 1-2 zeigt, wie verwirrend es selbst bei einer relativ kleinen Anzahl von Systemen werden kann.

Abbildung 1-2. Das Ergebnis der Hinzufügung weiterer Systeme ist ein komplexes Netz von Kommunikationskanälen, das schwer zu pflegen ist

Zu den Nachteilen des Client-Server-Modells gehören:

• Die Systeme sind eng miteinander gekoppelt, weil ihre Kommunikation von der Kenntnis der anderen Systeme abhängt. Das macht die Wartung und Aktualisierung dieser Systeme schwieriger, als es sein müsste.

• Die synchrone Kommunikation lässt wenig Spielraum für Fehler, da es keine Liefergarantien gibt , wenn eines der Systeme offline geht.

• Die Systeme können unterschiedliche Kommunikationsprotokolle, Skalierungsstrategien zur Bewältigung einer erhöhten Last, Strategien zur Fehlerbehandlung usw. verwenden. Infolgedessen kann es passieren, dass du mehrere Arten von Systemen zu warten hast(Software-Speziation), was der Wartbarkeit schadet und der allgemeinen Weisheit widerspricht, dass wir Anwendungen wie Vieh und nicht wie Haustiere behandeln sollten.

• Empfängersysteme können leicht überfordert sein, da sie das Tempo, mit dem neue Anfragen oder Daten eingehen, nicht kontrollieren können. Ohne einen Anfragepuffer sind sie den Launen der Anwendungen ausgeliefert, die Anfragen stellen.

• Es gibt keine klare Vorstellung davon, was zwischen diesen Systemen kommuniziert wird. Die Nomenklatur des Client-Server-Modells hat zu viel Wert auf Anfragen und Antworten gelegt und nicht genug Wert auf die Daten selbst. Bei datengesteuerten Systemen sollten die Daten im Mittelpunkt stehen.

• Kommunikation ist nicht wiederholbar. Das macht es schwierig, den Zustand eines Systems zu rekonstruieren.

Kafka vereinfacht die Kommunikation zwischen Systemen, indem es als zentraler Kommunikationsknotenpunkt fungiert (der oft mit einem zentralen Nervensystem verglichen wird), in dem Systeme Daten senden und empfangen können, ohne voneinander zu wissen. Das Kommunikationsmuster, das es implementiert, heißt Publish-Subscribe-Muster (oder einfach Pub/Sub). Das Ergebnis ist ein drastisch vereinfachtes Kommunikationsmodell, wie in Abbildung 1-3 dargestellt.

Abbildung 1-3. Kafka beseitigt die Komplexität der Punkt-zu-Punkt-Kommunikation, indem es als Kommunikationsdrehscheibe zwischen Systemen fungiert

Wenn wir das vorangegangene Diagramm weiter detaillieren, können wir die Hauptkomponenten des Kafka-Kommunikationsmodells identifizieren (siehe Abbildung 1-4).

Abbildung 1-4. Das Kafka-Kommunikationsmodell, neu gezeichnet mit mehr Details, um die Hauptkomponenten der Kafka-Plattform zu zeigen

Anstatt mehrere Systeme direkt miteinander kommunizieren zu lassen, veröffentlichen die Produzent/innen einfach ihre Daten in einem oder mehreren Themenbereichen, ohne sich darum zu kümmern, wer die Daten lesen kann.

Topics sind benannte Ströme (oder Kanäle) von zusammenhängenden Daten, die in einem Kafka-Cluster gespeichert werden. Sie erfüllen einen ähnlichen Zweck wie Tabellen in einer Datenbank (d.h., sie gruppieren zusammengehörige Daten). Sie sind jedoch nicht an ein bestimmtes Schema gebunden, sondern speichern die Rohdaten in Bytes, was sie sehr flexibel macht.¹

Konsumenten sind Prozesse, die Daten in einem oder mehreren Topics lesen (oder abonnieren). Sie kommunizieren nicht direkt mit den Produzenten, sondern hören sich die Daten eines Streams an, an dem sie interessiert sind.

Verbraucher können als Gruppe zusammenarbeiten (eine sogenannte Verbrauchergruppe), um die Arbeit von auf mehrere Prozesse zu verteilen.

Das Kommunikationsmodell von Kafka, das den Schwerpunkt auf fließende Datenströme legt, die leicht von mehreren Prozessen gelesen und beschrieben werden können, bietet mehrere Vorteile, darunter:

• Die Systeme werden entkoppelt und sind leichter zu warten, weil sie Daten produzieren und konsumieren können, ohne andere Systeme zu kennen.

• Die asynchrone Kommunikation ist mit stärkeren Liefergarantien verbunden. Wenn ein Verbraucher ausfällt, macht er einfach da weiter, wo er aufgehört hat, wenn er wieder online ist (oder, wenn mehrere Verbraucher in einer Verbrauchergruppe arbeiten, wird die Arbeit auf eines der anderen Mitglieder umverteilt).

• Die Systeme können das Kommunikationsprotokoll (für die Kommunikation mit Kafka-Clustern wird ein leistungsstarkes binäres TCP-Protokoll verwendet) sowie die Skalierungsstrategien und Fehlertoleranzmechanismen (die von den Verbrauchergruppen bestimmt werden) standardisieren. So können wir Software schreiben, die im Großen und Ganzen konsistent ist und in unseren Kopf passt.

• Die Verbraucher können die Daten in der Geschwindigkeit verarbeiten, die sie bewältigen können. Unverarbeitete Daten werden in Kafka dauerhaft und fehlertolerant gespeichert, bis der Verbraucher bereit ist, sie zu verarbeiten. Mit anderen Worten: Wenn der Stream, aus dem dein Verbraucher liest, plötzlich zu einem Feuerschlauch wird, fungiert der Kafka-Cluster als Puffer und verhindert, dass deine Verbraucher überfordert werden.

• Eine genauere Vorstellung davon, welche Daten kommuniziert werden, findest du unter in Form von Ereignissen. Ein Ereignis ist ein Teil der Daten mit einer bestimmten Struktur, die wir in "Ereignisse" besprechen werden . Der wichtigste Punkt ist, dass wir uns jetzt auf die Daten konzentrieren können, die durch unsere Streams fließen, anstatt so viel Zeit damit zu verbringen, die Kommunikationsschicht zu entwirren, wie wir es im Client-Server-Modell tun würden.

• Systeme können ihren Zustand jederzeit wiederherstellen, indem sie die Ereignisse in einem Thema wiederholen.

Ein wichtiger Unterschied zwischen dem Pub/Sub-Modell und dem Client-Server-Modell ist, dass die Kommunikation im Pub/Sub-Modell von Kafka nicht bidirektional ist. Mit anderen Worten: Streams fließen in eine Richtung. Wenn ein System Daten an ein Kafka-Topic sendet und sich darauf verlässt, dass ein anderes System etwas mit den Daten macht (d.h. sie anreichert oder umwandelt), müssen die angereicherten Daten in ein anderes Topic geschrieben und anschließend von dem ursprünglichen Prozess konsumiert werden. Das ist einfach zu koordinieren, aber es verändert die Art, wie wir über Kommunikation denken.

Solange du dich daran erinnerst, dass die Kommunikationskanäle (Topics) wie Ströme funktionieren (d.h. unidirektional fließen und mehrere Quellen und mehrere nachgelagerte Verbraucher haben können), ist es einfach, Systeme zu entwerfen, die einfach auf den Strom der fließenden Bytes hören, an denen sie interessiert sind, und Daten in Topics (benannte Streams) produzieren, wenn sie Daten mit einem oder mehreren Systemen austauschen wollen. In den folgenden Kapiteln werden wir viel mit Kafka-Topics arbeiten (jede Kafka-Streams- und ksqlDB-Anwendung, die wir erstellen, wird ein oder mehrere Kafka-Topics lesen und in der Regel auch in diese schreiben), so dass dir das am Ende dieses Buches zur zweiten Natur geworden sein wird.

Nachdem wir nun gesehen haben, wie das Kommunikationsmodell von Kafka die Art und Weise vereinfacht, wie Systeme miteinander kommunizieren, und dass benannte Streams, sogenannte Topics, als Kommunikationsmedium zwischen den Systemen fungieren, wollen wir nun ein tieferes Verständnis dafür entwickeln, wie Streams in Kafkas Speicherung ins Spiel kommen.

Wie werden die Streams gespeichert?

Als ein Team von LinkedIn-Ingenieuren² das Potenzial einer Stream-gesteuerten Datenplattform erkannte, mussten sie eine wichtige Frage beantworten: Wie sollten unbegrenzte und kontinuierliche Datenströme auf der Ebene der Speicherung modelliert werden?

Die von ihnen identifizierte Abstraktion für die Speicherung war bereits in vielen Datensystemen vorhanden, darunter traditionelle Datenbanken, Key-Value-Stores, Versionskontrollsysteme und mehr. Diese Abstraktion ist das einfache, aber leistungsstarke Commit-Log (oder einfach Log).

Hinweis

Wenn wir in diesem Buch von Protokollen sprechen, meinen wir nicht die Anwendungsprotokolle, die Informationen über einen laufenden Prozess ausgeben (z. B. HTTP-Serverprotokolle). Stattdessen beziehen wir uns auf eine bestimmte Datenstruktur, die in den folgenden Abschnitten beschrieben wird.

Logs sind reine Datenstrukturen, die eine geordnete Abfolge von Ereignissen festhalten. Schauen wir uns die kursiv gedruckten Attribute genauer an und machen uns ein Bild von denProtokollen, indem wir ein einfaches Protokoll über die Befehlszeile erstellen. Erstellen wir zum Beispiel ein Log mit dem Namen user_purchases und füllen es mit dem folgendenBefehl mit einigen Dummy-Daten auf:

# create the logfile
touch users.log

# generate four dummy records in our log
echo "timestamp=1597373669,user_id=1,purchases=1" >> users.log
echo "timestamp=1597373669,user_id=2,purchases=1" >> users.log
echo "timestamp=1597373669,user_id=3,purchases=1" >> users.log
echo "timestamp=1597373669,user_id=4,purchases=1" >> users.log

Wenn wir uns nun das von uns erstellte Protokoll ansehen, enthält es vier Benutzer, die einen einzigen Kauf getätigt haben:

# print the contents of the log
cat users.log

# output
timestamp=1597373669,user_id=1,purchases=1
timestamp=1597373669,user_id=2,purchases=1
timestamp=1597373669,user_id=3,purchases=1
timestamp=1597373669,user_id=4,purchases=1

Das erste Merkmal von Logs ist, dass sie nur angehängt werden. Das bedeutet, dass wir, wenn user_id=1 einen zweiten Kauf tätigt, den ersten Datensatz nicht aktualisieren, da jeder Datensatz in einem Log unveränderlich ist. Stattdessen fügen wir den neuen Datensatz einfach an das Ende an:

# append a new record to the log
echo "timestamp=1597374265,user_id=1,purchases=2" >> users.log

# print the contents of the log
cat users.log

# output
timestamp=1597373669,user_id=1,purchases=1 
timestamp=1597373669,user_id=2,purchases=1
timestamp=1597373669,user_id=3,purchases=1
timestamp=1597373669,user_id=4,purchases=1
timestamp=1597374265,user_id=1,purchases=2 

Sobald ein Datensatz in das Protokoll geschrieben wurde, gilt er als unveränderlich. Wenn wir also eine Aktualisierung vornehmen müssen (z. B. um die Anzahl der Käufe eines Nutzers zu ändern), bleibt der ursprüngliche Datensatz unangetastet.

Um die Aktualisierung zu modellieren, fügen wir den neuen Wert einfach an das Ende des Protokolls an. Das Protokoll enthält sowohl den alten als auch den neuen Datensatz, die beide unveränderlich sind.

Jedes System, das die Anzahl der Einkäufe jedes Nutzers überprüfen möchte, kann einfach jeden Datensatz im Protokoll der Reihe nach lesen, und der letzte Datensatz, den es für user_id=1 sieht, enthält den aktualisierten Kaufbetrag. Das bringt uns zur zweiten Eigenschaft von Protokollen: Sie sind geordnet.

Das vorangegangene Protokoll ist zufällig in Zeitstempelreihenfolge (siehe erste Spalte), aber das ist nicht das, was wir mit geordnet meinen. Tatsächlich speichert Kafka für jeden Datensatz im Protokoll einen Zeitstempel, aber die Datensätze müssen nicht in der Reihenfolge der Zeitstempel stehen. Wenn wir sagen, dass ein Protokoll geordnet ist, meinen wir damit, dass die Position eines Datensatzes im Protokoll feststeht und sich nie ändert. Wenn wir das Protokoll noch einmal ausdrucken, diesmal mit Zeilennummern, kannst du die Position in der ersten Spalte sehen:

# print the contents of the log, with line numbers
cat -n users.log

# output
1	timestamp=1597373669,user_id=1,purchases=1
2	timestamp=1597373669,user_id=2,purchases=1
3	timestamp=1597373669,user_id=3,purchases=1
4	timestamp=1597373669,user_id=4,purchases=1
5	timestamp=1597374265,user_id=1,purchases=2

Stell dir nun ein Szenario vor, in dem die Bestellung nicht garantiert werden kann. Mehrere Prozesse könnten die Aktualisierungen von user_id=1 in unterschiedlicher Reihenfolge lesen, so dass es zu Unstimmigkeiten über die tatsächliche Anzahl der Einkäufe dieses Nutzers kommt. Indem wir sicherstellen, dass die Protokolle geordnet sind, können die Daten deterministisch³ von mehreren Prozessen verarbeitet werden.⁴

Während die Position jedes Protokolleintrags im vorangegangenen Beispiel mit Zeilennummern angegeben wird, bezeichnet Kafka die Position jedes Eintrags in seinem verteilten Protokoll als Offset. Offsets beginnen bei 0 und ermöglichen ein wichtiges Verhalten: Sie ermöglichen es mehreren Verbrauchergruppen, jeweils aus demselben Log zu lesen und ihre eigene Position in dem Log/Stream, aus dem sie lesen, zu behalten. Dies ist in Abbildung 1-5 dargestellt.

Nachdem wir nun ein Gefühl für die logbasierte Speicherung von Kafka bekommen haben, indem wir unser eigenes Log über die Kommandozeile erstellt haben, wollen wir diese Ideen mit den übergeordneten Konstrukten verbinden, die wir im Kommunikationsmodell von Kafka identifiziert haben. Wir beginnen damit, unsere Diskussion über Topics fortzusetzen und etwas über sogenannte Partitionen zu lernen.

Abbildung 1-5. Mehrere Verbrauchergruppen können aus demselben Log lesen, wobei jede ihre Position auf der Grundlage des von ihr gelesenen/verarbeiteten Offsets beibehält

Themen und Partitionen

In unserer Diskussion über das Kommunikationsmodell von Kafka haben wir gelernt, dass Kafka das Konzept der benannten Streams, der Topics, hat. Außerdem sind Kafka-Topics extrem flexibel in Bezug auf das, was du in ihnen speicherst. Du kannst zum Beispiel homogene Topics haben, die nur eine Art von Daten enthalten, oder heterogene Topics, die mehrere Arten von Daten enthalten.⁵ Eine Darstellung dieser verschiedenen Strategien ist in Abbildung 1-6 zu sehen.

Abbildung 1-6. Es gibt verschiedene Strategien für die Speicherung von Ereignissen in Themen; homogene Themen enthalten in der Regel einen Ereignistyp (z. B. clicks), während heterogene Themen mehrere Ereignistypen enthalten (z. B. clicks und page_views)

Wir haben auch gelernt, dass Append-Only-Commit-Logs verwendet werden, um Streams in der Speicherung von Kafka zu modellieren. Heißt das also, dass jedes Topic mit einer Logdatei korreliert? Nicht ganz. Kafka ist ein verteiltes Protokoll, und es ist schwer, nur ein Exemplar von etwas zu verteilen. Wenn wir also ein gewisses Maß an Parallelität bei der Verteilung und Verarbeitung von Logs erreichen wollen, müssen wir viele von ihnen erstellen. Aus diesem Grund werden Kafka-Themen in kleinere Einheiten unterteilt, die Partitionen genannt werden.

Partitionen sind einzelne Logs (d. h. die Datenstrukturen, die wir im vorherigen Abschnitt besprochen haben), in denen Daten produziert und konsumiert werden. Da die Abstraktion des Commit-Logs auf der Partitionsebene implementiert ist, ist dies die Ebene, auf der die Ordnung garantiert wird, da jede Partition ihre eigene Reihe von Offsets hat. Auf der Themenebene wird keine globale Ordnung unterstützt, weshalb Produzenten zusammengehörige Datensätze oft an dieselbe Partition weiterleiten.⁶

Im Idealfall sind die Daten relativ gleichmäßig über alle Partitionen in einem Thema verteilt. Es kann aber auch vorkommen, dass die Partitionen unterschiedlich groß sind. Abbildung 1-7 zeigt ein Beispiel für ein Thema mit drei verschiedenen Partitionen.

Abbildung 1-7. Ein Kafka-Thema, das mit drei Partitionen konfiguriert ist

Die Anzahl der Partitionen für ein bestimmtes Thema ist konfigurierbar, und mehr Partitionen in einem Thema bedeuten in der Regel mehr Parallelität und Durchsatz, obwohl es einige Kompromisse gibt, wenn man zu viele Partitionen hat.⁷ Wir werden im Laufe des Buches noch mehr darüber sprechen, aber das Wichtigste ist, dass nur ein Verbraucher pro Verbrauchergruppe von einer Partition konsumieren kann (einzelne Mitglieder verschiedener Verbrauchergruppen können jedoch von der gleichen Partition konsumieren, wie in Abbildung 1-5 gezeigt).

Wenn du also die Verarbeitungslast auf N Verbraucher in einer einzigen Verbrauchergruppe verteilen willst, brauchst du N Partitionen. Wenn du weniger Mitglieder in einer Verbrauchergruppe hast, als es Partitionen im Quellthema (d.h. dem Thema, aus dem gelesen wird) gibt, ist das in Ordnung: Jeder Verbraucher kann mehrere Partitionen verarbeiten. Wenn du mehr Mitglieder in einer Verbrauchergruppe hast, als es Partitionen im Quellthema gibt, werden einige Verbraucher im Leerlauf sein.

Vor diesem Hintergrund können wir unsere Definition eines Topics verbessern. Ein Topic ist ein benannter Stream, der aus mehreren Partitionen besteht. Jede Partition wird als ein Commit-Log modelliert, das Daten in einer vollständig geordneten und nur anhängenden Reihenfolge speichert. Was genau wird also in einer Topic-Partition gespeichert? Das werden wir im nächsten Abschnitt untersuchen.

Veranstaltungen

In diesem Buch verbringen wir viel Zeit damit, über die Verarbeitung von Daten in Topics zu sprechen. Wir haben jedoch noch nicht vollständig verstanden, welche Art von Daten in einem Kafka-Topic (und genauer gesagt in den Partitionen eines Topics) gespeichert werden.

In der Literatur zu Kafka, einschließlich der offiziellen Dokumentation, wird eine Vielzahl von Begriffen verwendet, um die Daten in einem Topic zu beschreiben, darunter Nachrichten, Datensätze und Ereignisse. Diese Begriffe werden oft synonym verwendet, aber in diesem Buch bevorzugen wir den Begriff " Ereignis"(auch wenn wir die anderen Begriffe gelegentlich verwenden). Ein Ereignis ist ein Schlüssel-Wert-Paar mit Zeitstempel, das ein bestimmtes Ereignis aufzeichnet. Die grundlegende Anatomie jedes Ereignisses, das in einer Topic-Partition erfasst wird, ist in Abbildung 1-8 dargestellt.

Abbildung 1-8. Anatomie eines Ereignisses, das in Themenpartitionen gespeichert wird

Kopfzeilen auf Anwendungsebene enthalten optionale Metadaten über ein Ereignis. In diesem Buch arbeiten wir nicht sehr oft mit ihnen.

Schlüssel sind ebenfalls optional, spielen aber eine wichtige Rolle bei der Verteilung der Daten auf die Partitionen. Wir werden das in den nächsten Kapiteln sehen, aber im Allgemeinen werden sie verwendet, um zusammengehörige Datensätze zu identifizieren.

Jedes Ereignis ist mit einem Zeitstempel verbunden. In Kapitel 5 werden wir mehr über Zeitstempel erfahren.

Der Wert enthält den eigentlichen Inhalt der Nachricht, kodiert als Byte-Array. Die Kunden müssen die rohen Bytes in eine aussagekräftigere Struktur deserialisieren (z. B. ein JSON-Objekt oder einen Avro-Datensatz). Die Deserialisierung von Byte-Arrays wird im Abschnitt "Serialisierung/Deserialisierung" ausführlich behandelt .

Nachdem wir nun wissen, welche Daten in einem Topic gespeichert werden, wollen wir uns das Cluster-Deployment-Modell von Kafka genauer ansehen. Hier erfährst du mehr darüber, wie die Daten in Kafka physisch gespeichert werden.

Kafka Cluster und Makler

Ein zentraler Kommunikationspunkt bedeutet, dass Zuverlässigkeit und Fehlertoleranz extrem wichtig sind. Das bedeutet auch, dass das Kommunikations-Backbone skalierbar sein muss, d.h. es muss in der Lage sein, eine höhere Last zu bewältigen. Aus diesem Grund wird Kafka als Cluster betrieben, und mehrere Rechner, sogenannte Broker, sind an der Speicherung und dem Abruf von Daten beteiligt.

Kafka-Cluster können ziemlich groß sein und sich sogar über mehrere Rechenzentren und geografische Regionen erstrecken. In diesem Buch werden wir jedoch in der Regel mit einem Kafka-Cluster mit nur einem Knoten arbeiten, da dies alles ist, was wir brauchen, um mit Kafka-Streams und ksqlDB zu arbeiten. In der Produktion wirst du wahrscheinlich mindestens drei Broker benötigen und die Replikation deines Kafka-Topics so einstellen, dass deine Daten über mehrere Broker repliziert werden (das werden wir später im Tutorial dieses Kapitels sehen). So erreichen wir eine hohe Verfügbarkeit und vermeiden Datenverluste, falls ein Rechner ausfällt.

Wenn wir über die Speicherung und Replikation von Daten über verschiedene Broker hinweg sprechen, meinen wir eigentlich einzelne Partitionen in einem Thema. Ein Thema kann zum Beispiel drei Partitionen haben, die auf drei Broker verteilt sind, wie in Abbildung 1-9 dargestellt.

Abbildung 1-9. Die Partitionen sind über die verfügbaren Broker verteilt, was bedeutet, dass ein Thema mehrere Maschinen im Kafka-Cluster umfassen kann

Wie du siehst, können Topics auf diese Weise ziemlich groß werden, da sie über die Kapazität eines einzelnen Rechners hinauswachsen können. Um Fehlertoleranz und hohe Verfügbarkeit zu erreichen, kannst du bei der Konfiguration des Themas einen Replikationsfaktor festlegen. Mit einem Replikationsfaktor von 2 kann die Partition zum Beispiel auf zwei verschiedenen Brokern gespeichert werden. Dies ist in Abbildung 1-10 dargestellt.

Abbildung 1-10. Wenn du den Replikationsfaktor auf 2 erhöhst, werden die Partitionen auf zwei verschiedenen Brokern gespeichert

Wenn eine Partition auf mehrere Broker repliziert wird, wird ein Broker zum Leader ernannt, d.h. er bearbeitet alle Lese-/Schreibanfragen von Produzenten/Konsumenten für die jeweilige Partition. Die anderen Broker, die die replizierten Partitionen enthalten, werden Follower genannt und kopieren einfach die Daten des Leaders. Wenn der Leader fehlschlägt, wird einer der Follower zum neuen Leader ernannt.

Wenn die Belastung deines Clusters mit der Zeit zunimmt, kannst du deinen Cluster erweitern, indem du noch mehr Broker hinzufügst und eine Neuzuordnung der Partitionen auslöst. So kannst du die Daten von den alten Maschinen auf eine neue Maschine migrieren.

Schließlich spielen die Makler auch eine wichtige Rolle bei der Aufrechterhaltung der Mitgliedschaft in Verbrauchergruppen. Das werden wir im nächsten Abschnitt untersuchen.

Verbrauchergruppen

Kafka ist für hohen Durchsatz und niedrige Latenzzeiten optimiert. Um dies auf der Verbraucherseite auszunutzen, müssen wir die Arbeit über mehrere Prozesse hinweg parallelisieren können. Dies wird mit Verbrauchergruppen erreicht.

Verbrauchergruppen bestehen aus mehreren kooperierenden Verbrauchern, und die Zusammensetzung dieser Gruppen kann sich im Laufe der Zeit ändern. Zum Beispiel können neue Verbraucher online gehen, um die Verarbeitungslast zu erhöhen, und Verbraucher können auch offline gehen, entweder wegen geplanter Wartungsarbeiten oder wegen eines unerwarteten Ausfalls. Deshalb braucht Kafka eine Möglichkeit, die Mitgliedschaft in jeder Gruppe aufrechtzuerhalten und die Arbeit bei Bedarf umzuverteilen.

Um dies zu erleichtern, wird jede Verbrauchergruppe einem speziellen Broker, dem Gruppenkoordinator, zugewiesen, der dafür verantwortlich ist, Heartbeats von den Verbrauchern zu empfangen und eine Neuverteilung der Arbeit auszulösen, wenn ein Verbraucher als tot markiert wird. Abbildung 1-11 zeigt, wie die Verbraucher ihre Heartbeats an den Gruppenkoordinator zurücksenden.

Abbildung 1-11. Drei Verbraucher in einer Gruppe, Herzschlag zurück zum Gruppenkoordinator

Jedes aktive Mitglied der Verbrauchergruppe hat das Recht, eine Aufteilung zu erhalten. Die Verteilung der Arbeit auf drei gesunde Verbraucher kann zum Beispiel wie in Abbildung 1-12 dargestellt aussehen.

Abbildung 1-12. Drei gesunde Konsumenten teilen sich die Lese-/Verarbeitungslast eines dreigeteilten Kafka-Topics

Wenn jedoch eine Consumer-Instanz ungesund wird und keinen Heartbeat zurück zum Cluster hat, wird die Arbeit automatisch den gesunden Consumern zugewiesen. In Abbildung 1-13 zum Beispiel wurde dem mittleren Verbraucher die Partition zugewiesen, die zuvor von dem ungesunden Verbraucher bearbeitet wurde.

Abbildung 1-13. Arbeit wird umverteilt, wenn Verbraucherprozesse fehlschlagen

Wie du siehst, sind Verbrauchergruppen extrem wichtig, um hohe Verfügbarkeit und Fehlertoleranz in der Datenverarbeitungsschicht zu erreichen. Beginnen wir also mit unserem Tutorial und lernen, wie man Kafka installiert.

Kafka installieren

Eine detaillierte Anleitung zur manuellen Installation von Kafka findest du unter in der offiziellen Dokumentation. Um die Dinge jedoch so einfach wie möglich zu halten, verwendendie meisten Tutorials in diesem Buch Docker, mit dem wir Kafka und unsere Stream-Processing-Anwendungen in einerContainer-Umgebungeinsetzen können.

Deshalb werden wir Kafka mit Docker Compose installieren und Docker-Images verwenden, die von Confluent veröffentlicht werden.⁸ Der erste Schritt ist der Download und die Installation von Docker von der Docker-Installationsseite.

Als nächstes speicherst du die folgende Konfiguration in einer Datei namens docker-compose.yml:

---
version: '2'

services:
  zookeeper: 
    image: confluentinc/cp-zookeeper:6.0.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka: 
    image: confluentinc/cp-enterprise-kafka:6.0.0
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: |
         PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: |
         PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

Der erste Container, den wir zookeeper nennen, wird die ZooKeeper-Installation enthalten. Wir haben in dieser Einführung nicht über ZooKeeper gesprochen, da er zum Zeitpunkt der Erstellung dieses Artikels aktiv aus Kafka entfernt wird. Er ist jedoch ein zentraler Dienst zur Speicherung von Metadaten wie der Topic-Konfiguration. Bald wird er nicht mehr in Kafka enthalten sein, aber wir erwähnen ihn hier, da dieses Buch veröffentlicht wurde, bevor ZooKeeper vollständig entfernt wurde.

Der zweite Container, kafka, enthält die Kafka-Installation. Hier wird unser Broker (der unseren Single-Node-Cluster umfasst) laufen und wir werden einige der Kafka-Konsolenskripte für die Interaktion mit dem Cluster ausführen.

Zum Schluss führst du den folgenden Befehl aus, um einen lokalen Kafka-Cluster zu starten:

docker-compose up

Da unser Kafka-Cluster nun läuft, können wir mit unserem Tutorial fortfahren.

Hallo, Kafka

In diesem einfachen Tutorial zeigen wir dir, wie du ein Kafka-Topic erstellst, Daten mit einem Producer in ein Topic schreibst und schließlich mit einem Consumer Daten aus einem Topic liest. Als Erstes müssen wir uns in dem Container anmelden, in dem Kafka installiert ist. Dazu führen wir den folgenden Befehl aus:

docker-compose exec kafka bash

Jetzt erstellen wir ein Thema mit dem Namen users. Dazu verwenden wir eines der Konsolenskripte (kafka-topics), die in Kafka enthalten sind. Der folgende Befehl zeigt, wie das geht:

kafka-topics \ 
    --bootstrap-server localhost:9092 \ 
    --create \ 
    --topic users \ 
    --partitions 4 \ 
    --replication-factor 1 

# output
Created topic users.

kafka-topics ist ein Konsolenskript, das mit Kafka mitgeliefert wird.

Ein Bootstrap-Server ist das Host/IP-Paar für einen oder mehrere Makler.

Es gibt viele Flags für die Interaktion mit Kafka-Themen, darunter --list,--describeund --delete. Hier verwenden wir das Flag --create, da wir ein neues Thema erstellen.

Der Name des Themas ist users.

Teile unser Thema in vier Abschnitte auf.

Da wir einen Single-Node-Cluster betreiben, setzen wir den Replikationsfaktor auf 1. In der Produktion solltest du diesen Wert auf einen höheren Wert (z.B. 3) setzen, um einehohe Verfügbarkeit zu gewährleisten.

Hinweis

Die Konsolenskripte, die wir in diesem Abschnitt verwenden, sind in der Kafka-Quelldistribution enthalten. Bei einer normalen Kafka-Installation enthalten diese Skripte die Dateierweiterung .sh (z. B. kafka-topics.sh, kafka-console-producer.sh usw.). In Confluent Platform entfällt diese Dateierweiterung jedoch (deshalb haben wir im vorherigen Codeausschnitt kafka-topics statt kafka-topics.sh ausgeführt).

Sobald das Thema erstellt wurde, kannst du mit dem folgenden Befehl eine Beschreibung des Themas einschließlich seiner Konfiguration drucken:

kafka-topics \
    --bootstrap-server localhost:9092 \
    --describe \ 
    --topic users

# output
Topic: users	PartitionCount: 4	ReplicationFactor: 1	Configs:
	Topic: users	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: users	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: users	Partition: 2	Leader: 1	Replicas: 1	Isr: 1
	Topic: users	Partition: 3	Leader: 1	Replicas: 1	Isr: 1

Mit dem Flag --describe können wir Konfigurationsinformationen für ein bestimmtes Thema anzeigen.

Jetzt wollen wir mit dem eingebauten kafka-console-producer Skript einige Daten erzeugen:

kafka-console-producer \ 
    --bootstrap-server localhost:9092 \
    --property key.separator=, \ 
    --property parse.key=true \
    --topic users

Das kafka-console-producer Skript, das mit Kafka mitgeliefert wird, kann verwendet werden, um Daten für ein Topic zu produzieren. Sobald wir jedoch mit Kafka Streams und ksqlDB arbeiten, werden die Producer-Prozesse in die zugrunde liegende Java-Bibliothek eingebettet, sodass wir dieses Skript außerhalb von Test- und Entwicklungszwecken nicht mehr benötigen.

Wir werden eine Reihe von Schlüssel-Werte-Paaren für unser Thema users erstellen. Diese Eigenschaft besagt, dass unsere Schlüssel und Werte durch das Zeichen , getrennt werden.

Der vorherige Befehl führt dich zu einer interaktiven Eingabeaufforderung. Von hier aus können wir mehrere Schlüssel-Wert-Paare eingeben, um das Thema users zu erstellen. Wenn du fertig bist, drücke Control-C auf deiner Tastatur, um die Eingabeaufforderung zu verlassen:

>1,mitch
>2,elyse
>3,isabelle
>4,sammy

Nachdem wir die Daten für unser Thema erstellt haben, können wir das Skript kafka-console-consumer verwenden, um die Daten zu lesen. Der folgende Befehl zeigt, wie das geht:

kafka-console-consumer \ 
    --bootstrap-server localhost:9092 \
    --topic users \
    --from-beginning 

# output
mitch
elyse
isabelle
sammy

Das Skript kafka-console-consumer ist ebenfalls in der Kafka-Distribution enthalten. Ähnlich wie beim kafka-console-producer Skript werden die meisten Tutorials in diesem Buch die in Kafka Streams und ksqlDB integrierten Consumer-Prozesse nutzen, anstatt dieses eigenständige Konsolenskript zu verwenden (das zu Testzwecken nützlich ist).

Das --from-beginning Flag zeigt an, dass wir mit dem Konsumieren am Anfang des Kafka-Topics beginnen sollten.

Standardmäßig wird auf kafka-console-consumer nur der Wert der Nachricht ausgegeben. Aber wie wir bereits gelernt haben, enthalten Ereignisse mehr Informationen, z. B. einen Schlüssel, einen Zeitstempel und Kopfzeilen. Übergeben wir dem Konsolenverbraucher einige zusätzliche Eigenschaften, damit wir auch den Zeitstempel und die Schlüsselwerte sehen können:⁹

kafka-console-consumer \
    --bootstrap-server localhost:9092 \
    --topic users \
    --property print.timestamp=true \
    --property print.key=true \
    --property print.value=true \
    --from-beginning

# output
CreateTime:1598226962606	1	mitch
CreateTime:1598226964342	2	elyse
CreateTime:1598226966732	3	isabelle
CreateTime:1598226968731	4	sammy

Das war's! Du hast nun gelernt, wie du einige grundlegende Interaktionen miteinem Kafka-Cluster durchführst. Der letzte Schritt besteht darin, unseren lokalen Cluster mit folgendemBefehl abzubauen:

docker-compose down

Zusammenfassung

Das Kommunikationsmodell von Kafka macht es mehreren Systemen leicht, miteinander zu kommunizieren, und seine schnelle, dauerhafte und nur anhängende Speicherung ermöglicht es, mit sich schnell bewegenden Datenströmen zu arbeiten. Durch den Einsatz eines Clusters kann Kafka eine hohe Verfügbarkeit und Fehlertoleranz auf der Speicherebene erreichen, indem die Daten über mehrere Rechner, sogenannte Broker, repliziert werden. Darüber hinaus ermöglicht die Fähigkeit des Clusters, Heartbeats von Verbraucherprozessen zu empfangen und die Mitgliedschaft in Verbrauchergruppen zu aktualisieren, eine hohe Verfügbarkeit, Fehlertoleranz und Skalierbarkeit der Arbeitslast auf der Streamverarbeitungs- und Verbraucherebene. All diese Funktionen haben Kafka zu einer der beliebtesten Streamverarbeitungsplattformen überhaupt gemacht.

Du hast jetzt genug Hintergrundwissen über Kafka, um mit Kafka Streams und ksqlDB loszulegen. Im nächsten Abschnitt erfahren wir, wie Kafka Streams in das Kafka-Ökosystem passt und wie wir diese Bibliothek nutzen können, um mit Daten auf der Stream-Verarbeitungsebene zu arbeiten.

¹ Wir sprechen in Kapitel 3 über die rohen Byte-Arrays, die in Topics gespeichert werden, sowie über den Prozess der Deserialisierung der Bytes in höherwertige Strukturen wie JSON-Objekte/Avro-Datensätze.

² Jay Kreps, Neha Narkhede und Jun Rao leiteten zunächst die Entwicklung von Kafka.

³ Deterministisch bedeutet, dass dieselben Inputs dieselben Outputs erzeugen.

⁴ Aus diesem Grund verwenden traditionelle Datenbanken Logs für die Replikation. Logs werden verwendet, um jeden Schreibvorgang in der führenden Datenbank zu erfassen und dieselben Schreibvorgänge der Reihe nach in einer Replikat-Datenbank zu verarbeiten, um denselben Datensatz auf einer anderen Maschine deterministisch neu zu erstellen.

⁵ Martin Kleppmann hat einen interessanten Artikel zu diesem Thema verfasst, den du unter https://oreil.ly/tDZMm finden kannst. Er spricht über die verschiedenen Kompromisse und die Gründe, warum man sich für die eine oder andere Strategie entscheiden sollte. Der Folgeartikel von Robert Yokota geht außerdem näher darauf ein, wie man mehrere Ereignistypen unterstützt, wenn man Confluent Schema Registry für die Schemaverwaltung/-entwicklung verwendet.

⁶ Die Partitionierungsstrategie ist konfigurierbar, aber eine beliebte Strategie, die auch in Kafka Streams und ksqlDB implementiert ist, besteht darin, die Partitionierung auf der Grundlage des Datensatzschlüssels festzulegen (der aus der Nutzlast des Datensatzes extrahiert oder explizit festgelegt werden kann). Wir werden dies in den nächsten Kapiteln genauer erläutern.

⁷ Zu den Kompromissen gehören längere Wiederherstellungszeiten nach bestimmten Ausfallszenarien, eine höhere Ressourcennutzung (Dateideskriptoren, Speicher) und eine höhere End-to-End-Latenz.

⁸ Es gibt viele Docker-Images, aus denen du für den Betrieb von Kafka wählen kannst. Die Confluent-Images sind jedoch eine gute Wahl, da Confluent auch Docker-Images für einige der anderen Technologien anbietet, die wir in diesem Buch verwenden werden, darunter ksqlDB und Confluent Schema Registry.

⁹ Ab Version 2.7 kannst du auch das Flag --property print.headers=true verwenden, um die Kopfzeilen der Nachricht zu drucken.