Buch „Kafka Streams in Action. Anwendungen und Mikrodienste für die Echtzeitverarbeitung“

Hallo, Habrozhiteli! Dieses Buch eignet sich für jeden Entwickler, der die Stream-Verarbeitung verstehen möchte. Das Verständnis von verteiltem Programmieren hilft dabei, Kafka und Kafka Streams besser zu lernen. Es wäre gut, auch das Kafka-Framework zu kennen, aber das ist nicht zwingend erforderlich: Ich werde Ihnen alles erklären, was Sie wissen müssen. Erfahrene Kafka-Entwickler sowie Neulinge werden durch dieses Buch lernen, interessante Anwendungen für die Stream-Verarbeitung mit der Kafka Streams-Bibliothek zu erstellen. Java-Entwickler mittleren und hohen Niveaus, die bereits mit Begriffen wie Serialisierung vertraut sind, werden lernen, ihre Fähigkeiten zur Erstellung von Kafka Streams-Anwendungen anzuwenden. Der Quellcode des Buches ist in Java 8 geschrieben und nutzt erheblich die Syntax von Lambda-Ausdrücken in Java 8, sodass Sie mit Lambda-Funktionen (auch in einer anderen Programmiersprache) gut zurechtkommen sollten.

Ausschnitt. 5.3. Aggregation und Fensteroperationen

In diesem Abschnitt werden wir die vielversprechendsten Teile von Kafka Streams behandeln. Bisher haben wir die folgenden Aspekte von Kafka Streams betrachtet:

• Erstellung von Verarbeitungstopologien;
• Verwendung von Zuständen in Stream-Anwendungen;
• Ausführung von Datenstromverbindungen;
• Unterschiede zwischen Ereignisströmen (KStream) und Aktualisierungsströmen (KTable).

In den folgenden Beispielen werden wir all diese Elemente zusammenführen. Außerdem lernen Sie Fensteroperationen kennen – eine weitere großartige Möglichkeit für Stream-Anwendungen. Unser erstes Beispiel wird eine einfache Aggregation sein.

5.3.1. Aggregation des Verkaufsvolumens von Aktien nach Branchen

Aggregation und Gruppierung sind lebenswichtige Werkzeuge für die Arbeit mit Streaming-Daten. Das Studium einzelner Aufzeichnungen im Zeitpunkt des Eingangs ist oft nicht ausreichend. Um zusätzliche Informationen aus den Daten zu extrahieren, sind Gruppierung und Kombination erforderlich.

In diesem Beispiel werden Sie in die Rolle eines Intraday-Traders schlüpfen, der die Verkaufsvolumina von Aktien in mehreren Branchen verfolgen muss. Besonders interessieren Sie sich für fünf Unternehmen mit den höchsten Verkaufsvolumina in jeder Branche.

Für diese Art der Aggregation sind einige Schritte erforderlich, um die Daten in die gewünschte Form zu bringen (um es allgemein zu formulieren).

1. Erstellen Sie eine Quelle basierend auf dem Thema, die unverarbeitete Informationen zum Aktienhandel veröffentlicht. Wir müssen das Objekt vom Typ StockTransaction in ein Objekt vom Typ ShareVolume abbilden. Das liegt daran, dass das Objekt StockTransaction Metadaten zu Verkäufen enthält, wir jedoch nur die Daten über die Menge der verkauften Aktien benötigen.
2. Gruppieren Sie die Daten von ShareVolume nach Aktiensymbolen. Nach der Gruppierung nach Symbolen können diese Daten auf Zwischenbeträge des Verkaufsvolumens reduziert werden. Es ist erwähnenswert, dass die Methode KStream.groupBy eine Instanz vom Typ KGroupedStream zurückgibt. Eine Instanz von KTable kann erhalten werden, indem man anschließend die Methode KGroupedStream.reduce aufruft.

Was ist das KGroupedStream-Interface?

Die Methoden KStream.groupBy und KStream.groupByKey geben eine Instanz von KGroupedStream zurück. KGroupedStream ist eine Zwischenrepräsentation des Ereignisstroms nach der Gruppierung nach Schlüsseln. Es ist nicht für die direkte Arbeit mit ihm gedacht. Stattdessen wird KGroupedStream für Aggregationsoperationen verwendet, deren Ergebnis immer eine KTable ist. Da das Ergebnis von Aggregationsoperationen eine KTable ist und ein Statusspeicher verwendet wird, werden möglicherweise nicht alle Aktualisierungen im Ergebnis weiter entlang der Pipeline gesendet.

Die Methode KTable.groupBy gibt ein ähnliches KGroupedTable zurück – eine Zwischenrepräsentation des Stroms von Aktualisierungen, die nach Schlüssel umgruppiert wurden.

Machen wir eine kurze Pause und schauen wir uns Abbildung 5.9 an, die zeigt, was wir erreicht haben. Diese Topologie sollte Ihnen bereits gut bekannt sein.

Sehen wir uns nun den Code für diese Topologie an (zu finden in der Datei src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Listing 5.2).

Der angegebene Code zeichnet sich durch Kürze und eine Vielzahl an Aktionen aus, die in wenigen Zeilen durchgeführt werden. Im ersten Parameter der Methode builder.stream können Sie etwas Neues für sich bemerken: Der Wert des Enumerierungstyps AutoOffsetReset.EARLIEST (es gibt auch LATEST), der über die Methode Consumed.withOffsetResetPolicy festgelegt wird. Mit diesem Enumerierungstyp kann die Rücksetzstrategie für jeden KStream oder KTable angegeben werden, sie hat Vorrang vor dem Rücksetzparameter aus der Konfiguration.

GroupByKey und GroupBy

Im KStream-Interface gibt es zwei Methoden zur Gruppierung von Datensätzen: GroupByKey und GroupBy. Beide geben ein KGroupedTable zurück, sodass sich die berechtigte Frage stellt: Was ist der Unterschied zwischen ihnen und wann sollte man welche verwenden?

Die Methode GroupByKey wird angewendet, wenn die Schlüssel in KStream bereits nicht leer sind. Und das Wichtigste ist, dass das Flag „erfordert erneute Partitionierung“ niemals gesetzt wurde.

Die Methode GroupBy geht davon aus, dass Sie die Schlüssel zur Gruppierung geändert haben, sodass das Flag zur erneuten Partitionierung auf true gesetzt ist. Das Ausführen von Verknüpfungen, Aggregationen usw. nach der Methode GroupBy führt zu einer automatischen erneuten Partitionierung.
Zusammenfassung: Es sollte bei geringster Möglichkeit die Methode GroupByKey anstelle von GroupBy verwendet werden.

Was die Methoden mapValues und groupBy tun, ist klar, also schauen wir uns die Methode sum() an (sie finden Sie in der Datei src/main/java/bbejeck/model/ShareVolume.java) (Listing 5.3).

Die Methode ShareVolume.sum gibt die zwischenzeitliche Summe des Aktienverkaufsvolumens zurück, und das Ergebnis der gesamten Berechnungskette stellt ein Objekt KTable dar. Jetzt verstehen Sie, welche Rolle KTable spielt. Bei Eintreffen von Objekten ShareVolume wird im entsprechenden Objekt KTable das letzte aktuelle Update gespeichert. Es ist wichtig, sich daran zu erinnern, dass alle Updates im vorhergehenden shareVolumeKTable abgebildet werden, aber nicht alle werden weitergeleitet.

Anschließend führen wir mit diesem KTable eine Aggregation (nach der Anzahl der verkauften Aktien) durch, um die fünf Unternehmen mit dem höchsten Verkaufsvolumen in jeder Branche zu ermitteln. Unsere Schritte dabei sind den Schritten beim ersten Aggregieren ähnlich.

1. Führen Sie einen weiteren groupBy-Vorgang durch, um einzelne Objekte ShareVolume nach Branchen zu gruppieren.
2. Beginnen Sie mit der Summierung der Objekte ShareVolume. Diesmal stellt das Aggregatobjekt eine Prioritätswarteschlange mit fester Größe dar. In einer solchen Warteschlange fester Größe werden nur fünf Unternehmen mit den höchsten Verkaufszahlen an Aktien gespeichert.
3. Stellen Sie die Warteschlangen aus dem vorherigen Punkt in eine Zeichenfolge um und geben Sie die fünf am meisten verkauften Aktien in den Branchen zurück.
4. Schreiben Sie die Ergebnisse als Zeichenfolge in das Thema.

In Abb. 5.10 ist der Graph der Datenstromtopologie dargestellt. Wie Sie sehen, ist der zweite Verarbeitungszyklus recht einfach.

Nachdem Sie nun die Struktur dieses zweiten Verarbeitungszyklus klar verstanden haben, können Sie auf den Quellcode zugreifen (Sie finden ihn in der Datei src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Listing 5.4).

In diesem Initialisierer gibt es eine Variable fixedQueue. Es handelt sich um ein benutzerdefiniertes Objekt – einen Adapter für java.util.TreeSet, der verwendet wird, um die N größten Verkaufszahlen in absteigender Reihenfolge der verkauften Aktien zu verfolgen.

Sie sind bereits auf die Methoden groupBy und mapValues gestoßen, daher werden wir nicht länger darauf eingehen (wir rufen die Methode KTable.toStream auf, da die Methode KTable.print als veraltet gilt). Aber Sie haben bisher die KTable-Version der Methode aggregate() noch nicht gesehen, also werden wir ein wenig Zeit auf deren Diskussion verwenden.

Wie Sie sich erinnern, unterscheidet sich KTable dadurch, dass Datensätze mit denselben Schlüsseln als Aktualisierungen betrachtet werden. KTable ersetzt den alten Datensatz durch den neuen. Das Aggregieren erfolgt auf ähnliche Weise: Es werden die neuesten Datensätze mit demselben Schlüssel aggregiert. Wenn ein Datensatz eintrifft, wird er mit Hilfe des Summierers (dem zweiten Parameter im Methodenaufruf aggregate) zum FixedSizePriorityQueue-Objekt hinzugefügt; wenn bereits ein anderer Datensatz mit demselben Schlüssel existiert, wird der alte Datensatz durch den Subtrahierer (den dritten Parameter im Methodenaufruf aggregate) entfernt.

Das bedeutet alles, dass unser Aggregator, FixedSizePriorityQueue, nicht alle Werte mit demselben Schlüssel aggregiert, sondern eine gleitende Summe der N meistverkauften Aktienarten speichert. In jedem eingehenden Datensatz befindet sich die gesamte Anzahl der bisher verkauften Aktien. KTable gibt Ihnen Informationen darüber, welche Aktien von Unternehmen derzeit am meisten verkauft werden, eine gleitende Aggregation jeder Aktualisierung ist nicht erforderlich.

Wir haben gelernt, zwei wichtige Dinge zu tun:

• Werte in KTable nach ihrem gemeinsamen Schlüssel zu gruppieren;
• nützliche Operationen wie Faltung und Aggregierung an diesen gruppierten Werten durchzuführen.

Die Fähigkeit, diese Operationen durchzuführen, ist wichtig, um den Sinn der Daten zu verstehen, die durch die Kafka-Streams-Anwendung fließen, und um herauszufinden, welche Informationen sie enthalten.

Wir haben auch einige der Schlüsselkonzepte, die zuvor in diesem Buch besprochen wurden, zusammengeführt. In Kapitel 4 haben wir erläutert, wie wichtig für Streaming-Anwendungen ein ausfallsicherer, lokaler Zustand ist. Das erste Beispiel aus diesem Kapitel demonstrierte, warum lokaler Zustand so wichtig ist – er ermöglicht es, nachzuvollziehen, welche Informationen Sie bereits gesehen haben. Der lokale Zugriff verhindert Netzwerkverzögerungen, wodurch die Anwendung leistungsfähiger und fehlerresistenter wird.

Bei der Durchführung jeder Aggregations- oder Reduktionsoperation muss der Name des Statusspeichers angegeben werden. Aggregations- und Reduktionsoperationen geben eine Instanz von KTable zurück, und KTable verwendet den Statusspeicher, um alte Ergebnisse durch neue zu ersetzen. Wie Sie gesehen haben, werden nicht alle Aktualisierungen weiter durch die Pipeline geleitet, was wichtig ist, denn Aggregationsoperationen sind dazu gedacht, Endergebnisse zu erhalten. Ohne den lokalen Zustand wird KTable alle Ergebnisse der Aggregation und Reduktion weiterleiten.

Als Nächstes werden wir uns mit der Durchführung solcher Operationen wie Aggregation über einen bestimmten Zeitraum, den sogenannten Fensteroperationen (windowing operations), befassen.

5.3.2. Fensteroperationen

Im vorherigen Abschnitt haben wir uns mit der "gleitenden" Reduktion und Aggregation vertraut gemacht. Die Anwendung führte eine kontinuierliche Reduktion des Verkaufsvolumens von Aktien durch, gefolgt von der Aggregation der fünf meistverkauften Aktien an der Börse.

Manchmal sind solche kontinuierlichen Aggregationen und Reduktionen der Ergebnisse notwendig. Manchmal müssen jedoch Operationen nur über einen bestimmten Zeitraum durchgeführt werden. Zum Beispiel, um zu berechnen, wie viele Aktiengeschäfte eines bestimmten Unternehmens in den letzten 10 Minuten durchgeführt wurden. Oder wie viele Benutzer in den letzten 15 Minuten auf ein neues Werbebanner geklickt haben. Die Anwendung kann solche Operationen mehrfach durchführen, jedoch mit Ergebnissen, die sich nur auf die vorgegebenen Zeiträume (Zeitfenster) beziehen.

Zählung der Börsentransaktionen nach Käufer

Im nächsten Beispiel werden wir uns mit der Nachverfolgung von Börsentransaktionen über mehrere Trader beschäftigen – entweder große Organisationen oder pfiffige Einzelkämpfer im Finanzbereich.

Es gibt zwei mögliche Gründe für eine solche Überwachung. Einer davon ist die Notwendigkeit zu wissen, was die Marktführer kaufen/verkaufen. Wenn diese großen Akteure und erfahrenen Investoren für sich selbst erkennbare Chancen sehen, macht es Sinn, deren Strategien zu folgen. Der zweite Grund ist das Bestreben, mögliche Anzeichen für illegale Transaktionen unter Verwendung von Insiderinformationen zu identifizieren. Dazu müssen Sie die Korrelation zwischen großen Verkaufsanstiegen und wichtigen Pressemitteilungen analysieren.

Eine solche Überwachung besteht aus folgenden Phasen:

• Erstellen eines Streams zum Lesen aus dem Thema stock-transactions;
• Gruppierung der eingehenden Datensätze nach Käufer-ID und Börsensymbol. Der Methodenaufruf groupBy gibt eine Instanz der Klasse KGroupedStream zurück;
• Rückgabe eines Datenstroms, der durch einen Zeitrahmen begrenzt ist, mit der Methode KGroupedStream.windowedBy, was das zeitbasierte Aggregieren ermöglicht. Abhängig von der Fensterart wird entweder TimeWindowedKStream oder SessionWindowedKStream zurückgegeben;
• Zählen der Transaktionen für die Aggregationsoperation. Der fenstergebundene Datenstrom bestimmt, ob ein bestimmter Datensatz bei dieser Zählung berücksichtigt wird;
• Aufzeichnung der Ergebnisse in das Thema oder Ausgabe in die Konsole während der Entwicklung.

Die Topologie dieser Anwendung ist einfach, aber eine anschauliche Darstellung ist hilfreich. Schauen wir uns Abb. 5.11 an.

Im Folgenden werden wir die Funktionalität der Fensteroperationen und den entsprechenden Code betrachten.

Fenstertypen

In Kafka Streams gibt es drei Arten von Fenstern:

• Sitzungsfenster;
• „Tumbling“ Fenster;
• Gleitende/„Hüpfende“ Fenster (sliding/hopping).

Welches gewählt wird, hängt von den geschäftlichen Anforderungen ab. „Tumbling“ und „Hüpfende“ Fenster sind zeitlich begrenzt, während die Einschränkungen von Sitzungsfenstern von Benutzeraktionen abhängen — die Dauer der Sitzung(en) wird ausschließlich durch die Aktivität des Benutzers bestimmt. Wichtig ist zu beachten, dass alle Fenstertypen auf den Zeitstempeln der Datensätze basieren und nicht auf der Systemzeit.

Im Folgenden werden wir unsere Topologie für jeden Fenstertyp implementieren. Der vollständige Code wird nur im ersten Beispiel angegeben; für andere Fenstertypen ändert sich nichts außer der Art der Fensteroperation.

Sitzungsfenster

Sitzungsfenster unterscheiden sich stark von allen anderen Fensterarten. Sie sind nicht so sehr zeitlich eingeschränkt, sondern hängen von der Aktivität des Benutzers (oder der Entität, die Sie verfolgen möchten) ab. Sitzungsfenster werden durch Phasen der Inaktivität abgegrenzt.

Abbildung 5.12 veranschaulicht das Konzept der Sitzungsfenster. Eine kürzere Sitzung wird mit der Sitzung links von ihr verschmelzen. Die Sitzung rechts wird separat sein, da sie auf eine längere Phase der Inaktivität folgt. Sitzungsfenster basieren auf Benutzeraktivitäten, verwenden jedoch Zeitstempel aus den Aufzeichnungen, um zu bestimmen, zu welcher Sitzung der Datensatz gehört.

Die Verwendung von Sitzungsfenstern zur Verfolgung von Börsentransaktionen

Wir nutzen Sitzungsfenster, um Informationen über Börsentransaktionen zu erfassen. Die Implementierung von Sitzungsfenstern wird im Listing 5.5 gezeigt (das Sie in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java finden können).

Die meisten Operationen dieser Topologie haben Sie bereits kennengelernt, sodass es nicht nötig ist, sie hier erneut zu betrachten. Es gibt jedoch auch einige neue Elemente, die wir jetzt besprechen werden.

Bei jeder groupBy-Operation wird normalerweise eine Aggregationsoperation (Aggregation, Reduzierung oder Zählung) durchgeführt. Es kann entweder eine kumulative Aggregation mit laufendem Ergebnis oder eine Fensteraggregation durchgeführt werden, bei der Datensätze innerhalb eines bestimmten Zeitfensters berücksichtigt werden.

Der Code aus Listing 5.5 zählt die Anzahl der Transaktionen innerhalb der Sitzungsfenster. In Abb. 5.13 werden diese Aktionen Schritt für Schritt analysiert.

Mit dem Aufruf windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) erstellen wir ein Sitzungsfenster mit einer Inaktivitätsperiode von 20 Sekunden und einer Behaltedauer von 15 Minuten. Die Inaktivitätsperiode von 20 Sekunden bedeutet, dass die Anwendung jeden Datensatz einbezieht, der innerhalb von 20 Sekunden nach dem Ende oder Beginn der aktuellen Sitzung im aktuellen (aktiven) Sitzung eingeht.

Als Nächstes geben wir an, welche Aggregationsoperation im Sitzungsfenster ausgeführt werden soll – in diesem Fall count. Wenn der eingehende Datensatz außerhalb des Inaktivitätsintervalls (von beiden Seiten des Zeitstempels) liegt, erstellt die Anwendung eine neue Sitzung. Das Aufrechterhaltungsintervall bedeutet, dass die Sitzung für einen bestimmten Zeitraum aktiv bleibt und verspätete Daten erlaubt, die über den Zeitraum der Sitzung hinausgehen, aber immer noch angehängt werden können. Außerdem entsprechen Beginn und Ende der neuen Sitzung, die aus der Zusammenführung resultiert, dem frühesten und spätesten Zeitstempel.

Betrachten wir einige Datensätze aus der Methode count, um zu sehen, wie Sitzungen funktionieren (Tab. 5.1).

Beim Eintreffen von Datensätzen suchen wir nach bereits bestehenden Sitzungen mit dem gleichen Schlüssel, deren Endzeit weniger als der aktuelle Zeitstempel ist – Inaktivitätsintervall und deren Startzeit größer als der aktuelle Zeitstempel + Inaktivitätsintervall. Unter Berücksichtigung dessen werden vier Datensätze aus Tab. 5.1 zu einer Sitzung zusammengeführt.

1. Zuerst trifft Datensatz 1 ein, sodass die Startzeit gleich der Endzeit ist und 00:00:00 entspricht.

2. Als Nächstes trifft Datensatz 2 ein, und wir suchen nach Sitzungen, die nicht früher als 23:59:55 enden und nicht später als 00:00:35 beginnen. Wir finden Datensatz 1 und fusionieren die Sitzungen 1 und 2. Wir nehmen die Startzeit von Sitzung 1 (früher) und die Endzeit von Sitzung 2 (später), sodass unsere neue Sitzung um 00:00:00 beginnt und um 00:00:15 endet.

3. Datensatz 3 trifft ein, wir suchen nach Sitzungen zwischen 00:00:30 und 00:01:10 und finden keine. Wir fügen eine zweite Sitzung für den Schlüssel 123-345-654,FFBE hinzu, die um 00:00:50 beginnt und endet.

4. Datensatz 4 trifft ein, und wir suchen nach Sitzungen zwischen 23:59:45 und 00:00:25. Dieses Mal finden wir beide Sitzungen – 1 und 2. Alle drei Sitzungen werden zu einer zusammengeführt, mit einer Startzeit von 00:00:00 und einer Endzeit von 00:00:15.

Aus den Ausführungen in diesem Abschnitt sind folgende wichtige Aspekte zu merken:

• Sitzungen sind keine Fenster fester Größe. Die Dauer einer Sitzung wird durch die Aktivität innerhalb eines bestimmten Zeitrahmens bestimmt;
• Zeitstempel in den Daten bestimmen, ob ein Ereignis in eine bestehende Sitzung oder in das Inaktivitätsintervall fällt.

Als Nächstes besprechen wir die nächste Art von Fenstern – „rollende“ Fenster.

„Rollende“ Fenster

„Tumbling“-Fenster erfassen Ereignisse, die in einem bestimmten Zeitfenster auftreten. Stellen Sie sich vor, Sie müssten alle Börsentransaktionen eines Unternehmens alle 20 Sekunden erfassen, sodass Sie alle Ereignisse für diesen Zeitraum sammeln. Am Ende des 20-Sekunden-Intervalls „tumbelt“ das Fenster und wechselt zu einem neuen 20-Sekunden-Beobachtungsintervall. Abbildung 5.14 illustriert diese Situation.

Wie Sie sehen können, sind alle in den letzten 20 Sekunden eingegangenen Ereignisse im Fenster enthalten. Nach Ablauf dieses Zeitraums wird ein neues Fenster erstellt.

Im Listing 5.6 finden Sie den Code, der die Verwendung von „tumbling“ Fenstern zur Erfassung von Börsentransaktionen alle 20 Sekunden demonstriert (er ist in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java zu finden).

Durch diese kleine Änderung des Aufrufs der Methode TimeWindows.of kann ein „tumbling“ Fenster verwendet werden. In diesem Beispiel gibt es keinen Aufruf der Methode until(), weshalb das Standard-Speicherintervall von 24 Stunden verwendet wird.

Schließlich ist es an der Zeit, zur letzten Fensteroption überzugehen – „hopping“ Fenstern.

Gleitende („hopping“) Fenster

Gleitende/„hopping“ (sliding/hopping) Fenster ähneln den „tumbling“ Fenstern, unterscheiden sich jedoch in einem kleinen Punkt. Gleitende Fenster warten nicht auf das Ende des Zeitintervalls, bevor sie ein neues Fenster zur Verarbeitung der aktuellen Ereignisse erstellen. Sie starten neue Berechnungen nach einem kürzeren Warteintervall als der Fensterlaufzeit.

Um die Unterschiede zwischen „tumbling“ und „hopping“ Fenstern zu veranschaulichen, kehren wir zum Beispiel mit der Zählung der Börsentransaktionen zurück. Unser Ziel bleibt die Zählung der Transaktionen, aber wir möchten nicht den gesamten Zeitraum abwarten, bevor wir den Zähler aktualisieren. Stattdessen werden wir den Zähler in kürzeren Zeitabschnitten aktualisieren. Zum Beispiel zählen wir weiterhin die Transaktionen alle 20 Sekunden, aber wir aktualisieren den Zähler alle 5 Sekunden, wie in Abb. 5.15 gezeigt. Dabei haben wir drei Ergebnisfenster mit überlappenden Daten.

Im Listing 5.7 finden Sie den Code zur Festlegung von gleitenden Fenstern (er ist in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java zu finden).

Ein "wackelndes" Fenster kann durch einen Aufruf der Methode advanceBy() in ein "springendes" Fenster umgewandelt werden. In dem folgenden Beispiel beträgt das Intervall der Speicherung 15 Minuten.

In diesem Abschnitt haben Sie gesehen, wie man die Ergebnisse der Aggregation mithilfe von Zeitfenstern einschränken kann. Insbesondere sollten Sie sich aus diesem Abschnitt die folgenden drei Dinge merken:

• Die Größe der Sitzungsfenster wird nicht durch einen Zeitintervall, sondern durch die Aktivität der Benutzer begrenzt;
• „Wackelnde“ Fenster geben Einblick in Ereignisse innerhalb eines bestimmten Zeitraums;
• Die Dauer der „springenden“ Fenster ist festgelegt, aber sie werden häufig aktualisiert und können in allen Fenstern sich überschneidende Einträge enthalten.

Als Nächstes erfahren wir, wie man KTable zurück in KStream umwandelt, um Verbindungen herzustellen.

5.3.3. Verbindung von KStream- und KTable-Objekten

In Kapitel 4 haben wir die Verbindung von zwei KStream-Objekten besprochen. Jetzt lernen wir, KTable und KStream zu verbinden. Dies kann aus einem einfachen Grund notwendig sein. KStream ist ein Strom von Einträgen, während KTable ein Strom von Aktualisierungen der Einträge ist, aber manchmal kann es notwendig sein, dem Eintragsstrom zusätzlichen Kontext durch Aktualisierungen aus KTable hinzuzufügen.

Nehmen wir Daten über die Anzahl von Börsentransaktionen und verbinden sie mit Börsennachrichten aus den entsprechenden Branchen. Hier ist, was zu tun ist, um dies unter Berücksichtigung des bereits vorhandenen Codes zu erreichen.

1. Das KTable-Objekt mit den Daten zur Anzahl der Börsentransaktionen in KStream umwandeln und den Schlüssel durch einen Schlüssel ersetzen, der die entsprechende Branche für das jeweilige Wertpapier bezeichnet.
2. Ein KTable-Objekt erstellen, das Daten aus dem Thema mit den Börsennachrichten liest. Dieses neue KTable wird nach Branchen kategorisiert.
3. Aktualisierungen der Nachrichten mit Informationen über die Anzahl der Börsentransaktionen nach Branchen verbinden.

Jetzt sehen wir uns an, wie wir diesen Aktionsplan umsetzen können.

Umwandlung von KTable in KStream

Für die Umwandlung von KTable in KStream sind die folgenden Schritte erforderlich.

1. Die Methode KTable.toStream() aufrufen.
2. Durch den Aufruf der Methode KStream.map den Schlüssel durch den Namen der Branche ersetzen und dann aus der Instanz Windowed das Objekt TransactionSummary extrahieren.

Wir werden diese Operationen in einer Kette wie folgt verknüpfen (der Code ist in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java zu finden) (Listing 5.8).

Da wir die Operation KStream.map ausführen, erfolgt die erneute Partitionierung der zurückgegebenen KStream-Instanz automatisch bei deren Verwendung in einer Verbindung.

Wir haben den Transformationsprozess abgeschlossen, jetzt müssen wir ein KTable-Objekt erstellen, um Börsennachrichten zu lesen.

Erstellung eines KTable für Börsennachrichten

Glücklicherweise reicht eine einzige Codezeile aus, um ein KTable-Objekt zu erstellen (diesen Code finden Sie in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.9).

Es ist anzumerken, dass keine Serde-Objekte angegeben werden müssen, da in den Einstellungen stringbasierte Serde verwendet werden. Zudem wird dank der Verwendung der Enumeration EARLIEST die Tabelle zu Beginn mit Datensätzen gefüllt.

Jetzt können wir zum letzten Schritt übergehen — der Verbindung.

Verbindung von Nachrichtenupdates mit Transaktionsdaten

Die Erstellung einer Verbindung ist unkompliziert. Wir werden einen linken Join verwenden, falls in der entsprechenden Branche keine Börsennachrichten vorliegen (den benötigten Code finden Sie in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.10).

Dieser Operator leftJoin ist recht einfach. Im Gegensatz zu den Verbindungen aus Kapitel 4 wird die Methode JoinWindow nicht verwendet, da bei einer KStream-KTable-Verbindung für jeden Schlüssel in der KTable nur ein Datensatz vorhanden ist. Diese Verbindung ist nicht zeitlich begrenzt: Der Datensatz ist entweder in der KTable oder nicht. Die Hauptaussage ist: Mit KTable-Objekten kann KStream mit weniger häufig aktualisierten Referenzdaten angereichert werden.

Jetzt betrachten wir einen effizienteren Weg, um Ereignisse aus KStream anzureichern.

5.3.4. Objekte GlobalKTable

Wie Sie verstanden haben, besteht die Notwendigkeit, Ereignisströme anzureichern oder ihnen Kontext hinzuzufügen. In Kapitel 4 haben Sie die Verbindung zweier KStream-Objekte gesehen, und im vorherigen Abschnitt — die Verbindung zwischen KStream und KTable. In all diesen Fällen ist eine erneute Partitionierung des Datenstroms erforderlich, wenn die Schlüssel auf einen neuen Typ oder Wert abgebildet werden. Manchmal erfolgt die erneute Partitionierung explizit, manchmal erledigt Kafka Streams dies automatisch. Eine erneute Partitionierung ist notwendig, da sich die Schlüssel geändert haben und die Datensätze in neuen Partitionen landen müssen, andernfalls wäre eine Verbindung unmöglich (dies wurde in Kapitel 4 im Abschnitt „Erneute Partitionierung von Daten“ unter Abschnit 4.2.4 behandelt).

Wiederholte Partitionierung hat ihren Preis

Wiederholte Partitionierung erfordert Kosten – zusätzliche Ressourcen zur Erstellung von Zwischen-Topics, die Speicherung redundanter Daten in einem weiteren Topic; sie bedeutet auch eine Erhöhung der Latenz durch das Schreiben und Lesen aus diesem Topic. Darüber hinaus müssen bei der Notwendigkeit, mehr als einen Aspekt oder eine Dimension zu verbinden, Verbindungen in einer Kette organisiert, die Einträge mit neuen Schlüsseln abgebildet und der Prozess der Wiederholten Partitionierung erneut durchgeführt werden.

Verbindung zu Datenmengen kleinerer Größe

In einigen Fällen ist das Volumen der Referenzdaten, mit denen eine Verbindung geplant ist, relativ klein, sodass vollständige Kopien davon lokal auf jedem der Knoten Platz finden können. Für solche Situationen bietet Kafka Streams die Klasse GlobalKTable.

Instanzen von GlobalKTable sind einzigartig, da die Anwendung alle Daten auf jeden der Knoten repliziert. Und da auf jedem der Knoten alle Daten vorhanden sind, ist es nicht notwendig, den Ereignisstrom nach dem Schlüssel der Referenzdaten zu partitionieren, damit er allen Partitionen zur Verfügung steht. Mit GlobalKTable-Objekten können auch schlüssellose Verbindungen durchgeführt werden. Lassen Sie uns zu einem der vorherigen Beispiele zurückkehren, um diese Möglichkeit zu demonstrieren.

Verbindung von KStream-Objekten mit GlobalKTable-Objekten

Im Abschnitt 5.3.2 haben wir eine Fensteraggregation von Börsentransaktionen nach Kunden durchgeführt. Die Ergebnisse dieser Aggregation sahen ungefähr folgendermaßen aus:

{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16

Obwohl diese Ergebnisse dem angestrebten Ziel entsprachen, wäre es praktischer, wenn auch der Name des Kunden und der vollständige Name des Unternehmens ausgegeben würden. Um den Namen des Käufers und den Namen des Unternehmens hinzuzufügen, können herkömmliche Verbindungen hergestellt werden, dabei sind jedoch zwei Schlüsselabbildungen und eine erneute Partitionierung erforderlich. Mit GlobalKTable können solche Kosten vermieden werden.

Dazu verwenden wir das Objekt countStream aus Listing 5.11 (den entsprechenden Code finden Sie in der Datei src/main/java/bbejeck/chapter_5/GlobalKTableExample.java), indem wir es mit zwei GlobalKTable-Objekten verbinden.

Wir haben das bereits früher besprochen, deshalb will ich mich nicht wiederholen. Ich möchte jedoch darauf hinweisen, dass der Code in der Funktion toStream().map zur besseren Lesbarkeit in ein Funktionsobjekt abstrahiert wurde, anstelle einer eingebetteten Lambda-Ausdrucks.

Der nächste Schritt ist die Deklaration von zwei Instanzen von GlobalKTable (der bereitgestellte Code ist in der Datei src/main/java/bbejeck/chapter_5/GlobalKTableExample.java zu finden) (Listing 5.12).

Bitte beachten Sie, dass die Namen der Themen mithilfe von Enumerationen beschrieben werden.

Jetzt, da wir alle Komponenten vorbereitet haben, bleibt nur noch der Code für die Verknüpfung zu schreiben (der in der Datei src/main/java/bbejeck/chapter_5/GlobalKTableExample.java zu finden ist) (Listing 5.13).

Obwohl der oben genannte Code zwei Verknüpfungen enthält, sind sie in einer Kette organisiert, da keiner ihrer Ergebnisse separat verwendet wird. Die Ergebnisse werden am Ende der gesamten Operation ausgegeben.

Wenn Sie die oben genannte Verknüpfungsoperation ausführen, erhalten Sie Ergebnisse in der folgenden Form:

{customer='Barney, Smith' company="Exxon", transactions= 17}

Der Kern hat sich nicht verändert, aber diese Ergebnisse sind verständlicher.

Wenn man Kapitel 4 zählt, haben Sie bereits mehrere Arten von Verknüpfungen in Aktion gesehen. Diese sind in Tabelle 5.2 aufgeführt. Diese Tabelle zeigt die Verknüpfungsmöglichkeiten, die für die Version 1.0.0 von Kafka Streams relevant sind; in zukünftigen Versionen könnte sich etwas ändern.

Zusammenfassend möchte ich das Wesentliche in Erinnerung rufen: Sie können Ereignisströme (KStream) und Aktualisierungsströme (KTable) mit lokalem Status verknüpfen. Außerdem kann, sofern die Größe der Referenzdaten nicht zu groß ist, das Objekt GlobalKTable verwendet werden. GlobalKTable repliziert alle Partitionen auf jeden Knoten der Kafka Streams-Anwendung und sorgt so für die Verfügbarkeit aller Daten, unabhängig davon, zu welcher Partition der Schlüssel gehört.

Als nächstes werden wir die Möglichkeit von Kafka Streams sehen, die es erlaubt, Statusänderungen zu beobachten, ohne Daten aus dem Kafka-Topic zu konsumieren.

5.3.5. Abfragefähiger Status

Wir haben bereits mehrere Operationen mit Status durchgeführt und die Ergebnisse immer in der Konsole ausgegeben (zu Entwicklungszwecken) oder in einem Topic gespeichert (für industrielle Anwendungen). Bei der Speicherung von Ergebnissen in einem Topic müssen Sie einen Kafka-Verbraucher verwenden, um sie anzuzeigen.

Das Lesen von Daten aus diesen Themen kann als eine Form von materialisierten Sichten (materialized views) angesehen werden. Für unsere Zwecke kann die Definition einer materialisierten Sicht aus der „Wikipedia“ verwendet werden: „…ein physisches Objekt einer Datenbank, das die Ergebnisse der Ausführung einer Abfrage enthält. Es kann beispielsweise eine lokale Kopie entfernten Daten sein oder eine Teilmenge von Zeilen und/oder Spalten einer Tabelle oder Ergebnissen von Joins oder eine Pivot-Tabelle, die durch Aggregation erhalten wurde“ (https://de.wikipedia.org/wiki/Materialisierte_Sicht).

Kafka Streams ermöglicht auch die Ausführung interaktiver Abfragen (interactive queries) an den Zustands-Speichern, was die direkte Lesbarkeit dieser materialisierten Sichten ermöglicht. Es ist wichtig zu beachten, dass eine Abfrage an den Zustands-Speicher eine „Nur-Lese“-Operation ist. Dank dessen müssen Sie sich keine Sorgen machen, die Konsistenz des Zustands während der Datenverarbeitung durch die Anwendung versehentlich zu stören.

Die Möglichkeit von direkten Abfragen an den Zustands-Speichern ist von großer Bedeutung. Sie bedeutet, dass Sie Anwendungen – Dashboards – erstellen können, ohne zuvor Daten von einem Kafka-Konsumenten abzurufen. Sie steigert auch die Effizienz der Anwendung, da es nicht erforderlich ist, die Daten erneut zu speichern:

• durch die Lokalisierung der Daten kann schnell darauf zugegriffen werden;
• Daten-Duplikate werden ausgeschlossen, da sie nicht in einem externen Speicher gespeichert werden.

Das Wichtigste, was ich möchte, dass Sie sich merken: Sie können direkt Abfragen an den Zustand aus der Anwendung ausführen. Die Möglichkeiten, die sich daraus ergeben, können nicht hoch genug eingeschätzt werden. Anstatt Daten aus Kafka zu konsumieren und Datensätze in einer Datenbank für die Anwendung zu speichern, können Abfragen an die Zustands-Speicher durchgeführt werden, mit demselben Ergebnis. Direkte Abfragen an die Zustands-Speicher bedeuten weniger Code (keinen Verbraucher) und weniger Software (kein Bedarf an einer Datenbanktabelle zur Speicherung der Ergebnisse).

In diesem Kapitel haben wir eine beträchtliche Menge an Informationen behandelt, daher pausieren wir vorerst unser Gespräch über interaktive Abfragen an Zustandsdatenbanken. Aber keine Sorge: In Kapitel 9 werden wir eine einfache Anwendung erstellen – ein Dashboard mit interaktiven Abfragen. Zur Demonstration interaktiver Abfragen und der Möglichkeiten, sie in Kafka Streams-Anwendungen einzufügen, werden einige Beispiele aus diesem sowie aus dem vorherigen Kapitel verwendet.

Zusammenfassung

• KStream-Objekte repräsentieren Ereignisströme, ähnlich wie Einfügungen in einer Datenbank. KTable-Objekte hingegen repräsentieren Aktualisierungsströme und ähneln mehr den Aktualisierungen in einer Datenbank. Die Größe eines KTable-Objekts wächst nicht, alte Einträge werden durch neue ersetzt.
• KTable-Objekte sind erforderlich für Aggregationsoperationen.
• Mit Fensteroperationen können aggregierte Daten in zeitliche Zeitfenster unterteilt werden.
• Durch GlobalKTable-Objekte kann auf Referenzdaten an beliebigen Stellen der Anwendung zugegriffen werden, unabhängig von der Partitionierung.
• Es können Verbindungen zwischen KStream-, KTable- und GlobalKTable-Objekten hergestellt werden.

Bis jetzt haben wir uns auf die Erstellung von Kafka Streams-Anwendungen mit der hochgradig abstrahierten DSL KStream konzentriert. Obwohl der hochgradige Ansatz die Erstellung sauberer und prägnanter Programme ermöglicht, stellt seine Anwendung einen bestimmten Kompromiss dar. Die Arbeit mit der DSL KStream bedeutet mehr Prägnanz des Codes auf Kosten des Kontrolle. Im nächsten Kapitel werden wir die Low-Level-API von Processing Nodes betrachten und probieren andere Kompromisse aus. Die Programme werden länger sein als bisher, aber wir werden die Möglichkeit haben, praktisch jeden benötigten Processing Node zu erstellen.

→ Mehr Details zur Buch finden Sie unter der Verlagswebsite

→ Für Habr-Leser gibt es 25% Rabatt mit dem Coupon — Kafka Streams

→ Nach Zahlung für die gedruckte Buchausgabe wird das E-Book per E-Mail versendet.

Quelle: habr.com