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

Hallo, Habr-Nutzer! Dieses Buch eignet sich für jeden Entwickler, der sich mit Stream-Processing vertrautmachen möchte. Ein Verständnis des verteilten Programmierens wird Ihnen helfen, Kafka und Kafka Streams besser zu verstehen. Es wäre von Vorteil, auch mit dem Kafka-Framework vertraut zu sein, ist aber nicht zwingend erforderlich: Ich werde Ihnen alles Notwendige erklären. Sowohl erfahrene Kafka-Entwickler als auch Anfänger werden mit diesem Buch lernen, interessante Anwendungen für die Stream-Verarbeitung mit der Kafka Streams-Bibliothek zu erstellen. Java-Entwickler auf mittlerem und höherem Niveau, die mit Konzepten wie Serialisierung bereits vertraut sind, werden lernen, wie sie ihre Fähigkeiten zur Erstellung von Kafka Streams-Anwendungen einsetzen können. Der Quellcode des Buches ist in Java 8 geschrieben und nutzt stark die Syntax von Lambda-Ausdrücken in Java 8, sodass Kenntnisse im Umgang mit Lambda-Funktionen (auch in einer anderen Programmiersprache) nützlich sein werden.

Auszug. 5.3. Aggregation und Fensteroperationen

In diesem Abschnitt werden wir uns mit den vielversprechendsten Teilen von Kafka Streams beschäftigen. Bisher haben wir folgende Aspekte von Kafka Streams betrachtet:

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

In den folgenden Beispielen werden wir all diese Elemente zusammenführen. Zudem werden Sie mit Fensteroperationen vertraut gemacht — einer weiteren großartigen Funktion von Streaming-Anwendungen. Unser erstes Beispiel wird eine einfache Aggregation sein.

5.3.1. Aggregation des Umsatzvolumens von Aktien nach Branchen

Aggregation und Gruppierung sind essentielle Werkzeuge im Umgang mit Streaming-Daten. Die Analyse einzelner Datensätze in Echtzeit reicht oft nicht aus. Um zusätzliche Informationen aus den Daten zu gewinnen, sind Gruppierung und Kombination notwendig.

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

Für eine solche Aggregation sind einige Schritte zur Umformung der Daten in das benötigte Format erforderlich (um es allgemein zu halten).

1. Erstellen Sie eine Quelle auf Basis eines Themas, die unverarbeitete Informationen über den Aktienhandel veröffentlicht. Wir müssen das Objekt vom Typ StockTransaction in ein Objekt vom Typ ShareVolume umwandeln. Das liegt daran, dass das StockTransaction-Objekt Metadaten zu Verkäufen enthält, wir jedoch nur die Daten zur Anzahl der verkauften Aktien benötigen.
2. Die ShareVolume-Daten nach Aktiensymbolen gruppieren. Nach der Gruppierung nach Symbolen können diese Daten auf die Zwischensummen der Verkaufsvolumen von Aktien zusammengefasst werden. Es ist erwähnenswert, dass die Methode KStream.groupBy eine Instanz vom Typ KGroupedStream zurückgibt. Um eine Instanz von KTable zu erhalten, kann die Methode KGroupedStream.reduce aufgerufen werden.

Was ist die Schnittstelle KGroupedStream?

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 dafür gedacht, direkt damit zu arbeiten. Stattdessen wird KGroupedStream für Aggregationsoperationen verwendet, deren Ergebnis immer eine KTable ist. Da das Ergebnis dieser Aggregationsoperationen eine KTable ist und ein Zustands-Backend verwendet wird, werden möglicherweise nicht alle Aktualisierungen weiter durch die Pipeline gesendet.

Die Methode KTable.groupBy gibt eine ähnliche KGroupedTable zurück – eine Zwischenrepräsentation eines Streams von Aktualisierungen, die nach Schlüssel neu gruppiert sind.

Lassen Sie uns eine kurze Pause machen und einen Blick auf Abb. 5.9 werfen, die zeigt, was wir erreicht haben. Diese Topologie sollte Ihnen bereits gut bekannt sein.

Schauen 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 bereitgestellte Code ist kompakt und führt in wenigen Zeilen viele Aktionen aus. Im ersten Parameter der Methode builder.stream finden Sie etwas Neues für sich: den Wert des Enumerations Typs AutoOffsetReset.EARLIEST (es gibt auch LATEST), der durch die Methode Consumed.withOffsetResetPolicy festgelegt wird. Mit diesem Enumerations Typ können Sie die Offset-Rücksetzstrategie für jeden KStream oder KTable angeben, die Priorität über den Offset-Rücksetzparameter aus der Konfiguration hat.

GroupByKey und GroupBy

Im KStream-Interface gibt es zwei Methoden zum Gruppieren von Datensätzen: GroupByKey und GroupBy. Beide geben eine KGroupedTable zurück, sodass Sie sich möglicherweise fragen: Was ist der Unterschied zwischen ihnen und wann sollten Sie welche verwenden?

Die Methode GroupByKey wird angewendet, wenn die Schlüssel im KStream bereits nicht leer sind. Außerdem wurde das Flag „erfordert Wiederpartitionierung“ niemals gesetzt.

Die Methode GroupBy geht davon aus, dass Sie die Schlüssel zum Gruppieren geändert haben, sodass das Flag für die Wiederpartitionierung auf true gesetzt ist. Ausführungen von Verbindungen, Aggregationen usw. nach der Methode GroupBy führen automatisch zu einer Wiederpartitionierung.
Zusammenfassung: Es ist ratsam, wann immer möglich, GroupByKey anstelle von GroupBy zu verwenden.

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

Die Methode ShareVolume.sum gibt die Zwischensumme des Aktienhandelsvolumens zurück, und das Ergebnis der gesamten Berechnungskette ist ein Objekt KTable<String, ShareVolume>. Jetzt verstehen Sie, welche Rolle die KTable spielt. Wenn Objekte ShareVolume eintreffen, wird im entsprechenden Objekt KTable das letzte aktuelle Update gespeichert. Es ist wichtig, sich daran zu erinnern, dass alle Updates im vorhergehenden shareVolumeKTable reflektiert werden, aber nicht alle werden weitergeleitet.

Mit dieser KTable führen wir eine Aggregation durch (basierend auf der Anzahl der verkauften Aktien), um die fünf Unternehmen mit den höchsten Verkaufszahlen in jeder Branche zu ermitteln. Unser Vorgehen wird dabei ähnlich sein wie beim ersten Aggregieren.

1. Führen Sie eine weitere groupBy-Operation durch, um die einzelnen ShareVolume-Objekte nach Branchen zu gruppieren.
2. Beginnen Sie mit der Summierung der ShareVolume-Objekte. Diesmal stellt das Aggregationsobjekt eine Queue mit fester Priorität dar. In dieser Queue mit fester Größe werden nur die fünf Unternehmen mit den höchsten Verkaufszahlen gespeichert.
3. Darstellen der Queues aus dem vorherigen Punkt als Zeichenfolge und Rückgabe der fünf am meisten verkauften Aktien in den verschiedenen Branchen.
4. Die Ergebnisse in Zeichenfolgenform in das Thema schreiben.

In Abb. 5.10 ist das Diagramm der Datendatenfluss-Topologie dargestellt. Wie Sie sehen, ist der zweite Verarbeitungszyklus recht einfach.

Nachdem Sie nun die Struktur dieser zweiten Verarbeitungsstufe klar verstanden haben, können wir uns dem Quellcode zuwenden (den finden Sie in der Datei src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Listing 5.4).

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

Sie sind bereits mit den Aufrufen groupBy und mapValues vertraut, daher lassen wir diese beiseite (wir rufen die Methode KTable.toStream auf, da die Methode KTable.print als veraltet gilt). Sie haben jedoch noch die KTable-Version der Methode aggregate() nicht gesehen, daher werden wir etwas Zeit darauf verwenden, diese zu besprechen.

Wie Sie sich erinnern, zeichnet sich KTable dadurch aus, dass Einträge mit denselben Schlüsseln als Aktualisierungen betrachtet werden. KTable ersetzt den alten Eintrag durch den neuen. Die Aggregation erfolgt auf ähnliche Weise: Es werden die letzten Einträge mit einem Schlüssel aggregiert. Bei Eintreffen eines Eintrags wird dieser mithilfe des Summierers (zweiter Parameter im Methodenaufruf aggregate) in eine Instanz der Klasse FixedSizePriorityQueue eingefügt. Besteht bereits ein anderer Eintrag mit demselben Schlüssel, wird der alte Eintrag mithilfe des Subtrahierers (dritter Parameter im Methodenaufruf aggregate) entfernt.

Das bedeutet, dass unser Aggregator, FixedSizePriorityQueue, nicht alle Werte mit einem Schlüssel aggregiert, sondern eine gleitende Summe der Mengen der N meistverkauften Aktien speichert. In jedem eingehenden Eintrag befindet sich die Gesamtzahl der bisher verkauften Aktien. KTable gibt Ihnen Aufschluss darüber, welche Unternehmen momentan die meisten Aktien verkaufen; eine gleitende Aggregation jedes Updates ist nicht erforderlich.

Wir haben gelernt, zwei wichtige Dinge zu tun:

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

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

Wir haben auch einige der Schlüsselkonzepte zusammengeführt, die zuvor in diesem Buch behandelt wurden. In Kapitel 4 haben wir erläutert, wie wichtig es für eine Streaming-Anwendung ist, eine ausfallsichere lokale Speicherung zu haben. Das erste Beispiel in diesem Kapitel zeigte, warum lokale Speicherung so entscheidend ist – sie ermöglicht es, nachzuvollziehen, welche Informationen Sie bereits gesehen haben. Der lokale Zugriff ermöglicht es, Netzwerkverzögerungen zu vermeiden, wodurch die Anwendung leistungsfähiger und fehlertoleranter wird.

Bei der Durchführung von Faltungs- oder Aggregationsoperationen muss der Name des Zustandsspeichers angegeben werden. Faltungs- und Aggregationsoperationen geben eine Instanz von KTable zurück, wobei KTable den Zustandsspeicher verwendet, um alte Ergebnisse durch neue zu ersetzen. Wie Sie gesehen haben, werden nicht alle Aktualisierungen weiter im Kanal gesendet, was wichtig ist, da Aggregationsoperationen darauf abzielen, endgültige Informationen zu liefern. Wenn kein lokaler Zustand angewendet wird, wird KTable alle Ergebnisse der Aggregation und Faltung weiterleiten.

Als Nächstes werden wir uns mit der Ausführung von Operationen wie Aggregation in einem bestimmten Zeitraum befassen – den sogenannten Fensteroperationen.

5.3.2. Fensteroperationen

Im vorherigen Abschnitt haben wir die „gleitenden“ Faltungen und Aggregationen kennengelernt. Die Anwendung führte eine kontinuierliche Faltung des Handelsvolumens von Aktien mit anschließender Aggregation der fünf meistverkauften Aktien an der Börse durch.

Manchmal sind kontinuierliche Aggregationen und Zusammenfassungen der Ergebnisse notwendig. In anderen Fällen müssen jedoch Operationen nur für einen bestimmten Zeitraum ausgeführt werden. Zum Beispiel, um zu berechnen, wie viele Börsentransaktionen mit den Aktien eines bestimmten Unternehmens in den letzten 10 Minuten durchgeführt wurden. Oder wie viele Nutzer auf das neue Werbebanner in den letzten 15 Minuten geklickt haben. Die Anwendung kann solche Aktionen wiederholt ausführen, jedoch nur mit Ergebnissen, die sich auf die festgelegten Zeiträume (Zeitfenster) beziehen.

Zählung von Börsentransaktionen nach Käufer

Im nächsten Beispiel werden wir Börsentransaktionen für mehrere Trader - entweder große Unternehmen oder fähige Einzelkämpfer - verfolgen.

Es gibt zwei mögliche Gründe für eine solche Überwachung. Der erste ist die Notwendigkeit zu wissen, was Marktführer kaufen oder verkaufen. Wenn diese großen Akteure und erfahrenen Investoren attraktive Möglichkeiten sehen, macht es Sinn, ihrer Strategie zu folgen. Der zweite Grund ist das Bedürfnis, mögliche Anzeichen für illegale Geschäfte mit Insiderinformationen zu erkennen. Dazu müssen Sie die Korrelation zwischen großen Verkaufsanstiegen und wichtigen Pressemitteilungen analysieren.

Dieser Überwachungsprozess besteht aus mehreren Schritten, wie zum Beispiel:

• Erstellung eines Leseflusses 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 des Datenstroms durch die Methode KGroupedStream.windowedBy, die auf ein zeitlich begrenztes Fenster beschränkt ist, wodurch die Möglichkeit zur Durchführung von Fensteraggregationen besteht. Je nach Art des Fensters wird entweder eine TimeWindowedKStream oder eine SessionWindowedKStream zurückgegeben;
• Zählen der Transaktionen für die Aggregationsoperation. Der Fensterdatenstrom bestimmt, ob ein bestimmter Datensatz in dieser Zählung berücksichtigt wird;
• Das Protokollieren von Ergebnissen in das Thema oder das Ausgeben auf der Konsole während der Entwicklung.

Die Topologie dieser Anwendung ist einfach, aber eine visuelle Darstellung kann nicht schaden. Schauen wir uns Abbildung 5.11 an.

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

Fenstertypen

In Kafka Streams gibt es drei Fensterarten:

• Sitzungsfenster;
• „Tumbling“-Fenster;
• Gleitende/„Hopping“-Fenster.

Welche man wählt, hängt von den geschäftlichen Anforderungen ab. „Tumbling“- und „Hopping“-Fenster sind zeitlich begrenzt, während die Einschränkungen der Sitzungsfenster mit Benutzeraktionen verknüpft sind – die Dauer der Sitzung(en) wird ausschließlich durch die Aktivität des Benutzers bestimmt. Wichtig ist, dass alle Fenstertypen auf Zeitstempeln der Aufzeichnungen basieren und nicht auf der Systemzeit.

Anschließend werden wir unsere Topologie mit jedem der Fenstertypen implementieren. Der vollständige Code wird nur im ersten Beispiel angegeben; für andere Fenstertypen ändert sich nichts, außer dem Typ der Fensteroperation.

Sitzungsfenster

Sitzungsfenster unterscheiden sich grundlegend von allen anderen Fensterarten. Sie sind nicht so sehr zeitlich begrenzt, sondern hängen von der Aktivität des Nutzers (oder der Entität, die Sie verfolgen möchten) ab. Sitzungsfenster werden durch Perioden der Inaktivität getrennt.

Abbildung 5.12 veranschaulicht das Konzept der Sitzungsfenster. Eine kleinere Sitzung wird mit der Sitzung links von ihr verschmelzen. Die Sitzung rechts wird hingegen separat sein, da sie auf eine längere Inaktivitätsperiode folgt. Sitzungsfenster basieren auf den Aktivitäten der Nutzer, verwenden jedoch Zeitstempel aus den Aufzeichnungen, um zu bestimmen, welcher Sitzung der Datensatz zugeordnet wird.

Verwendung von Sitzungsfenstern zur Verfolgung von Börsentransaktionen

Nutzen wir Sitzungsfenster, um Informationen über Börsentransaktionen zu erfassen. Die Implementierung von Sitzungsfenstern ist im Listing 5.5 dargestellt (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 wir sie hier nicht erneut betrachten müssen. Es gibt jedoch auch einige neue Elemente, die wir jetzt besprechen werden.

Bei jeder groupBy-Operation wird in der Regel eine Art von Aggregation durchgeführt (Aggregation, Zusammenfassung oder Zählung). Es kann entweder eine kumulative Aggregation mit fortlaufenden Ergebnissen oder eine Fensteraggregation durchgeführt werden, bei der die 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 Abbildung 5.13 werden diese Schritte der Analyse schrittweise dargestellt.

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

Im Folgenden geben wir an, welche Aggregationsoperation im Sitzungfenster durchgeführt werden soll – in diesem Fall ein `count`. Wenn der eingehende Datensatz den Inaktivitätszeitraum (von einem der Zeitstempel) überschreitet, erstellt die Anwendung eine neue Sitzung. Die Speicherdauer bedeutet, dass die Sitzung für einen bestimmten Zeitraum aufrechterhalten wird und verspätete Daten zulässt, die außerhalb des Inaktivitätszeitraums liegen, aber dennoch angefügt werden können. Darüber hinaus entsprechen der Beginn und das Ende der neuen Sitzung, die durch die Zusammenführung entstanden ist, dem frühesten und spätesten Zeitstempel.

Lassen Sie uns einige Datensätze aus der `count`-Methode betrachten, um zu sehen, wie Sitzungen funktionieren (Tabelle 5.1).

Bei eintreffenden Datensätzen suchen wir nach bereits bestehenden Sitzungen mit dem gleichen Schlüssel, wobei die Endzeit kleiner als der aktuelle Zeitstempel ist — Inaktivitätszeitraum und die Startzeit größer als der aktuelle Zeitstempel plus Inaktivitätszeitraum. In Anbetracht dessen werden vier Datensätze aus Tabelle 5.1 folgendermaßen zu einer Sitzung zusammengeführt.

1. Zuerst kommt Datensatz 1 an, sodass die Startzeit gleich der Endzeit ist und beide 00:00:00 betragen.

2. Dann wird der Eintrag 2 hinzugefügt, und wir suchen nach Sitzungen, die nicht vor 23:59:55 enden und nicht später als 00:00:35 beginnen. Wir finden den Eintrag 1 und kombinieren die Sitzungen 1 und 2. Wir nehmen die Startzeit der Sitzung 1 (die frühere) und die Endzeit der Sitzung 2 (die spätere), sodass unsere neue Sitzung um 00:00:00 beginnt und um 00:00:15 endet.

3. Eintrag 3 wird hinzugefügt, wir suchen nach Sitzungen zwischen 00:00:30 und 00:01:10 und finden keine. Der zweite Sitzung für den Schlüssel 123-345-654,FFBE wird hinzugefügt, die um 00:00:50 beginnt und endet.

4. Eintrag 4 wird hinzugefügt, und wir suchen nach Sitzungen zwischen 23:59:45 und 00:00:25. Diesmal 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 dem, was in diesem Abschnitt erzählt wurde, sind folgende wichtige Aspekte zu merken:

• Sitzungen sind keine Fenster fester Größe. Die Dauer einer Sitzung wird durch die Aktivität innerhalb eines festgelegten Zeitraums bestimmt;
• Zeitstempel in den Daten bestimmen, ob ein Ereignis in eine bestehende Sitzung oder in einen Zeitraum ohne Aktivität fällt.

Als Nächstes werden wir die nächste Art von Fenstern besprechen — die „wackelnden“ Fenster.

„Wackelnde“ Fenster

„Kippende“ Fenster erfassen Ereignisse, die in einen bestimmten Zeitrahmen fallen. Stellen Sie sich vor, Sie müssen alle Börsentransaktionen eines Unternehmens alle 20 Sekunden erfassen, damit Sie alle Ereignisse in diesem Zeitraum sammeln. Nach Ablauf des 20-Sekunden-Intervalls „kippt“ das Fenster und wechselt zu einem neuen 20-Sekunden-Beobachtungsintervall. Abbildung 5.14 veranschaulicht 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.

In Listing 5.6 finden Sie den Code, der die Verwendung von „kippenden“ 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 Methodenaufrufs TimeWindows.of kann ein „kippendes“ Fenster verwendet werden. In diesem Beispiel gibt es keinen Aufruf der Methode until(), was bedeutet, dass das Standard-Speicherintervall von 24 Stunden verwendet wird.

Schließlich ist es an der Zeit, zum letzten Fenstertyp überzugehen – den „springenden“ Fenstern.

Gleitende („springende“) Fenster

Gleitende/„springende“ Fenster ähneln den „rollenden“, jedoch mit einem kleinen Unterschied. Gleitende Fenster warten nicht auf das Ende des Zeitintervalls, bevor sie ein neues Fenster zur Verarbeitung der letzten Ereignisse erstellen. Sie starten neue Berechnungen nach einem kürzeren Warteintervall als der Dauer des Fensters.

Um die Unterschiede zwischen „rollenden“ und „springenden“ Fenstern zu veranschaulichen, kehren wir zu unserem Beispiel mit der Zählung von Börsentransaktionen zurück. Unser Ziel bleibt es, die Anzahl der Transaktionen zu zählen, jedoch möchten wir nicht den gesamten Zeitabschnitt abwarten, bevor wir den Zähler aktualisieren. Stattdessen werden wir den Zähler in kürzeren Zeitabständen aktualisieren. Zum Beispiel werden wir die Anzahl der Transaktionen weiterhin alle 20 Sekunden zählen, aber den Zähler alle 5 Sekunden aktualisieren, wie in Abb. 5.15 gezeigt. Dadurch haben wir drei Ergebnisfenster mit überlappenden Daten.

Im Listing 5.7 finden Sie den Code zur Definition der gleitenden Fenster (dieser befindet sich in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Ein "rollendes" Fenster kann durch den Aufruf der Methode advanceBy() in ein "springendes" Fenster umgewandelt werden. Im gegebenen Beispiel beträgt das Speicherintervall 15 Minuten.

In diesem Abschnitt haben Sie gelernt, wie man die Ergebnisse der Aggregation mit Zeitfenstern einschränkt. Insbesondere sollten Sie sich folgende drei Punkte aus diesem Abschnitt merken:

• Die Größe der Sitzungsfenster wird nicht durch den Zeitrahmen, sondern durch die Aktivität der Nutzer begrenzt.
• "Rollende" Fenster geben Einblick in Ereignisse innerhalb eines bestimmten Zeitraums.
• Die Dauer von "springenden" Fenstern ist festgelegt, aber sie werden häufig aktualisiert und können in allen Fenstern überlappende Einträge enthalten.

Als Nächstes werden wir lernen, wie man ein KTable zurück in einen KStream umwandelt, um eine Verbindung herzustellen.

5.3.3. Verbindung von KStream- und KTable-Objekten

In Kapitel 4 haben wir die Verbindung zweier KStream-Objekte besprochen. Jetzt müssen wir lernen, KTable und KStream zu verbinden. Dies kann aus folgendem einfachen Grund notwendig sein: KStream ist ein Strom von Einträgen, während KTable ein Strom von Aktualisierungen der Einträge ist. Manchmal ist es jedoch notwendig, dem Strom von Einträgen zusätzlichen Kontext durch Aktualisierungen aus KTable hinzuzufügen.

Wir nehmen die Daten über die Anzahl der Börsentransaktionen und verknüpfen sie mit den entsprechenden Branchennews. So erreichen wir dies unter Berücksichtigung des bereits vorhandenen Codes.

1. Das KTable-Objekt mit den Daten über die Anzahl der Börsentransaktionen in einen KStream umwandeln, wobei der Schlüssel durch den Schlüssel ersetzt wird, der die entsprechende Branche für dieses Aktienzeichen bezeichnet.
2. Ein KTable-Objekt erstellen, das Daten aus dem Topic mit den Börsennachrichten liest. Dieses neue KTable wird nach Branchen kategorisiert.
3. Die Neuigkeiten aktualisieren und mit den Informationen über die Anzahl der Börsentransaktionen nach Branchen verknüpfen.

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

Umwandlung von KTable in KStream

Um KTable in KStream umzuwandeln, sind folgende 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 anschließend aus der Instanz des Windowed-Objekts TransactionSummary extrahieren.

Wir verknüpfen diese Operationen in einer Kette folgendermaßen (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, wird die Neupartitionierung für die zurückgegebene Instanz von KStream automatisch vorgenommen, wenn sie in einer Verbindung verwendet wird.

Wir haben den Umwandlungsprozess abgeschlossen; nun müssen wir ein KTable-Objekt für das Lesen von Börsennachrichten erstellen.

Erstellung eines KTable für Börsennachrichten

Glücklicherweise genügt zur Erstellung eines KTable-Objekts eine einzige Codezeile (diesen Code finden Sie in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.9).

Es ist erwähnenswert, dass keine Serde-Objekte angegeben werden müssen, da die Konfiguration String-Serde verwendet. Zudem wird durch die Verwendung der Enumeration EARLIEST die Tabelle von Anfang an mit Einträgen gefüllt.

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

Verbindung von Nachrichtenaktualisierungen mit Transaktionszahlen

Die Erstellung der Verbindung ist unkompliziert. Wir verwenden einen linken Join für den Fall, dass es in der betreffenden Branche keine Börsennachrichten gibt (den benötigten Code finden Sie in der Datei src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.10).

Dieser leftJoin-Operator ist recht einfach. Im Gegensatz zu den Verknüpfungen aus Kapitel 4 wird die Methode JoinWindow nicht verwendet, da bei der Ausführung der KStream-KTable-Verknüpfung nur ein Eintrag pro Schlüssel in der KTable vorhanden ist. Diese Verknüpfung ist nicht zeitlich begrenzt: Eintrag ist entweder in der KTable vorhanden oder nicht. Die wichtigste Erkenntnis: Mit KTable-Objekten kann der KStream mit seltener aktualisierten Referenzdaten angereichert werden.

Jetzt betrachten wir eine effizientere Methode zur Anreicherung von Ereignissen aus dem KStream.

5.3.4. GlobalKTable-Objekte

Wie Sie verstanden haben, besteht die Notwendigkeit, die Ereignisströme anzureichern oder ihnen Kontext hinzuzufügen. In Kapitel 4 haben Sie die Verbindung zwischen zwei KStreams gesehen, und im vorherigen Abschnitt wurde die Verbindung zwischen KStream und KTable behandelt. In all diesen Fällen ist eine Neupartitionierung des Datenstroms erforderlich, wenn die Schlüssel auf einen neuen Typ oder Wert abgebildet werden. Manchmal erfolgt die Neupartitionierung ausdrücklich, manchmal übernimmt Kafka Streams dies automatisch. Die Neupartitionierung 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 nicht möglich (dies wurde in Kapitel 4 im Abschnitt „Neupartitionierung von Daten“ unter Punkt 4.2.4 behandelt).

Neupartitionierung hat ihren Preis

Die erneute Partitionierung erfordert zusätzliche Kosten – es fallen zusätzliche Ressourcen für die Erstellung von Zwischenthemen an und es werden doppelte Daten in einem weiteren Thema gespeichert; dies führt auch zu einer erhöhten Verzögerung aufgrund des Schreibens und Lesens aus diesem Thema. Zudem müssen bei Bedarf Verbindungen in einer Kette organisiert werden, wenn mehr als ein Aspekt oder Maßstab beteiligt ist, um die Datensätze mit neuen Schlüsseln anzuzeigen und den Prozess der erneuten Partitionierung erneut durchzuführen.

Verbindungen zu kleineren Datensätzen

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

GlobalKTable-Instanzen sind einzigartig, da die Anwendung alle Daten auf jedem Knoten repliziert. Da auf jedem Knoten alle Daten vorhanden sind, ist es nicht notwendig, den Datenstrom nach dem Schlüssel der Referenzdaten zu partitionieren, damit er für alle Abschnitte verfügbar ist. Mit GlobalKTable-Objekten können Sie auch schlüsselunabhängige Joins durchführen. Lassen Sie uns zu einem der vorherigen Beispiele zurückkehren, um diese Möglichkeit zu demonstrieren.

Einen KStream mit GlobalKTable-Objekten verbinden

Im Abschnitt 5.3.2 haben wir das fensterbasierte Aggregieren von Börsentransaktionen nach Käufern durchgeführt. Die Ergebnisse dieser Aggregation sahen ungefähr so aus:

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

Obwohl diese Ergebnisse den festgelegten Zielen 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 Firmennamen hinzuzufügen, können Standard-Joins durchgeführt werden, aber es müssen zwei Schlüsselzuordnungen und eine erneute Partitionierung vorgenommen werden. Mit GlobalKTable können solche Kosten vermieden werden.

Dafür 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) und verbinden es mit zwei Objekten GlobalKTable.

Wir haben dies bereits zuvor besprochen, daher werde ich nicht wiederholen. Ich möchte jedoch darauf hinweisen, dass der Code in der Funktion toStream().map zur besseren Lesbarkeit in eine Funktionsobjekt abstrahiert wurde, anstelle eines eingebetteten Lambda-Ausdrucks.

Der nächste Schritt besteht darin, zwei Instanzen von GlobalKTable zu deklarieren (der entsprechende Code befindet sich in der Datei src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (Listing 5.12).

Bitte beachten Sie, dass die Namen der Themen durch enumerierte Typen beschrieben werden.

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

Obwohl dieser Code zwei Verbindungen aufweist, sind diese als Kette organisiert, da keiner ihrer Ergebnisse einzeln genutzt wird. Die Ergebnisse werden am Ende der gesamten Operation ausgegeben.

Wenn Sie die oben genannte Verbindungsoperation ausführen, erhalten Sie Ergebnisse in folgender Form:

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

Die Essenz bleibt unverändert, aber diese Ergebnisse sind klarer dargestellt.

Wenn man Kapitel 4 einbezieht, haben Sie bereits mehrere Verbindungstypen in Aktion gesehen. Diese sind in Tabelle 5.2 aufgeführt. Diese Tabelle spiegelt die Verbindungsmöglichkeiten wider, 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) mithilfe des lokalen Zustands verbinden. Darüber hinaus kann, wenn die Größe der Referenzdaten nicht zu groß ist, das Objekt GlobalKTable verwendet werden. GlobalKTable repliziert alle Partitionen auf jedem Knoten der Kafka Streams-Anwendung, wodurch alle Daten verfügbar sind, unabhängig davon, zu welcher Partition der Schlüssel gehört.

Als nächstes werden wir die Funktionalität von Kafka Streams sehen, mit der Änderungen des Zustands beobachtet werden können, ohne Daten aus dem Kafka-Topic zu konsumieren.

5.3.5. Abfragbarer Zustand

Wir haben bereits mehrere Operationen mit dem Zustand durchgeführt und die Ergebnisse immer in die Konsole ausgegeben (zu Entwicklungszwecken) oder in ein Topic geschrieben (für industrielle Anwendungen). Um die Ergebnisse im Topic zu speichern, verwenden wir den Kafka-Client zur Ansicht.

Das Lesen von Daten aus diesen Topics kann als eine Form von materialisierten Sichten angesehen werden. Für unsere Zwecke verwenden wir die Definition einer materialisierten Sicht aus der „Wikipedia“: „… ein physisches Objekt der Datenbank, das die Ergebnisse einer Abfrage enthält. Zum Beispiel kann es eine lokale Kopie entfernten Daten, oder eine Teilmenge von Zeilen und/oder Spalten einer Tabelle oder von Ergebnissen einer Join-Abfrage oder eine Pivot-Tabelle, die durch Aggregation erhalten wurde, sein“ (https://en.wikipedia.org/wiki/Materialized_view).

Kafka Streams ermöglicht auch interaktive Abfragen an den Zustandspeichern, was das direkte Lesen dieser materialisierten Ansichten erlaubt. Es ist wichtig zu beachten, dass die Abfrage an den Zustandspeicher eine "nur-Lese"-Operation ist. Dadurch können Sie sicher sein, dass der Zustand während der Datenverarbeitung durch die Anwendung nicht versehentlich inkonsistent wird.

Die Möglichkeit, direkte Abfragen an Zustandspeicher durchzuführen, ist von großer Bedeutung. Sie bedeutet, dass Anwendungen wie Dashboards erstellt werden können, ohne zunächst Daten vom Kafka-Consumer abrufen zu müssen. Dadurch wird die Effizienz der Anwendung gesteigert, da eine erneute Datenspeicherung entfällt:

• aufgrund der Lokalität der Daten können sie schnell abgerufen werden;
• Datenverdopplungen werden ausgeschlossen, da sie nicht in ein externes Lager geschrieben werden.

Das Wichtigste, was Sie sich merken sollten: Anfragen an den Zustand können direkt aus der Anwendung ausgeführt werden. Die Möglichkeiten, die sich daraus ergeben, sind nicht zu unterschätzen. Anstatt Daten aus Kafka zu konsumieren und Datensätze in einer Datenbank für die Anwendung zu speichern, können Sie Anfragen an die Statusspeicher mit demselben Ergebnis stellen. Direkte Anfragen an Statusspeicher bedeuten weniger Code (keine Konsumenten) und weniger Software (kein Bedarf an einer Datenbanktabelle zur Speicherung der Ergebnisse).

Wir haben in diesem Kapitel eine erhebliche Menge an Informationen behandelt, daher werden wir unsere Diskussion über interaktive Anfragen an Statusspeicher vorübergehend pausieren. Aber keine Sorge: In Kapitel 9 werden wir eine einfache Anwendung erstellen – ein Dashboard mit interaktiven Anfragen. Zur Demonstration der interaktiven Anfragen und deren Integration in Anwendungen wird es einige Beispiele aus diesem und dem vorherigen Kapitel verwenden.

Zusammenfassung

• KStream-Objekte repräsentieren Ereignisströme, die mit Datenbankeinfügungen vergleichbar sind. KTable-Objekte repräsentieren Aktualisierungsströme und ähneln eher Updates in einer Datenbank. Die Größe eines KTable-Objekts wächst nicht, alte Einträge werden durch neue ersetzt.
• KTable-Objekte sind für Aggregierungsoperationen unerlässlich.
• Mit Fensteroperationen lassen sich aggregierte Daten in zeitliche Intervalle unterteilen.
• Dank GlobalKTable-Objekten kann auf Referenzdaten an jedem Punkt der Anwendung zugegriffen werden, unabhängig von der Partitionierung.
• Es sind Verbindungen zwischen KStream-, KTable- und GlobalKTable-Objekten möglich.

Bisher haben wir uns auf die Erstellung von Kafka Streams-Anwendungen mit der hochgradigen DSL KStream konzentriert. Dieser Ansatz ermöglicht zwar die Entwicklung klarer und prägnanter Programme, bringt jedoch gewisse Kompromisse mit sich. Die Arbeit mit der DSL KStream verbesserte zwar die Kürze des Codes, reduzierte jedoch den Grad der Kontrolle. Im nächsten Kapitel werden wir die Low-Level-API für Bearbeitungs-Node betrachten und andere Kompromisse erkunden. Die Programme werden länger sein als bisher, aber wir erhalten die Möglichkeit, nahezu jede benötigte Bearbeitungs-Node zu erstellen.

→ Weitere Informationen zur Buch finden Sie auf der Verlagswebsite

→ Für Habr-Nutzer gibt es einen Rabatt von 25 % mit dem Gutschein — Kafka Streams

→ Nach der Zahlung der gedruckten Buchversion wird das E-Book per E-Mail zugesendet.

Quelle: habr.com