{"id":35786,"date":"2019-10-31T22:06:19","date_gmt":"2019-10-31T19:06:19","guid":{"rendered":"https:\/\/prohoster.info\/blog\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\/"},"modified":"2019-10-31T22:06:19","modified_gmt":"2019-10-31T19:06:19","slug":"kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","status":"publish","type":"post","link":"https:\/\/prohoster.info\/de\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","title":{"rendered":"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"<p><noindex><a rel=\"nofollow\" href=\"https:\/\/habr.com\/ru\/company\/piter\/blog\/457756\/\"><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/6f8bd2b31b87b0c760c1148515893c43.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/a><\/noindex> Hallo, Habrozhiteli! Dieses Buch eignet sich f\u00fcr jeden Entwickler, der die Stream-Verarbeitung verstehen m\u00f6chte. Das Verst\u00e4ndnis von verteiltem Programmieren hilft dabei, Kafka und Kafka Streams besser zu lernen. Es w\u00e4re gut, auch das Kafka-Framework zu kennen, aber das ist nicht zwingend erforderlich: Ich werde Ihnen alles erkl\u00e4ren, was Sie wissen m\u00fcssen. Erfahrene Kafka-Entwickler sowie Neulinge werden durch dieses Buch lernen, interessante Anwendungen f\u00fcr 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\u00e4higkeiten zur Erstellung von Kafka Streams-Anwendungen anzuwenden. Der Quellcode des Buches ist in Java 8 geschrieben und nutzt erheblich die Syntax von Lambda-Ausdr\u00fccken in Java 8, sodass Sie mit Lambda-Funktionen (auch in einer anderen Programmiersprache) gut zurechtkommen sollten.<br \/>\n<noindex><a rel=\"nofollow\" name=\"habracut\"><\/a><\/noindex><\/p>\n<h3>Ausschnitt. 5.3. Aggregation und Fensteroperationen<\/h3>\n<p>\nIn diesem Abschnitt werden wir die vielversprechendsten Teile von Kafka Streams behandeln. Bisher haben wir die folgenden Aspekte von Kafka Streams betrachtet:<\/p>\n<ul>\n<li>Erstellung von Verarbeitungstopologien;<\/li>\n<li>Verwendung von Zust\u00e4nden in Stream-Anwendungen;<\/li>\n<li>Ausf\u00fchrung von Datenstromverbindungen;<\/li>\n<li>Unterschiede zwischen Ereignisstr\u00f6men (KStream) und Aktualisierungsstr\u00f6men (KTable).<\/li>\n<\/ul>\n<p>\nIn den folgenden Beispielen werden wir all diese Elemente zusammenf\u00fchren. Au\u00dferdem lernen Sie Fensteroperationen kennen \u2013 eine weitere gro\u00dfartige M\u00f6glichkeit f\u00fcr Stream-Anwendungen. Unser erstes Beispiel wird eine einfache Aggregation sein.<\/p>\n<h3>5.3.1. Aggregation des Verkaufsvolumens von Aktien nach Branchen<\/h3>\n<p>\nAggregation und Gruppierung sind lebenswichtige Werkzeuge f\u00fcr die Arbeit mit Streaming-Daten. Das Studium einzelner Aufzeichnungen im Zeitpunkt des Eingangs ist oft nicht ausreichend. Um zus\u00e4tzliche Informationen aus den Daten zu extrahieren, sind Gruppierung und Kombination erforderlich.<\/p>\n<p>In diesem Beispiel werden Sie in die Rolle eines Intraday-Traders schl\u00fcpfen, der die Verkaufsvolumina von Aktien in mehreren Branchen verfolgen muss. Besonders interessieren Sie sich f\u00fcr f\u00fcnf Unternehmen mit den h\u00f6chsten Verkaufsvolumina in jeder Branche.<\/p>\n<p>F\u00fcr diese Art der Aggregation sind einige Schritte erforderlich, um die Daten in die gew\u00fcnschte Form zu bringen (um es allgemein zu formulieren).<\/p>\n<ol>\n<li>Erstellen Sie eine Quelle basierend auf dem Thema, die unverarbeitete Informationen zum Aktienhandel ver\u00f6ffentlicht. Wir m\u00fcssen das Objekt vom Typ StockTransaction in ein Objekt vom Typ ShareVolume abbilden. Das liegt daran, dass das Objekt StockTransaction Metadaten zu Verk\u00e4ufen enth\u00e4lt, wir jedoch nur die Daten \u00fcber die Menge der verkauften Aktien ben\u00f6tigen.<\/li>\n<li>Gruppieren Sie die Daten von ShareVolume nach Aktiensymbolen. Nach der Gruppierung nach Symbolen k\u00f6nnen diese Daten auf Zwischenbetr\u00e4ge des Verkaufsvolumens reduziert werden. Es ist erw\u00e4hnenswert, dass die Methode KStream.groupBy eine Instanz vom Typ KGroupedStream zur\u00fcckgibt. Eine Instanz von KTable kann erhalten werden, indem man anschlie\u00dfend die Methode KGroupedStream.reduce aufruft.<\/li>\n<\/ol>\n<p><\/p>\n<blockquote><p><b>Was ist das KGroupedStream-Interface?<\/b><\/p>\n<p>Die Methoden KStream.groupBy und KStream.groupByKey geben eine Instanz von KGroupedStream zur\u00fcck. KGroupedStream ist eine Zwischenrepr\u00e4sentation des Ereignisstroms nach der Gruppierung nach Schl\u00fcsseln. Es ist nicht f\u00fcr die direkte Arbeit mit ihm gedacht. Stattdessen wird KGroupedStream f\u00fcr Aggregationsoperationen verwendet, deren Ergebnis immer eine KTable ist. Da das Ergebnis von Aggregationsoperationen eine KTable ist und ein Statusspeicher verwendet wird, werden m\u00f6glicherweise nicht alle Aktualisierungen im Ergebnis weiter entlang der Pipeline gesendet.<\/p>\n<p>Die Methode KTable.groupBy gibt ein \u00e4hnliches KGroupedTable zur\u00fcck \u2013 eine Zwischenrepr\u00e4sentation des Stroms von Aktualisierungen, die nach Schl\u00fcssel umgruppiert wurden.<\/p><\/blockquote>\n<p>\nMachen 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.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/9fd61317cde376362adcaeec72908919.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nSehen wir uns nun den Code f\u00fcr diese Topologie an (zu finden in der Datei src\/main\/java\/bbejeck\/chapter_5\/AggregationsAndReducingExample.java) (Listing 5.2).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/f937287e448295fbd467c283ceca316a.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDer angegebene Code zeichnet sich durch K\u00fcrze und eine Vielzahl an Aktionen aus, die in wenigen Zeilen durchgef\u00fchrt werden. Im ersten Parameter der Methode builder.stream k\u00f6nnen Sie etwas Neues f\u00fcr sich bemerken: Der Wert des Enumerierungstyps AutoOffsetReset.EARLIEST (es gibt auch LATEST), der \u00fcber die Methode Consumed.withOffsetResetPolicy festgelegt wird. Mit diesem Enumerierungstyp kann die R\u00fccksetzstrategie f\u00fcr jeden KStream oder KTable angegeben werden, sie hat Vorrang vor dem R\u00fccksetzparameter aus der Konfiguration.<\/p>\n<blockquote><p><b>GroupByKey und GroupBy<\/b><\/p>\n<p>Im KStream-Interface gibt es zwei Methoden zur Gruppierung von Datens\u00e4tzen: GroupByKey und GroupBy. Beide geben ein KGroupedTable zur\u00fcck, sodass sich die berechtigte Frage stellt: Was ist der Unterschied zwischen ihnen und wann sollte man welche verwenden?<\/p>\n<p>Die Methode GroupByKey wird angewendet, wenn die Schl\u00fcssel in KStream bereits nicht leer sind. Und das Wichtigste ist, dass das Flag \u201eerfordert erneute Partitionierung\u201c niemals gesetzt wurde.<\/p>\n<p>Die Methode GroupBy geht davon aus, dass Sie die Schl\u00fcssel zur Gruppierung ge\u00e4ndert haben, sodass das Flag zur erneuten Partitionierung auf true gesetzt ist. Das Ausf\u00fchren von Verkn\u00fcpfungen, Aggregationen usw. nach der Methode GroupBy f\u00fchrt zu einer automatischen erneuten Partitionierung.<br \/>\nZusammenfassung: Es sollte bei geringster M\u00f6glichkeit die Methode GroupByKey anstelle von GroupBy verwendet werden.<\/p><\/blockquote>\n<p>\nWas 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).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/8e9a6f873594f9b5fef9a96c42353a61.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDie Methode ShareVolume.sum gibt die zwischenzeitliche Summe des Aktienverkaufsvolumens zur\u00fcck, 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.<\/p>\n<p>Anschlie\u00dfend f\u00fchren wir mit diesem KTable eine Aggregation (nach der Anzahl der verkauften Aktien) durch, um die f\u00fcnf Unternehmen mit dem h\u00f6chsten Verkaufsvolumen in jeder Branche zu ermitteln. Unsere Schritte dabei sind den Schritten beim ersten Aggregieren \u00e4hnlich.<\/p>\n<ol>\n<li>F\u00fchren Sie einen weiteren groupBy-Vorgang durch, um einzelne Objekte ShareVolume nach Branchen zu gruppieren.<\/li>\n<li>Beginnen Sie mit der Summierung der Objekte ShareVolume. Diesmal stellt das Aggregatobjekt eine Priorit\u00e4tswarteschlange mit fester Gr\u00f6\u00dfe dar. In einer solchen Warteschlange fester Gr\u00f6\u00dfe werden nur f\u00fcnf Unternehmen mit den h\u00f6chsten Verkaufszahlen an Aktien gespeichert.<\/li>\n<li>Stellen Sie die Warteschlangen aus dem vorherigen Punkt in eine Zeichenfolge um und geben Sie die f\u00fcnf am meisten verkauften Aktien in den Branchen zur\u00fcck.<\/li>\n<li>Schreiben Sie die Ergebnisse als Zeichenfolge in das Thema.<\/li>\n<\/ol>\n<p>\nIn Abb. 5.10 ist der Graph der Datenstromtopologie dargestellt. Wie Sie sehen, ist der zweite Verarbeitungszyklus recht einfach.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/dabd1507eee267038edb7f8d76d8d8ae.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nNachdem Sie nun die Struktur dieses zweiten Verarbeitungszyklus klar verstanden haben, k\u00f6nnen Sie auf den Quellcode zugreifen (Sie finden ihn in der Datei src\/main\/java\/bbejeck\/chapter_5\/AggregationsAndReducingExample.java) (Listing 5.4).<\/p>\n<p>In diesem Initialisierer gibt es eine Variable fixedQueue. Es handelt sich um ein benutzerdefiniertes Objekt \u2013 einen Adapter f\u00fcr java.util.TreeSet, der verwendet wird, um die N gr\u00f6\u00dften Verkaufszahlen in absteigender Reihenfolge der verkauften Aktien zu verfolgen.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/488b072b0d91b81c925ca72291e69e48.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nSie sind bereits auf die Methoden groupBy und mapValues gesto\u00dfen, daher werden wir nicht l\u00e4nger 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.<\/p>\n<p>Wie Sie sich erinnern, unterscheidet sich KTable dadurch, dass Datens\u00e4tze mit denselben Schl\u00fcsseln als Aktualisierungen betrachtet werden. KTable ersetzt den alten Datensatz durch den neuen. Das Aggregieren erfolgt auf \u00e4hnliche Weise: Es werden die neuesten Datens\u00e4tze mit demselben Schl\u00fcssel aggregiert. Wenn ein Datensatz eintrifft, wird er mit Hilfe des Summierers (dem zweiten Parameter im Methodenaufruf aggregate) zum FixedSizePriorityQueue-Objekt hinzugef\u00fcgt; wenn bereits ein anderer Datensatz mit demselben Schl\u00fcssel existiert, wird der alte Datensatz durch den Subtrahierer (den dritten Parameter im Methodenaufruf aggregate) entfernt.<\/p>\n<p>Das bedeutet alles, dass unser Aggregator, FixedSizePriorityQueue, nicht alle Werte mit demselben Schl\u00fcssel 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\u00fcber, welche Aktien von Unternehmen derzeit am meisten verkauft werden, eine gleitende Aggregation jeder Aktualisierung ist nicht erforderlich.<\/p>\n<p>Wir haben gelernt, zwei wichtige Dinge zu tun:<\/p>\n<ul>\n<li>Werte in KTable nach ihrem gemeinsamen Schl\u00fcssel zu gruppieren;<\/li>\n<li>n\u00fctzliche Operationen wie Faltung und Aggregierung an diesen gruppierten Werten durchzuf\u00fchren.<\/li>\n<\/ul>\n<p>\nDie F\u00e4higkeit, diese Operationen durchzuf\u00fchren, ist wichtig, um den Sinn der Daten zu verstehen, die durch die Kafka-Streams-Anwendung flie\u00dfen, und um herauszufinden, welche Informationen sie enthalten.<\/p>\n<p>Wir haben auch einige der Schl\u00fcsselkonzepte, die zuvor in diesem Buch besprochen wurden, zusammengef\u00fchrt. In Kapitel 4 haben wir erl\u00e4utert, wie wichtig f\u00fcr Streaming-Anwendungen ein ausfallsicherer, lokaler Zustand ist. Das erste Beispiel aus diesem Kapitel demonstrierte, warum lokaler Zustand so wichtig ist \u2013 er erm\u00f6glicht es, nachzuvollziehen, welche Informationen Sie bereits gesehen haben. Der lokale Zugriff verhindert Netzwerkverz\u00f6gerungen, wodurch die Anwendung leistungsf\u00e4higer und fehlerresistenter wird.<\/p>\n<p>Bei der Durchf\u00fchrung jeder Aggregations- oder Reduktionsoperation muss der Name des Statusspeichers angegeben werden. Aggregations- und Reduktionsoperationen geben eine Instanz von KTable zur\u00fcck, 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.<\/p>\n<p>Als N\u00e4chstes werden wir uns mit der Durchf\u00fchrung solcher Operationen wie Aggregation \u00fcber einen bestimmten Zeitraum, den sogenannten Fensteroperationen (windowing operations), befassen.<\/p>\n<h3>5.3.2. Fensteroperationen<\/h3>\n<p>\nIm vorherigen Abschnitt haben wir uns mit der \"gleitenden\" Reduktion und Aggregation vertraut gemacht. Die Anwendung f\u00fchrte eine kontinuierliche Reduktion des Verkaufsvolumens von Aktien durch, gefolgt von der Aggregation der f\u00fcnf meistverkauften Aktien an der B\u00f6rse.<\/p>\n<p>Manchmal sind solche kontinuierlichen Aggregationen und Reduktionen der Ergebnisse notwendig. Manchmal m\u00fcssen jedoch Operationen nur \u00fcber einen bestimmten Zeitraum durchgef\u00fchrt werden. Zum Beispiel, um zu berechnen, wie viele Aktiengesch\u00e4fte eines bestimmten Unternehmens in den letzten 10 Minuten durchgef\u00fchrt wurden. Oder wie viele Benutzer in den letzten 15 Minuten auf ein neues Werbebanner geklickt haben. Die Anwendung kann solche Operationen mehrfach durchf\u00fchren, jedoch mit Ergebnissen, die sich nur auf die vorgegebenen Zeitr\u00e4ume (Zeitfenster) beziehen.<\/p>\n<h3>Z\u00e4hlung der B\u00f6rsentransaktionen nach K\u00e4ufer<\/h3>\n<p>\nIm n\u00e4chsten Beispiel werden wir uns mit der Nachverfolgung von B\u00f6rsentransaktionen \u00fcber mehrere Trader besch\u00e4ftigen \u2013 entweder gro\u00dfe Organisationen oder pfiffige Einzelk\u00e4mpfer im Finanzbereich.<\/p>\n<p>Es gibt zwei m\u00f6gliche Gr\u00fcnde f\u00fcr eine solche \u00dcberwachung. Einer davon ist die Notwendigkeit zu wissen, was die Marktf\u00fchrer kaufen\/verkaufen. Wenn diese gro\u00dfen Akteure und erfahrenen Investoren f\u00fcr sich selbst erkennbare Chancen sehen, macht es Sinn, deren Strategien zu folgen. Der zweite Grund ist das Bestreben, m\u00f6gliche Anzeichen f\u00fcr illegale Transaktionen unter Verwendung von Insiderinformationen zu identifizieren. Dazu m\u00fcssen Sie die Korrelation zwischen gro\u00dfen Verkaufsanstiegen und wichtigen Pressemitteilungen analysieren.<\/p>\n<p>Eine solche \u00dcberwachung besteht aus folgenden Phasen:<\/p>\n<ul>\n<li>Erstellen eines Streams zum Lesen aus dem Thema stock-transactions;<\/li>\n<li>Gruppierung der eingehenden Datens\u00e4tze nach K\u00e4ufer-ID und B\u00f6rsensymbol. Der Methodenaufruf groupBy gibt eine Instanz der Klasse KGroupedStream zur\u00fcck;<\/li>\n<li>R\u00fcckgabe eines Datenstroms, der durch einen Zeitrahmen begrenzt ist, mit der Methode KGroupedStream.windowedBy, was das zeitbasierte Aggregieren erm\u00f6glicht. Abh\u00e4ngig von der Fensterart wird entweder TimeWindowedKStream oder SessionWindowedKStream zur\u00fcckgegeben;<\/li>\n<li>Z\u00e4hlen der Transaktionen f\u00fcr die Aggregationsoperation. Der fenstergebundene Datenstrom bestimmt, ob ein bestimmter Datensatz bei dieser Z\u00e4hlung ber\u00fccksichtigt wird;<\/li>\n<li>Aufzeichnung der Ergebnisse in das Thema oder Ausgabe in die Konsole w\u00e4hrend der Entwicklung.<\/li>\n<\/ul>\n<p>\nDie Topologie dieser Anwendung ist einfach, aber eine anschauliche Darstellung ist hilfreich. Schauen wir uns Abb. 5.11 an.<\/p>\n<p>Im Folgenden werden wir die Funktionalit\u00e4t der Fensteroperationen und den entsprechenden Code betrachten.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/67d9d8d8acb1370a7c7fe5cd9a8b6aa7.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<h3>Fenstertypen<\/h3>\n<p>\nIn Kafka Streams gibt es drei Arten von Fenstern:<\/p>\n<ul>\n<li>Sitzungsfenster;<\/li>\n<li>\u201eTumbling\u201c Fenster;<\/li>\n<li>Gleitende\/\u201eH\u00fcpfende\u201c Fenster (sliding\/hopping).<\/li>\n<\/ul>\n<p>\nWelches gew\u00e4hlt wird, h\u00e4ngt von den gesch\u00e4ftlichen Anforderungen ab. \u201eTumbling\u201c und \u201eH\u00fcpfende\u201c Fenster sind zeitlich begrenzt, w\u00e4hrend die Einschr\u00e4nkungen von Sitzungsfenstern von Benutzeraktionen abh\u00e4ngen \u2014 die Dauer der Sitzung(en) wird ausschlie\u00dflich durch die Aktivit\u00e4t des Benutzers bestimmt. Wichtig ist zu beachten, dass alle Fenstertypen auf den Zeitstempeln der Datens\u00e4tze basieren und nicht auf der Systemzeit.<\/p>\n<p>Im Folgenden werden wir unsere Topologie f\u00fcr jeden Fenstertyp implementieren. Der vollst\u00e4ndige Code wird nur im ersten Beispiel angegeben; f\u00fcr andere Fenstertypen \u00e4ndert sich nichts au\u00dfer der Art der Fensteroperation.<\/p>\n<h3>Sitzungsfenster<\/h3>\n<p>\nSitzungsfenster unterscheiden sich stark von allen anderen Fensterarten. Sie sind nicht so sehr zeitlich eingeschr\u00e4nkt, sondern h\u00e4ngen von der Aktivit\u00e4t des Benutzers (oder der Entit\u00e4t, die Sie verfolgen m\u00f6chten) ab. Sitzungsfenster werden durch Phasen der Inaktivit\u00e4t abgegrenzt.<\/p>\n<p>Abbildung 5.12 veranschaulicht das Konzept der Sitzungsfenster. Eine k\u00fcrzere Sitzung wird mit der Sitzung links von ihr verschmelzen. Die Sitzung rechts wird separat sein, da sie auf eine l\u00e4ngere Phase der Inaktivit\u00e4t folgt. Sitzungsfenster basieren auf Benutzeraktivit\u00e4ten, verwenden jedoch Zeitstempel aus den Aufzeichnungen, um zu bestimmen, zu welcher Sitzung der Datensatz geh\u00f6rt.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/d39a5db7af7d6aa2194802622b2b47fd.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<h3>Die Verwendung von Sitzungsfenstern zur Verfolgung von B\u00f6rsentransaktionen<\/h3>\n<p>\nWir nutzen Sitzungsfenster, um Informationen \u00fcber B\u00f6rsentransaktionen 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\u00f6nnen).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/2dcbd9a36baec0e746aad165121451b3.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDie meisten Operationen dieser Topologie haben Sie bereits kennengelernt, sodass es nicht n\u00f6tig ist, sie hier erneut zu betrachten. Es gibt jedoch auch einige neue Elemente, die wir jetzt besprechen werden.<\/p>\n<p>Bei jeder groupBy-Operation wird normalerweise eine Aggregationsoperation (Aggregation, Reduzierung oder Z\u00e4hlung) durchgef\u00fchrt. Es kann entweder eine kumulative Aggregation mit laufendem Ergebnis oder eine Fensteraggregation durchgef\u00fchrt werden, bei der Datens\u00e4tze innerhalb eines bestimmten Zeitfensters ber\u00fccksichtigt werden.<\/p>\n<p>Der Code aus Listing 5.5 z\u00e4hlt die Anzahl der Transaktionen innerhalb der Sitzungsfenster. In Abb. 5.13 werden diese Aktionen Schritt f\u00fcr Schritt analysiert.<\/p>\n<p>Mit dem Aufruf windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) erstellen wir ein Sitzungsfenster mit einer Inaktivit\u00e4tsperiode von 20 Sekunden und einer Behaltedauer von 15 Minuten. Die Inaktivit\u00e4tsperiode 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.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/9bd47b04698086872fd135b4c67eb938.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nAls N\u00e4chstes geben wir an, welche Aggregationsoperation im Sitzungsfenster ausgef\u00fchrt werden soll \u2013 in diesem Fall count. Wenn der eingehende Datensatz au\u00dferhalb des Inaktivit\u00e4tsintervalls (von beiden Seiten des Zeitstempels) liegt, erstellt die Anwendung eine neue Sitzung. Das Aufrechterhaltungsintervall bedeutet, dass die Sitzung f\u00fcr einen bestimmten Zeitraum aktiv bleibt und versp\u00e4tete Daten erlaubt, die \u00fcber den Zeitraum der Sitzung hinausgehen, aber immer noch angeh\u00e4ngt werden k\u00f6nnen. Au\u00dferdem entsprechen Beginn und Ende der neuen Sitzung, die aus der Zusammenf\u00fchrung resultiert, dem fr\u00fchesten und sp\u00e4testen Zeitstempel.<\/p>\n<p>Betrachten wir einige Datens\u00e4tze aus der Methode count, um zu sehen, wie Sitzungen funktionieren (Tab. 5.1).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/ec04aae466d88c2d2349474069c8d541.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nBeim Eintreffen von Datens\u00e4tzen suchen wir nach bereits bestehenden Sitzungen mit dem gleichen Schl\u00fcssel, deren Endzeit weniger als der aktuelle Zeitstempel ist \u2013 Inaktivit\u00e4tsintervall und deren Startzeit gr\u00f6\u00dfer als der aktuelle Zeitstempel + Inaktivit\u00e4tsintervall. Unter Ber\u00fccksichtigung dessen werden vier Datens\u00e4tze aus Tab. 5.1 zu einer Sitzung zusammengef\u00fchrt.<\/p>\n<p>1. Zuerst trifft Datensatz 1 ein, sodass die Startzeit gleich der Endzeit ist und 00:00:00 entspricht.<\/p>\n<p>2. Als N\u00e4chstes trifft Datensatz 2 ein, und wir suchen nach Sitzungen, die nicht fr\u00fcher als 23:59:55 enden und nicht sp\u00e4ter als 00:00:35 beginnen. Wir finden Datensatz 1 und fusionieren die Sitzungen 1 und 2. Wir nehmen die Startzeit von Sitzung 1 (fr\u00fcher) und die Endzeit von Sitzung 2 (sp\u00e4ter), sodass unsere neue Sitzung um 00:00:00 beginnt und um 00:00:15 endet.<\/p>\n<p>3. Datensatz 3 trifft ein, wir suchen nach Sitzungen zwischen 00:00:30 und 00:01:10 und finden keine. Wir f\u00fcgen eine zweite Sitzung f\u00fcr den Schl\u00fcssel 123-345-654,FFBE hinzu, die um 00:00:50 beginnt und endet.<\/p>\n<p>4. Datensatz 4 trifft ein, und wir suchen nach Sitzungen zwischen 23:59:45 und 00:00:25. Dieses Mal finden wir beide Sitzungen \u2013 1 und 2. Alle drei Sitzungen werden zu einer zusammengef\u00fchrt, mit einer Startzeit von 00:00:00 und einer Endzeit von 00:00:15.<\/p>\n<p>Aus den Ausf\u00fchrungen in diesem Abschnitt sind folgende wichtige Aspekte zu merken:<\/p>\n<ul>\n<li>Sitzungen sind keine Fenster fester Gr\u00f6\u00dfe. Die Dauer einer Sitzung wird durch die Aktivit\u00e4t innerhalb eines bestimmten Zeitrahmens bestimmt;<\/li>\n<li>Zeitstempel in den Daten bestimmen, ob ein Ereignis in eine bestehende Sitzung oder in das Inaktivit\u00e4tsintervall f\u00e4llt.<\/li>\n<\/ul>\n<p>\nAls N\u00e4chstes besprechen wir die n\u00e4chste Art von Fenstern \u2013 \u201erollende\u201c Fenster.<\/p>\n<h3>\u201eRollende\u201c Fenster<\/h3>\n<p>\n\u201eTumbling\u201c-Fenster erfassen Ereignisse, die in einem bestimmten Zeitfenster auftreten. Stellen Sie sich vor, Sie m\u00fcssten alle B\u00f6rsentransaktionen eines Unternehmens alle 20 Sekunden erfassen, sodass Sie alle Ereignisse f\u00fcr diesen Zeitraum sammeln. Am Ende des 20-Sekunden-Intervalls \u201etumbelt\u201c das Fenster und wechselt zu einem neuen 20-Sekunden-Beobachtungsintervall. Abbildung 5.14 illustriert diese Situation.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/e90e560d9ddda5e2e524b7c387ad9874.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nWie Sie sehen k\u00f6nnen, sind alle in den letzten 20 Sekunden eingegangenen Ereignisse im Fenster enthalten. Nach Ablauf dieses Zeitraums wird ein neues Fenster erstellt.<\/p>\n<p>Im Listing 5.6 finden Sie den Code, der die Verwendung von \u201etumbling\u201c Fenstern zur Erfassung von B\u00f6rsentransaktionen alle 20 Sekunden demonstriert (er ist in der Datei src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java zu finden).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/e055bb1b288c7d500b64372fe3fbf064.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDurch diese kleine \u00c4nderung des Aufrufs der Methode TimeWindows.of kann ein \u201etumbling\u201c Fenster verwendet werden. In diesem Beispiel gibt es keinen Aufruf der Methode until(), weshalb das Standard-Speicherintervall von 24 Stunden verwendet wird.<\/p>\n<p>Schlie\u00dflich ist es an der Zeit, zur letzten Fensteroption \u00fcberzugehen \u2013 \u201ehopping\u201c Fenstern.<\/p>\n<h3>Gleitende (\u201ehopping\u201c) Fenster<\/h3>\n<p>\nGleitende\/\u201ehopping\u201c (sliding\/hopping) Fenster \u00e4hneln den \u201etumbling\u201c 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\u00fcrzeren Warteintervall als der Fensterlaufzeit.<\/p>\n<p>Um die Unterschiede zwischen \u201etumbling\u201c und \u201ehopping\u201c Fenstern zu veranschaulichen, kehren wir zum Beispiel mit der Z\u00e4hlung der B\u00f6rsentransaktionen zur\u00fcck. Unser Ziel bleibt die Z\u00e4hlung der Transaktionen, aber wir m\u00f6chten nicht den gesamten Zeitraum abwarten, bevor wir den Z\u00e4hler aktualisieren. Stattdessen werden wir den Z\u00e4hler in k\u00fcrzeren Zeitabschnitten aktualisieren. Zum Beispiel z\u00e4hlen wir weiterhin die Transaktionen alle 20 Sekunden, aber wir aktualisieren den Z\u00e4hler alle 5 Sekunden, wie in Abb. 5.15 gezeigt. Dabei haben wir drei Ergebnisfenster mit \u00fcberlappenden Daten.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/415c8cd9f2b60d453a1a01c3bc99331f.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nIm 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).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/ab2d1a64380d256d3fb084e16597417c.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nEin \"wackelndes\" Fenster kann durch einen Aufruf der Methode advanceBy() in ein \"springendes\" Fenster umgewandelt werden. In dem folgenden Beispiel betr\u00e4gt das Intervall der Speicherung 15 Minuten.<\/p>\n<p>In diesem Abschnitt haben Sie gesehen, wie man die Ergebnisse der Aggregation mithilfe von Zeitfenstern einschr\u00e4nken kann. Insbesondere sollten Sie sich aus diesem Abschnitt die folgenden drei Dinge merken:<\/p>\n<ul>\n<li>Die Gr\u00f6\u00dfe der Sitzungsfenster wird nicht durch einen Zeitintervall, sondern durch die Aktivit\u00e4t der Benutzer begrenzt;<\/li>\n<li>\u201eWackelnde\u201c Fenster geben Einblick in Ereignisse innerhalb eines bestimmten Zeitraums;<\/li>\n<li>Die Dauer der \u201espringenden\u201c Fenster ist festgelegt, aber sie werden h\u00e4ufig aktualisiert und k\u00f6nnen in allen Fenstern sich \u00fcberschneidende Eintr\u00e4ge enthalten.<\/li>\n<\/ul>\n<p>\nAls N\u00e4chstes erfahren wir, wie man KTable zur\u00fcck in KStream umwandelt, um Verbindungen herzustellen.<\/p>\n<h3>5.3.3. Verbindung von KStream- und KTable-Objekten<\/h3>\n<p>\nIn 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\u00e4gen, w\u00e4hrend KTable ein Strom von Aktualisierungen der Eintr\u00e4ge ist, aber manchmal kann es notwendig sein, dem Eintragsstrom zus\u00e4tzlichen Kontext durch Aktualisierungen aus KTable hinzuzuf\u00fcgen.<\/p>\n<p>Nehmen wir Daten \u00fcber die Anzahl von B\u00f6rsentransaktionen und verbinden sie mit B\u00f6rsennachrichten aus den entsprechenden Branchen. Hier ist, was zu tun ist, um dies unter Ber\u00fccksichtigung des bereits vorhandenen Codes zu erreichen.<\/p>\n<ol>\n<li>Das KTable-Objekt mit den Daten zur Anzahl der B\u00f6rsentransaktionen in KStream umwandeln und den Schl\u00fcssel durch einen Schl\u00fcssel ersetzen, der die entsprechende Branche f\u00fcr das jeweilige Wertpapier bezeichnet.<\/li>\n<li>Ein KTable-Objekt erstellen, das Daten aus dem Thema mit den B\u00f6rsennachrichten liest. Dieses neue KTable wird nach Branchen kategorisiert.<\/li>\n<li>Aktualisierungen der Nachrichten mit Informationen \u00fcber die Anzahl der B\u00f6rsentransaktionen nach Branchen verbinden.<\/li>\n<\/ol>\n<p>\nJetzt sehen wir uns an, wie wir diesen Aktionsplan umsetzen k\u00f6nnen.<\/p>\n<h3>Umwandlung von KTable in KStream<\/h3>\n<p>\nF\u00fcr die Umwandlung von KTable in KStream sind die folgenden Schritte erforderlich.<\/p>\n<ol>\n<li>Die Methode KTable.toStream() aufrufen.<\/li>\n<li>Durch den Aufruf der Methode KStream.map den Schl\u00fcssel durch den Namen der Branche ersetzen und dann aus der Instanz Windowed das Objekt TransactionSummary extrahieren.<\/li>\n<\/ol>\n<p>\nWir werden diese Operationen in einer Kette wie folgt verkn\u00fcpfen (der Code ist in der Datei src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java zu finden) (Listing 5.8).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/0d43c2650f6e66e2816ed383da3a29c2.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDa wir die Operation KStream.map ausf\u00fchren, erfolgt die erneute Partitionierung der zur\u00fcckgegebenen KStream-Instanz automatisch bei deren Verwendung in einer Verbindung.<\/p>\n<p>Wir haben den Transformationsprozess abgeschlossen, jetzt m\u00fcssen wir ein KTable-Objekt erstellen, um B\u00f6rsennachrichten zu lesen.<\/p>\n<h3>Erstellung eines KTable f\u00fcr B\u00f6rsennachrichten<\/h3>\n<p>\nGl\u00fccklicherweise 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).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/6e83a393fdc9ab74fda4cbdddddb5213.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nEs ist anzumerken, dass keine Serde-Objekte angegeben werden m\u00fcssen, da in den Einstellungen stringbasierte Serde verwendet werden. Zudem wird dank der Verwendung der Enumeration EARLIEST die Tabelle zu Beginn mit Datens\u00e4tzen gef\u00fcllt.<\/p>\n<p>Jetzt k\u00f6nnen wir zum letzten Schritt \u00fcbergehen \u2014 der Verbindung.<\/p>\n<h3>Verbindung von Nachrichtenupdates mit Transaktionsdaten<\/h3>\n<p>\nDie Erstellung einer Verbindung ist unkompliziert. Wir werden einen linken Join verwenden, falls in der entsprechenden Branche keine B\u00f6rsennachrichten vorliegen (den ben\u00f6tigten Code finden Sie in der Datei src\/main\/java\/bbejeck\/chapter_5\/CountingWindowingAndKtableJoinExample.java) (Listing 5.10).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/92ed70f98927d2f778ad14dbc2a5aa26.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nDieser 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\u00fcr jeden Schl\u00fcssel 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\u00e4ufig aktualisierten Referenzdaten angereichert werden.<\/p>\n<p>Jetzt betrachten wir einen effizienteren Weg, um Ereignisse aus KStream anzureichern.<\/p>\n<h3>5.3.4. Objekte GlobalKTable<\/h3>\n<p>\nWie Sie verstanden haben, besteht die Notwendigkeit, Ereignisstr\u00f6me anzureichern oder ihnen Kontext hinzuzuf\u00fcgen. In Kapitel 4 haben Sie die Verbindung zweier KStream-Objekte gesehen, und im vorherigen Abschnitt \u2014 die Verbindung zwischen KStream und KTable. In all diesen F\u00e4llen ist eine erneute Partitionierung des Datenstroms erforderlich, wenn die Schl\u00fcssel 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\u00fcssel ge\u00e4ndert haben und die Datens\u00e4tze in neuen Partitionen landen m\u00fcssen, andernfalls w\u00e4re eine Verbindung unm\u00f6glich (dies wurde in Kapitel 4 im Abschnitt \u201eErneute Partitionierung von Daten\u201c unter Abschnit 4.2.4 behandelt).<\/p>\n<h3>Wiederholte Partitionierung hat ihren Preis<\/h3>\n<p>\nWiederholte Partitionierung erfordert Kosten \u2013 zus\u00e4tzliche Ressourcen zur Erstellung von Zwischen-Topics, die Speicherung redundanter Daten in einem weiteren Topic; sie bedeutet auch eine Erh\u00f6hung der Latenz durch das Schreiben und Lesen aus diesem Topic. Dar\u00fcber hinaus m\u00fcssen bei der Notwendigkeit, mehr als einen Aspekt oder eine Dimension zu verbinden, Verbindungen in einer Kette organisiert, die Eintr\u00e4ge mit neuen Schl\u00fcsseln abgebildet und der Prozess der Wiederholten Partitionierung erneut durchgef\u00fchrt werden.<\/p>\n<h3>Verbindung zu Datenmengen kleinerer Gr\u00f6\u00dfe<\/h3>\n<p>\nIn einigen F\u00e4llen ist das Volumen der Referenzdaten, mit denen eine Verbindung geplant ist, relativ klein, sodass vollst\u00e4ndige Kopien davon lokal auf jedem der Knoten Platz finden k\u00f6nnen. F\u00fcr solche Situationen bietet Kafka Streams die Klasse GlobalKTable.<\/p>\n<p>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\u00fcssel der Referenzdaten zu partitionieren, damit er allen Partitionen zur Verf\u00fcgung steht. Mit GlobalKTable-Objekten k\u00f6nnen auch schl\u00fcssellose Verbindungen durchgef\u00fchrt werden. Lassen Sie uns zu einem der vorherigen Beispiele zur\u00fcckkehren, um diese M\u00f6glichkeit zu demonstrieren.<\/p>\n<h3>Verbindung von KStream-Objekten mit GlobalKTable-Objekten<\/h3>\n<p>\nIm Abschnitt 5.3.2 haben wir eine Fensteraggregation von B\u00f6rsentransaktionen nach Kunden durchgef\u00fchrt. Die Ergebnisse dieser Aggregation sahen ungef\u00e4hr folgenderma\u00dfen aus:<\/p>\n<pre><code class=\"plaintext\">{customerId='074-09-3705', stockTicker='GUTM'}, 17\n{customerId='037-34-5184', stockTicker='CORK'}, 16<\/code><\/pre>\n<p>\nObwohl diese Ergebnisse dem angestrebten Ziel entsprachen, w\u00e4re es praktischer, wenn auch der Name des Kunden und der vollst\u00e4ndige Name des Unternehmens ausgegeben w\u00fcrden. Um den Namen des K\u00e4ufers und den Namen des Unternehmens hinzuzuf\u00fcgen, k\u00f6nnen herk\u00f6mmliche Verbindungen hergestellt werden, dabei sind jedoch zwei Schl\u00fcsselabbildungen und eine erneute Partitionierung erforderlich. Mit GlobalKTable k\u00f6nnen solche Kosten vermieden werden.<\/p>\n<p>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.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/fc4d91bbe062ceb94f5650224840b81e.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nWir haben das bereits fr\u00fcher besprochen, deshalb will ich mich nicht wiederholen. Ich m\u00f6chte jedoch darauf hinweisen, dass der Code in der Funktion toStream().map zur besseren Lesbarkeit in ein Funktionsobjekt abstrahiert wurde, anstelle einer eingebetteten Lambda-Ausdrucks.<\/p>\n<p>Der n\u00e4chste 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).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/db3918895f174c8cfb5fc927b55f87a1.jpg\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<p>Bitte beachten Sie, dass die Namen der Themen mithilfe von Enumerationen beschrieben werden.<\/p>\n<p>Jetzt, da wir alle Komponenten vorbereitet haben, bleibt nur noch der Code f\u00fcr die Verkn\u00fcpfung zu schreiben (der in der Datei src\/main\/java\/bbejeck\/chapter_5\/GlobalKTableExample.java zu finden ist) (Listing 5.13).<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/799360cc99f1920c190a61fd4685d4ff.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nObwohl der oben genannte Code zwei Verkn\u00fcpfungen enth\u00e4lt, sind sie in einer Kette organisiert, da keiner ihrer Ergebnisse separat verwendet wird. Die Ergebnisse werden am Ende der gesamten Operation ausgegeben.<\/p>\n<p>Wenn Sie die oben genannte Verkn\u00fcpfungsoperation ausf\u00fchren, erhalten Sie Ergebnisse in der folgenden Form:<\/p>\n<pre><code class=\"plaintext\">{customer='Barney, Smith' company=\"Exxon\", transactions= 17}<\/code><\/pre>\n<p>\nDer Kern hat sich nicht ver\u00e4ndert, aber diese Ergebnisse sind verst\u00e4ndlicher.<\/p>\n<p>Wenn man Kapitel 4 z\u00e4hlt, haben Sie bereits mehrere Arten von Verkn\u00fcpfungen in Aktion gesehen. Diese sind in Tabelle 5.2 aufgef\u00fchrt. Diese Tabelle zeigt die Verkn\u00fcpfungsm\u00f6glichkeiten, die f\u00fcr die Version 1.0.0 von Kafka Streams relevant sind; in zuk\u00fcnftigen Versionen k\u00f6nnte sich etwas \u00e4ndern.<\/p>\n<p><img decoding=\"async\" alt=\"Buch \u201eKafka Streams in Action. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c\" src=\"\/wp-content\/uploads\/8e4cf35c64a8431bda43a5e748de275f.jpg\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nZusammenfassend m\u00f6chte ich das Wesentliche in Erinnerung rufen: Sie k\u00f6nnen Ereignisstr\u00f6me (KStream) und Aktualisierungsstr\u00f6me (KTable) mit lokalem Status verkn\u00fcpfen. Au\u00dferdem kann, sofern die Gr\u00f6\u00dfe der Referenzdaten nicht zu gro\u00df ist, das Objekt GlobalKTable verwendet werden. GlobalKTable repliziert alle Partitionen auf jeden Knoten der Kafka Streams-Anwendung und sorgt so f\u00fcr die Verf\u00fcgbarkeit aller Daten, unabh\u00e4ngig davon, zu welcher Partition der Schl\u00fcssel geh\u00f6rt.<\/p>\n<p>Als n\u00e4chstes werden wir die M\u00f6glichkeit von Kafka Streams sehen, die es erlaubt, Status\u00e4nderungen zu beobachten, ohne Daten aus dem Kafka-Topic zu konsumieren.<\/p>\n<h3>5.3.5. Abfragef\u00e4higer Status<\/h3>\n<p>\nWir haben bereits mehrere Operationen mit Status durchgef\u00fchrt und die Ergebnisse immer in der Konsole ausgegeben (zu Entwicklungszwecken) oder in einem Topic gespeichert (f\u00fcr industrielle Anwendungen). Bei der Speicherung von Ergebnissen in einem Topic m\u00fcssen Sie einen Kafka-Verbraucher verwenden, um sie anzuzeigen.<\/p>\n<p>Das Lesen von Daten aus diesen Themen kann als eine Form von materialisierten Sichten (materialized views) angesehen werden. F\u00fcr unsere Zwecke kann die Definition einer materialisierten Sicht aus der \u201eWikipedia\u201c verwendet werden: \u201e\u2026ein physisches Objekt einer Datenbank, das die Ergebnisse der Ausf\u00fchrung einer Abfrage enth\u00e4lt. 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\u201c (https:\/\/de.wikipedia.org\/wiki\/Materialisierte_Sicht).<\/p>\n<p>Kafka Streams erm\u00f6glicht auch die Ausf\u00fchrung interaktiver Abfragen (interactive queries) an den Zustands-Speichern, was die direkte Lesbarkeit dieser materialisierten Sichten erm\u00f6glicht. Es ist wichtig zu beachten, dass eine Abfrage an den Zustands-Speicher eine \u201eNur-Lese\u201c-Operation ist. Dank dessen m\u00fcssen Sie sich keine Sorgen machen, die Konsistenz des Zustands w\u00e4hrend der Datenverarbeitung durch die Anwendung versehentlich zu st\u00f6ren.<\/p>\n<p>Die M\u00f6glichkeit von direkten Abfragen an den Zustands-Speichern ist von gro\u00dfer Bedeutung. Sie bedeutet, dass Sie Anwendungen \u2013 Dashboards \u2013 erstellen k\u00f6nnen, 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:<\/p>\n<ul>\n<li>durch die Lokalisierung der Daten kann schnell darauf zugegriffen werden;<\/li>\n<li>Daten-Duplikate werden ausgeschlossen, da sie nicht in einem externen Speicher gespeichert werden.<\/li>\n<\/ul>\n<p>\nDas Wichtigste, was ich m\u00f6chte, dass Sie sich merken: Sie k\u00f6nnen direkt Abfragen an den Zustand aus der Anwendung ausf\u00fchren. Die M\u00f6glichkeiten, die sich daraus ergeben, k\u00f6nnen nicht hoch genug eingesch\u00e4tzt werden. Anstatt Daten aus Kafka zu konsumieren und Datens\u00e4tze in einer Datenbank f\u00fcr die Anwendung zu speichern, k\u00f6nnen Abfragen an die Zustands-Speicher durchgef\u00fchrt 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).<\/p>\n<p>In diesem Kapitel haben wir eine betr\u00e4chtliche Menge an Informationen behandelt, daher pausieren wir vorerst unser Gespr\u00e4ch \u00fcber interaktive Abfragen an Zustandsdatenbanken. Aber keine Sorge: In Kapitel 9 werden wir eine einfache Anwendung erstellen \u2013 ein Dashboard mit interaktiven Abfragen. Zur Demonstration interaktiver Abfragen und der M\u00f6glichkeiten, sie in Kafka Streams-Anwendungen einzuf\u00fcgen, werden einige Beispiele aus diesem sowie aus dem vorherigen Kapitel verwendet.<\/p>\n<h3>Zusammenfassung<\/h3>\n<p><\/p>\n<ul>\n<li>KStream-Objekte repr\u00e4sentieren Ereignisstr\u00f6me, \u00e4hnlich wie Einf\u00fcgungen in einer Datenbank. KTable-Objekte hingegen repr\u00e4sentieren Aktualisierungsstr\u00f6me und \u00e4hneln mehr den Aktualisierungen in einer Datenbank. Die Gr\u00f6\u00dfe eines KTable-Objekts w\u00e4chst nicht, alte Eintr\u00e4ge werden durch neue ersetzt.<\/li>\n<li>KTable-Objekte sind erforderlich f\u00fcr Aggregationsoperationen.<\/li>\n<li>Mit Fensteroperationen k\u00f6nnen aggregierte Daten in zeitliche Zeitfenster unterteilt werden.<\/li>\n<li>Durch GlobalKTable-Objekte kann auf Referenzdaten an beliebigen Stellen der Anwendung zugegriffen werden, unabh\u00e4ngig von der Partitionierung.<\/li>\n<li>Es k\u00f6nnen Verbindungen zwischen KStream-, KTable- und GlobalKTable-Objekten hergestellt werden.<\/li>\n<\/ul>\n<p>\nBis 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\u00e4gnanter Programme erm\u00f6glicht, stellt seine Anwendung einen bestimmten Kompromiss dar. Die Arbeit mit der DSL KStream bedeutet mehr Pr\u00e4gnanz des Codes auf Kosten des Kontrolle. Im n\u00e4chsten Kapitel werden wir die Low-Level-API von Processing Nodes betrachten und probieren andere Kompromisse aus. Die Programme werden l\u00e4nger sein als bisher, aber wir werden die M\u00f6glichkeit haben, praktisch jeden ben\u00f6tigten Processing Node zu erstellen.<\/p>\n<p>\u2192 Mehr Details zur Buch finden Sie unter <noindex><a rel=\"nofollow\" href=\"https:\/\/www.piter.com\/collection\/best\/product\/kafka-streams-v-deystvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\">der Verlagswebsite<\/a><\/noindex><\/p>\n<p>\u2192 F\u00fcr Habr-Leser gibt es 25% Rabatt mit dem Coupon \u2014 <b>Kafka Streams<\/b><\/p>\n<p>\u2192 Nach Zahlung f\u00fcr die gedruckte Buchausgabe wird das E-Book per E-Mail versendet.<br \/>\n<br \/>Quelle: <a content=\"nofollow\" rel=\"nofollow\" href=\"https:\/\/habr.com\/ru\/company\/piter\/blog\/457756\/\">habr.com<\/a><\/p>","protected":false,"gt_translate_keys":[{"key":"rendered","format":"html"}]},"excerpt":{"rendered":"<p>\u041f\u0440\u0438\u0432\u0435\u0442, \u0425\u0430\u0431\u0440\u043e\u0436\u0438\u0442\u0435\u043b\u0438! \u042d\u0442\u0430 \u043a\u043d\u0438\u0433\u0430 \u043f\u043e\u0434\u043e\u0439\u0434\u0435\u0442 \u0434\u043b\u044f \u043b\u044e\u0431\u043e\u0433\u043e \u0440\u0430\u0437\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a\u0430, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0445\u043e\u0447\u0435\u0442 \u0440\u0430\u0437\u043e\u0431\u0440\u0430\u0442\u044c\u0441\u044f \u0432 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u0439 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0435. \u041f\u043e\u043d\u0438\u043c\u0430\u043d\u0438\u0435 \u0440\u0430\u0441\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u043d\u043e\u0433\u043e \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f \u043f\u043e\u043c\u043e\u0436\u0435\u0442 \u043b\u0443\u0447\u0448\u0435 \u0438\u0437\u0443\u0447\u0438\u0442\u044c Kafka \u0438 Kafka Streams. \u0411\u044b\u043b\u043e \u0431\u044b \u043d\u0435\u043f\u043b\u043e\u0445\u043e \u0437\u043d\u0430\u0442\u044c \u0438 \u0441\u0430\u043c \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a Kafka, \u043d\u043e \u044d\u0442\u043e \u043d\u0435 \u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e: \u044f \u0440\u0430\u0441\u0441\u043a\u0430\u0436\u0443 \u0432\u0430\u043c \u0432\u0441\u0435, \u0447\u0442\u043e \u043d\u0443\u0436\u043d\u043e. \u041e\u043f\u044b\u0442\u043d\u044b\u0435 \u0440\u0430\u0437\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a\u0438 Kafka, \u043a\u0430\u043a \u0438 \u043d\u043e\u0432\u0438\u0447\u043a\u0438, \u0431\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f \u044d\u0442\u043e\u0439 \u043a\u043d\u0438\u0433\u0435 \u043e\u0441\u0432\u043e\u044f\u0442 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u0435 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u043d\u044b\u0445 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 [&hellip;]<\/p>\n","protected":false,"gt_translate_keys":[{"key":"rendered","format":"html"}]},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[688],"tags":[],"class_list":["post-35786","post","type-post","status-publish","format-standard","hentry","category-administrirovanie"],"aioseo_notices":[],"aioseo_head":"\n\t\t<!-- All in One SEO 5.0.1.1 - aioseo.com -->\n\t<meta name=\"robots\" content=\"max-image-preview:large\" \/>\n\t<meta name=\"author\" content=\"Yuri Gagarin\"\/>\n\t<link rel=\"canonical\" href=\"https:\/\/prohoster.info\/de\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\" \/>\n\t<meta name=\"generator\" content=\"All in One SEO (AIOSEO) 5.0.1.1\" \/>\n\t\t<meta property=\"og:locale\" content=\"de_DE\" \/>\n\t\t<meta property=\"og:site_name\" content=\"ProHoster | \u041a\u0443\u043f\u0438\u0442\u044c \u043d\u0430\u0434\u0435\u0436\u043d\u044b\u0439 \u0445\u043e\u0441\u0442\u0438\u043d\u0433 \u0434\u043b\u044f \u0441\u0430\u0439\u0442\u043e\u0432 \u0441 \u0437\u0430\u0449\u0438\u0442\u043e\u0439 \u043e\u0442 DDoS, VPS VDS \u0441\u0435\u0440\u0432\u0435\u0440\u044b\" \/>\n\t\t<meta property=\"og:type\" content=\"article\" \/>\n\t\t<meta property=\"og:title\" content=\"\ud83e\udd47\u041a\u043d\u0438\u0433\u0430 \u00abKafka Streams \u0432 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0438. \u041f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f \u0438 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438\u00bb | ProHoster\" \/>\n\t\t<meta property=\"og:url\" content=\"https:\/\/prohoster.info\/de\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni\" \/>\n\t\t<meta property=\"og:image\" content=\"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg\" \/>\n\t\t<meta property=\"og:image:secure_url\" content=\"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg\" \/>\n\t\t<meta property=\"og:image:width\" content=\"350\" \/>\n\t\t<meta property=\"og:image:height\" content=\"350\" \/>\n\t\t<meta property=\"article:published_time\" content=\"2019-10-31T19:06:19+00:00\" \/>\n\t\t<meta property=\"article:modified_time\" content=\"2019-10-31T19:06:19+00:00\" \/>\n\t\t<meta property=\"article:publisher\" content=\"https:\/\/www.facebook.com\/prohoster\" \/>\n\t\t<meta property=\"article:author\" content=\"https:\/\/www.facebook.com\/prohoster\" \/>\n\t\t<!-- All in One SEO -->\n\n","aioseo_head_json":{"title":"\ud83e\udd47Buch \u201eKafka Streams in Aktion. Anwendungen und Mikrodienste f\u00fcr die Echtzeitverarbeitung\u201c | ProHoster","description":"","canonical_url":"https:\/\/prohoster.info\/de\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","robots":"max-image-preview:large","keywords":"","webmasterTools":{"miscellaneous":""},"schema":null,"og:locale":"de_DE","og:site_name":"ProHoster | \u041a\u0443\u043f\u0438\u0442\u044c \u043d\u0430\u0434\u0435\u0436\u043d\u044b\u0439 \u0445\u043e\u0441\u0442\u0438\u043d\u0433 \u0434\u043b\u044f \u0441\u0430\u0439\u0442\u043e\u0432 \u0441 \u0437\u0430\u0449\u0438\u0442\u043e\u0439 \u043e\u0442 DDoS, VPS VDS \u0441\u0435\u0440\u0432\u0435\u0440\u044b","og:type":"article","og:title":"\ud83e\udd47\u041a\u043d\u0438\u0433\u0430 \u00abKafka Streams \u0432 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0438. \u041f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f \u0438 \u043c\u0438\u043a\u0440\u043e\u0441\u0435\u0440\u0432\u0438\u0441\u044b \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438\u00bb | ProHoster","og:url":"https:\/\/prohoster.info\/de\/blog\/administrirovanie\/kniga-kafka-streams-v-dejstvii-prilozheniya-i-mikroservisy-dlya-raboty-v-realnom-vremeni","og:image":"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg","og:image:secure_url":"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg","og:image:width":350,"og:image:height":350,"article:published_time":"2019-10-31T19:06:19+00:00","article:modified_time":"2019-10-31T19:06:19+00:00","article:publisher":"https:\/\/www.facebook.com\/prohoster","article:author":"https:\/\/www.facebook.com\/prohoster"},"aioseo_meta_data":{"post_id":"35786","title":null,"description":null,"keywords":null,"keyphrases":null,"primary_term":null,"canonical_url":null,"og_title":null,"og_description":null,"og_object_type":"default","og_image_type":"default","og_image_url":null,"og_image_width":null,"og_image_height":null,"og_image_custom_url":null,"og_image_custom_fields":null,"og_video":null,"og_custom_url":null,"og_article_section":null,"og_article_tags":null,"twitter_use_og":false,"twitter_card":"default","twitter_image_type":"default","twitter_image_url":null,"twitter_image_custom_url":null,"twitter_image_custom_fields":null,"twitter_title":null,"twitter_description":null,"schema":{"blockGraphs":[],"customGraphs":[],"default":{"data":{"Article":[],"Course":[],"Dataset":[],"FAQPage":[],"Movie":[],"Person":[],"Product":[],"ProductReview":[],"Car":[],"Recipe":[],"Service":[],"SoftwareApplication":[],"WebPage":[]},"graphName":"","isEnabled":true},"graphs":[]},"schema_type":null,"schema_type_options":null,"pillar_content":false,"robots_default":true,"robots_noindex":false,"robots_noarchive":false,"robots_nosnippet":false,"robots_nofollow":false,"robots_noimageindex":false,"robots_noodp":false,"robots_notranslate":false,"robots_max_snippet":null,"robots_max_videopreview":null,"robots_max_imagepreview":"large","priority":null,"frequency":null,"local_seo":null,"seo_analyzer_scan_date":"2026-01-22 00:45:19","breadcrumb_settings":null,"limit_modified_date":false,"reviewed_by":null,"ai":null,"created":"2021-03-01 01:56:32","updated":"2026-01-22 00:45:19","focus_keyword":null,"additional_keywords":null,"truseo_locale":null},"gt_translate_keys":[{"key":"link","format":"url"}],"_links":{"self":[{"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/posts\/35786","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/comments?post=35786"}],"version-history":[{"count":0,"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/posts\/35786\/revisions"}],"wp:attachment":[{"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/media?parent=35786"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/categories?post=35786"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/prohoster.info\/de\/wp-json\/wp\/v2\/tags?post=35786"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}