Die boek "Kafka-strome in aksie. Toepassings en mikrodienste vir intydse werk"

Hallo Habrites! Hierdie boek is geskik vir enige ontwikkelaar wat streaming wil verstaan. Om verspreide programmering te verstaan, sal jou help om Kafka en Kafka Streams beter te verstaan. Dit sal lekker wees om die Kafka-raamwerk self te ken, maar dit is nie nodig nie: ek sal jou alles vertel wat jy nodig het. Ervare Kafka-ontwikkelaars sowel as beginners sal met hierdie boek leer hoe om interessante stroomtoepassings te bou deur die Kafka Streams-biblioteek te gebruik. Intermediêre tot gevorderde Java-ontwikkelaars, wat reeds vertroud is met konsepte soos serialisering, sal leer hoe om hul vaardighede toe te pas om Kafka Streams-toepassings te bou. Die boek se bronkode is in Java 8 geskryf en maak baie gebruik van die Java 8 lambda-uitdrukkingsintaksis, so om te weet hoe om met lambda-funksies te werk (selfs in 'n ander programmeertaal) sal handig te pas kom.

Uittreksel. 5.3. Aggregasie en venster bedrywighede

В этом разделе мы перейдем к изучению наиболее многообещающих частей Kafka Streams. Пока мы рассмотрели следующие аспекты Kafka Streams:

• skepping van verwerkingstopologie;
• gebruik van staat in stroomtoepassings;
• die uitvoering van datastroomverbindings;
• verskille tussen gebeurtenisstrome (KStream) en opdateringstrome (KTable).

В следующих же примерах мы соберем все эти элементы воедино. Кроме того, вы познакомитесь с оконными операциями — еще одной замечательной возможностью потоковых приложений. Первым нашим примером будет простое агрегирование.

5.3.1. Samevoeging van aandeelverkope volgens industrie

Агрегирование и группировка — жизненно необходимые инструменты при работе с потоковыми данными. Исследования отдельных записей по мере поступления часто оказывается недостаточно. Для извлечения из данных дополнительной информации необходимы их группировка и комбинирование.

В этом примере вам предстоит примерить костюм внутридневного трейдера, которому нужно отслеживать объемы продаж акций компаний в нескольких отраслях промышленности. В частности, вас интересуют пять компаний с наибольшими объемами продаж акций в каждой из отраслей промышленности.

Для подобного агрегирования потребуется несколько следующих шагов по переводу данных в нужный вид (если говорить в общих чертах).

1. Создать источник на основе топика, публикующий необработанную информацию по торговле акциями. Нам придется отобразить объект типа StockTransaction в объект типа ShareVolume. Дело в том, что объект StockTransaction содержит метаданные продаж, а нам нужны только данные о количестве продаваемых акций.
2. Сгруппировать данные ShareVolume по символам акций. После группировки по символам можно свернуть эти данные до промежуточных сумм объемов продаж акций. Стоит отметить, что метод KStream.groupBy возвращает экземпляр типа KGroupedStream. А получить экземпляр KTable можно, вызвав далее метод KGroupedStream.reduce.

Wat is die KGroupedStream-koppelvlak

Die KStream.groupBy- en KStream.groupByKey-metodes gee 'n KGroupedStream-instansie terug. KGroupedStream is 'n intermediêre voorstelling van die stroom gebeurtenisse na groepering volgens sleutels. Dit is glad nie ontwerp om direk met hom te werk nie. In plaas daarvan word KGroupedStream gebruik vir samevoegingsbewerkings wat altyd 'n KTable tot gevolg het. En aangesien die resultaat van die samevoegingsoperasies 'n KTable is en hulle 'n staatswinkel gebruik, is dit moontlik dat nie alle opdaterings in die resultaat verder in die pyplyn gestuur word nie.

Метод KTable.groupBy возвращает аналогичный KGroupedTable — промежуточное представление потока обновлений, перегруппированных по ключу.

Сделаем небольшой перерыв и посмотрим на рис. 5.9, на котором показано, чего мы добились. Эта топология должна быть вам уже хорошо знакома.

Взглянем теперь на код для этой топологии (его можно найти в файле src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (листинг 5.2).

Bogenoemde kode is opvallend vir sy beknoptheid en groot volume aksies wat in verskeie reëls uitgevoer word. Jy sal dalk iets nuuts in die eerste parameter van die builder.stream-metode opmerk: die waarde van die AutoOffsetReset.EARLIEST opgesomde tipe (daar is ook LATEST), wat met die Consumed.withOffsetResetPolicy-metode gestel word. Hierdie opgesomde tipe kan gebruik word om 'n offset-terugstellingstrategie vir elke KStream of KTable te spesifiseer, en geniet voorkeur bo die offset-terugstellinginstelling vanaf die konfigurasie.

GroupByKey en GroupBy

В интерфейсе KStream есть два метода для группировки записей: GroupByKey и GroupBy. Оба возвращают KGroupedTable, так что у вас может появиться закономерный вопрос: в чем же различие между ними и когда использовать какой из них?

Метод GroupByKey применяется, когда ключи в KStream уже непустые. А главное, флаг «требует повторного секционирования» никогда не устанавливался.

Метод GroupBy предполагает, что вы меняли ключи для группировки, так что флаг повторного секционирования установлен в true. Выполнение после метода GroupBy соединений, агрегирования и т. п. приведет к автоматическому повторному секционированию.
Opsomming: Jy moet GroupByKey gebruik, nie GroupBy nie, waar moontlik.

Что делают методы mapValues и groupBy — понятно, так что взглянем на метод sum() (его можно найти в файле src/main/java/bbejeck/model/ShareVolume.java) (листинг 5.3).

Die ShareVolume.sum-metode gee die subtotaal van die aandeelverkoopvolume terug, en die resultaat van die hele berekeningsketting is 'n KTable-objek . Nou verstaan ​​jy watter rol KTable speel. Wanneer ShareVolume-voorwerpe aankom, word die nuutste bygewerkte opdatering in die ooreenstemmende KTable-voorwerp gestoor. Dit is belangrik om te onthou dat alle opdaterings in die vorige shareVolumeKTable weerspieël word, maar nie almal word verder gestuur nie.

Далее с помощью этого KTable мы выполняем агрегирование (по количеству продаваемых акций), чтобы получить пять компаний с наибольшими объемами продаж акций в каждой из отраслей промышленности. Наши действия при этом будут аналогичны действиям при первом агрегировании.

1. Выполнить еще одну операцию groupBy для группировки отдельных объектов ShareVolume по отраслям промышленности.
2. Приступить к суммированию объектов ShareVolume. На этот раз объект агрегирования представляет собой очередь по приоритету фиксированного размера. В такой очереди фиксированного размера сохраняются только пять компаний с наибольшими количествами проданных акций.
3. Отобразить очереди из предыдущего пункта в строковое значение и вернуть пять наиболее продаваемых по количеству акций по отраслям промышленности.
4. Skryf die resultate in stringvorm by die onderwerp neer.

На рис. 5.10 показан граф топологии движения данных. Как вы видите, второй круг обработки достаточно прост.

Теперь, четко уяснив себе структуру этого второго круга обработки, можно обратиться к его исходному коду (вы найдете его в файле src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (листинг 5.4).

В данном инициализаторе есть переменная fixedQueue. Это пользовательский объект — адаптер для java.util.TreeSet, который применяется для отслеживания N наибольших результатов в порядке убывания количества проданных акций.

Вы уже встречались с вызовами groupBy и mapValues, так что не будем на них останавливаться (мы вызываем метод KTable.toStream, поскольку метод KTable.print считается устаревшим). Но вы пока еще не видели KTable-версию метода aggregate(), так что мы потратим немного времени на его обсуждение.

Soos u onthou, is KTable anders deurdat rekords met dieselfde sleutel as opdaterings beskou word. KTable vervang die ou inskrywing met die nuwe een. Aggregasie werk op 'n soortgelyke manier: die jongste rekords met dieselfde sleutel word saamgevoeg. Wanneer 'n inskrywing aankom, word dit by 'n instansie van die FixedSizePriorityQueue-klas gevoeg deur 'n adder te gebruik (die tweede parameter in die oproep na die saamgevoegde metode), maar as 'n ander inskrywing reeds bestaan ​​met dieselfde sleutel, dan word die ou inskrywing verwyder met die aftrekker (die derde parameter in die oproep na die totale metode).

Dit beteken alles dat ons aggregator, FixedSizePriorityQueue, glad nie alle waardes met een sleutel saamvoeg nie, maar 'n rollende som van N hoeveelhede van die mees verhandelde soorte aandele stoor. Elke inkomende inskrywing bevat die totale aantal aandele wat tot dusver verkoop is. KTable sal jou inligting gee oor watter aandele tans die meeste verkoop, deurlopende samevoeging van elke opdatering is nie nodig nie.

Ons het geleer om twee belangrike dinge te doen:

• groepeer waardes in KTable volgens hul gemeenskaplike sleutel;
• выполнять над этими сгруппированными значениями такие полезные операции, как свертка и агрегирование.

Умение выполнять эти операции важно для понимания смысла данных, движущихся через приложение Kafka Streams, и выяснения того, какую информацию они несут.

Ons het ook van die sleutelbegrippe wat vroeër in hierdie boek bespreek is, bymekaar gebring. In Hoofstuk 4 het ons die belangrikheid van foutverdraagsame, plaaslike staat in 'n stroomtoepassing bespreek. Die eerste voorbeeld in hierdie hoofstuk het gewys hoekom plaaslike staat so belangrik is—dit gee jou die vermoë om tred te hou met watter inligting jy reeds gesien het. Plaaslike toegang vermy netwerkvertragings, wat die toepassing meer doeltreffend en foutverdraagsaam maak.

Wanneer u enige oprol- of samevoegingsbewerking uitvoer, moet u die naam van die staatswinkel spesifiseer. Die vou- en totale bewerkings gee 'n KTable-instansie terug, en die KTable gebruik 'n staatswinkel om ou resultate met nuwes te vervang. Soos u gesien het, word nie alle opdaterings verder in die pyplyn gestuur nie, en dit is belangrik, aangesien samevoegingsbewerkings ontwerp is om opsommende inligting te verkry. As jy nie die plaaslike staat toepas nie, sal KTable al die resultate van samevoeging en vermindering verder stuur.

Далее мы посмотрим на выполнение таких операций, как агрегирование, в пределах конкретного промежутка времени — так называемых оконных операций (windowing operations).

5.3.2. Vensterbedrywighede

В предыдущем разделе мы познакомились со «скользящими» сверткой и агрегированием. Приложение производило непрерывную свертку объема продаж акций с последующим агрегированием пяти наиболее продаваемых на бирже акций.

Soms is sulke voortdurende samevoeging en konvolusie van resultate nodig. En soms moet u slegs operasies oor 'n gegewe tydperk uitvoer. Byvoorbeeld, om te bereken hoeveel ruiltransaksies in die laaste 10 minute met die aandele van 'n spesifieke maatskappy gemaak is. Of hoeveel gebruikers die afgelope 15 minute op 'n nuwe banieradvertensie geklik het. 'n Toepassing kan sulke bewerkings herhaaldelik uitvoer, maar met resultate wat slegs verband hou met gespesifiseerde tydintervalle (tydvensters).

Tel ruiltransaksies deur koper

В следующем примере мы займемся отслеживанием биржевых транзакций по нескольким трейдерам — либо крупным организациям, либо смышленым финансистам-одиночкам.

Daar is twee moontlike redes vir hierdie dop. Een daarvan is die behoefte om te weet wat die markleiers koop/verkoop. As hierdie groot spelers en gesofistikeerde beleggers 'n geleentheid vir hulself sien, maak dit sin om hul strategie te volg. Die tweede rede is die begeerte om enige moontlike tekens van onwettige transaksies met behulp van binne-inligting raak te sien. Om dit te doen, sal jy die korrelasie van groot verkoopspieke met belangrike persvrystellings moet ontleed.

Hierdie opsporing bestaan ​​uit die volgende stappe:

• skep 'n draad vir lees uit die voorraadtransaksie-onderwerp;
• группировка входящих записей по идентификатору покупателя и биржевому символу акции. Вызов метода groupBy возвращает экземпляр класса KGroupedStream;
• возвращение методом KGroupedStream.windowedBy потока данных, ограниченного временным окном, что позволяет выполнять оконное агрегирование. В зависимости от типа окна возвращается либо TimeWindowedKStream, либо SessionWindowedKStream;
• подсчет транзакций для операции агрегирования. Оконный поток данных определяет, учитывается ли при этом подсчете конкретная запись;
• запись результатов в топик или вывод их в консоль во время разработки.

Топология данного приложения проста, но наглядная ее картинка не помешает. Взглянем на рис. 5.11.

Далее мы рассмотрим функциональность оконных операций и соответствующий код.

Tipes vensters

Daar is drie tipes vensters in Kafka Streams:

• sessie;
• "tuimel" (tuimel);
• gly / "spring" (gly / huppel).

Какое выбрать — зависит от бизнес-требований. «Кувыркающиеся» и «прыгающие» окна ограничиваются по времени, в то время как ограничения сеансовых связаны с действиями пользователей — длительность сеанса (-ов) определяется исключительно тем, насколько активно ведет себя пользователь. Главное — не забывать, что все типы окон основываются на метках даты/времени записей, а не на системном времени.

Далее мы реализуем нашу топологию с каждым из типов окон. Полный код будет приведен только в первом примере, для других типов окон ничего не изменится, кроме типа оконной операции.

Sessie vensters

Сеансовые окна сильно отличаются от всех остальных типов окон. Они ограничиваются не столько по времени, сколько активностью пользователя (или активностью той сущности, которую вы хотели бы отслеживать). Сеансовые окна разграничиваются периодами бездействия.

Рисунок 5.12 иллюстрирует понятие сеансовых окон. Меньший сеанс будет сливаться с сеансом слева от него. А сеанс справа будет отдельным, поскольку следует за длительным периодом бездействия. Сеансовые окна основываются на действиях пользователей, но применяют метки даты/времени из записей для определения того, к какому сеансу относится запись.

Gebruik sessievensters om ruiltransaksies op te spoor

Воспользуемся сеансовыми окнами для захвата информации о биржевых транзакциях. Реализация сеансовых окон показана в листинге 5.5 (который можно найти в файле src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).

Большинство операций этой топологии вы уже встречали, так что нет нужды рассматривать их тут снова. Но есть здесь и несколько новых элементов, которые мы сейчас обсудим.

При всякой операции groupBy обычно выполняется какая-либо операция агрегирования (агрегирование, свертка или подсчет количества). Можно выполнить или накопительное агрегирование с нарастающим итогом, или оконное агрегирование, при котором учитываются записи в пределах заданного временного окна.

Код из листинга 5.5 выполняет подсчет количества транзакций в пределах сеансовых окон. На рис. 5.13 эти действия анализируются пошагово.

С помощью вызова windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) мы создаем сеансовое окно с интервалом бездействия 20 секунд и интервалом сохранения 15 минут. Интервал бездействия 20 секунд означает, что приложение будет включать любую запись, которая поступит в пределах 20 секунд от окончания или начала текущего сеанса в текущий (активный) сеанс.

Vervolgens spesifiseer ons watter samevoegingsbewerking om in die sessievenster uit te voer - tel in hierdie geval. As die inkomende skrywe verder gaan as die ledige interval (aan weerskante van die datum/tydstempel), dan skep die toepassing 'n nuwe sessie. 'n Volhardingsinterval beteken om 'n sessie vir 'n sekere tyd lewendig te hou en laat data toe wat verder gaan as die sessie se ledige tydperk, maar steeds aangeheg kan word. Daarbenewens stem die begin en einde van die nuwe sessie as gevolg van die samesmelting ooreen met die vroegste en jongste datum-/tydstempels.

Рассмотрим несколько записей из метода count, чтобы увидеть, как работают сеансы (табл. 5.1).

При поступлении записей мы ищем уже существующие сеансы с тем же ключом, временем окончания меньше чем текущая метка даты/времени — интервал бездействия и временем начала больше чем текущая метка даты/времени + интервал бездействия. С учетом этого четыре записи из табл. 5.1 сливаются в единый сеанс следующим образом.

1. Первой поступает запись 1, так что время начала равно времени окончания и равно 00:00:00.

2. Далее поступает запись 2, и мы ищем сеансы, заканчивающиеся не раньше 23:59:55 и начинающиеся не позднее 00:00:35. Находим запись 1 и объединяем сеансы 1 и 2. Берем время начала сеанса 1 (более раннее) и время окончания сеанса 2 (более позднее), так что наш новый сеанс начинается в 00:00:00 и заканчивается в 00:00:15.

3. Поступает запись 3, мы ищем сеансы между 00:00:30 и 00:01:10 и не находим ни одного. Добавляем второй сеанс для ключа 123-345-654,FFBE, начинающийся и заканчивающийся в 00:00:50.

4. Поступает запись 4, и мы ищем сеансы между 23:59:45 и 00:00:25. На этот раз находятся оба сеанса — 1 и 2. Все три сеанса объединяются в один, с временем начала 00:00:00 и временем окончания 00:00:15.

Hier is 'n paar belangrike dinge om in gedagte te hou uit hierdie afdeling:

• сеансы — не окна фиксированного размера. Длительность сеанса определяется активностью в рамках заданного промежутка времени;
• метки даты/времени в данных определяют, попадает событие в существующий сеанс или в промежуток бездействия.

Vervolgens sal ons die volgende tipe venster bespreek - "tuimel" vensters.

"Tuimelende" vensters

«Кувыркающиеся» (tumbling) окна захватывают события, попадающие в определенный промежуток времени. Представьте себе, что вам нужно захватывать все биржевые транзакции какой-то компании каждые 20 секунд, так что вы собираете все события за этот промежуток времени. По окончании 20-секундного интервала окно «кувыркается» и переходит на новый 20-секундный интервал наблюдения. Рисунок 5.14 иллюстрирует эту ситуацию.

Как вы можете видеть, все поступившие за последние 20 секунд события включены в окно. По окончании этого промежутка времени создается новое окно.

В листинге 5.6 приведен код, демонстрирующий использование «кувыркающихся» окон для захвата каждые 20 секунд биржевых транзакций (его можно найти в файле src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Благодаря этому небольшому изменению вызова метода TimeWindows.of можно использовать «кувыркающееся» окно. В данном примере нет вызова метода until(), вследствие чего будет использоваться интервал сохранения по умолчанию, равный 24 часам.

Наконец, пора перейти к последнему из вариантов окон — «прыгающим» (hopping) окнам.

Glyende ("springende") vensters

Скользящие/«прыгающие» (sliding/hopping) окна похожи на «кувыркающиеся», но с небольшим отличием. Скользящие окна не ждут окончания интервала времени перед созданием нового окна для обработки недавних событий. Они запускают новые вычисления после интервала ожидания, меньшего чем длительность окна.

Om die verskille tussen "tuimel" en "spring" vensters te illustreer, kom ons gaan terug na die voorbeeld van die tel van voorraadtransaksies. Ons doelwit is steeds om die aantal transaksies te tel, maar ons wil nie heeltyd wag voordat ons die teller opdateer nie. In plaas daarvan sal ons die teller met korter tussenposes opdateer. Ons sal byvoorbeeld steeds die aantal transaksies elke 20 sekondes tel, maar die teller elke 5 sekondes opdateer, soos in Fig. 5.15. In hierdie geval het ons drie resultaatvensters met oorvleuelende data.

В листинге 5.7 приведен код для задания скользящих окон (его можно найти в файле src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

«Кувыркающееся» окно можно преобразовать в «прыгающее» с помощью добавления вызова метода advanceBy(). В приведенном примере интервал сохранения равен 15 минутам.

Вы увидели в этом разделе, как ограничивать результаты агрегирования временными окнами. В частности, хотелось бы, чтобы вы запомнили из этого раздела следующие три вещи:

• размер сеансовых окон ограничивается не промежутком времени, а активностью пользователей;
• "Tumble" vensters gee 'n idee van gebeure binne 'n gegewe tydperk;
• длительность работы «прыгающих» окон фиксирована, но они часто обновляются и могут содержать во всех окнах пересекающиеся записи.

Далее мы узнаем, как преобразовать KTable обратно в KStream для соединения.

5.3.3. Verbind KStream en KTable-voorwerpe

В главе 4 мы обсуждали соединение двух объектов KStream. Теперь нам предстоит научиться соединять KTable и KStream. Понадобиться это может по следующей простой причине. KStream — поток записей, а KTable — поток обновлений записей, но иногда может быть нужно добавить дополнительный контекст к потоку записей с помощью обновлений из KTable.

Возьмем данные о количестве биржевых транзакций и соединим их с биржевыми новостями по соответствующим отраслям промышленности. Вот что нужно сделать, что добиться этого с учетом уже имеющегося кода.

1. Преобразовать объект KTable с данными о количестве биржевых транзакций в KStream с последующей заменой ключа на ключ, обозначающий отрасль промышленности, соответствующую данному символу акций.
2. Создать объект KTable, читающий данные из топика с биржевыми новостями. Этот новый KTable будет категоризован по отраслям промышленности.
3. Соединить обновления новостей с информацией о количестве биржевых транзакций по отраслям промышленности.

Kom ons kyk nou hoe om hierdie aksieplan te implementeer.

Skakel KTable om na KStream

Om KTable na KStream om te skakel, moet jy die volgende doen.

1. Roep die KTable.toStream() metode.
2. С помощью вызова метода KStream.map заменить ключ названием отрасли промышленности, после чего извлечь из экземпляра Windowed объект TransactionSummary.

Мы свяжем эти операции цепочкой следующим образом (код можно найти в файле src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (листинг 5.8).

Поскольку мы выполняем операцию KStream.map, то повторное секционирование для возвращаемого экземпляра KStream производится автоматически при его использовании в соединении.

Мы завершили процесс преобразования, далее нам нужно создать объект KTable для чтения биржевых новостей.

Skep 'n KTable vir voorraadnuus

К счастью, для создания объекта KTable достаточно одной строки кода (этот код можно найти в файле src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (листинг 5.9).

Стоит отметить, что никаких объектов Serde указывать не требуется, поскольку в настройках используются строковые Serde. Также благодаря применению перечисления EARLIEST таблица заполняется записями в самом начале.

Nou kan ons aanbeweeg na die laaste stap – die verbinding.

Koppel nuusopdaterings met transaksietellingdata

Создание соединения не представляет сложностей. Мы воспользуемся левым соединением на случай, если по соответствующей отрасли промышленности нет биржевых новостей (нужный код можно найти в файле src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (листинг 5.10).

Hierdie leftJoin-operateur is redelik eenvoudig. Anders as die koppelings in Hoofstuk 4, word die JoinWindow-metode nie gebruik nie, want wanneer 'n KStream-KTable-koppeling gemaak word, is daar net een inskrywing in die KTable vir elke sleutel. So 'n verband is nie in tyd beperk nie: die inskrywing bestaan ​​óf in die KTable óf dit bestaan ​​nie. Sleutel wegneemetes: Deur KTable-voorwerpe te gebruik, kan jy 'n KStream verryk met minder gereeld bygewerkte verwysingsdata.

А теперь мы рассмотрим более эффективный способ обогащения событий из KStream.

5.3.4. GlobalKTable voorwerpe

Soos u verstaan, is daar 'n behoefte om die gebeurtenisstrome te verryk of konteks daarby te voeg. In Hoofstuk 4 het jy die verbinding van twee KStream-voorwerpe gesien, en in die vorige afdeling het jy die verbinding van KStream en KTable gesien. In al hierdie gevalle is dit nodig om die datastroom te herpartisioneer wanneer sleutels na 'n nuwe tipe of waarde gekarteer word. Soms word herpartisionering eksplisiet gedoen, en soms doen Kafka Streams dit outomaties. Die herpartisionering is nodig omdat die sleutels verander het en die rekords in die nuwe partisies moet beland of die koppeling sal nie moontlik wees nie (dit is bespreek in Hoofstuk 4, "Data Herpartisionering" in Afdeling 4.2.4).

Herpartisionering het 'n prys

Herpartisionering kom teen 'n koste - bykomende hulpbronkoste vir die skep van intermediêre onderwerpe, die stoor van duplikaatdata in 'n ander onderwerp; dit beteken ook verhoogde latensie as gevolg van skryf na en lees van daardie onderwerp. Ook, as jy op meer as een aspek of dimensie wil aansluit, moet jy kettingverbinding maak, rekords met nuwe sleutels karteer en die herverdelingsproses weer laat loop.

Koppel aan kleiner datastelle

В некоторых случаях объем справочных данных, с которыми планируется соединение, относительно невелик, так что полные их копии вполне могут поместиться локально на каждом из узлов. Для подобных ситуаций в Kafka Streams предусмотрен класс GlobalKTable.

GlobalKTable-gevalle is uniek omdat die toepassing alle data na elk van die nodusse repliseer. En aangesien elkeen van die nodusse al die data het, is dit nie nodig om die gebeurtenisstroom deur die verwysingsdatasleutel te partisieer sodat dit vir alle partisies beskikbaar is nie. U kan ook sleutellose verbindings uitvoer deur GlobalKTable-voorwerpe te gebruik. Kom ons gaan terug na een van die vorige voorbeelde om hierdie moontlikheid te demonstreer.

Koppel KStream-voorwerpe aan GlobalKTable-voorwerpe

В подразделе 5.3.2 мы выполнили оконное агрегирование биржевых транзакций по покупателям. Результаты этого агрегирования выглядели примерно следующим образом:

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

Хотя эти результаты соответствовали поставленной цели, было бы удобнее, если бы выводилось также имя клиента и полное название компании. Чтобы добавить имя покупателя и название компании, можно выполнять обычные соединения, но при этом понадобится произвести два отображения ключей и повторное секционирование. С помощью GlobalKTable можно избежать затрат на подобные операции.

Для этого мы воспользуемся объектом countStream из листинга 5.11 (соответствующий код можно найти в файле src/main/java/bbejeck/chapter_5/GlobalKTableExample.java), соединив его с двумя объектами GlobalKTable.

Мы уже обсуждали это ранее, так что не стану повторяться. Но отмечу, что код в функции toStream().map ради удобочитаемости абстрагирован в объект-функцию вместо встраиваемого лямбда-выражения.

Следующий этап — объявление двух экземпляров GlobalKTable (приведенный код можно найти в файле src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (листинг 5.12).

Обратите внимание, что названия топиков описываются с помощью перечисляемых типов.

Теперь, когда мы подготовили все компоненты, осталось написать код для соединения (который можно найти в файле src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (листинг 5.13).

Хотя в этом коде присутствуют два соединения, они организованы в виде цепочки, поскольку отдельно ни один из их результатов не используется. Результаты выводятся в конце всей операции.

При запуске вышеприведенной операции соединения вы получите результаты следующего вида:

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

Die essensie het nie verander nie, maar hierdie resultate lyk meer verstaanbaar.

Если считать главу 4, вы уже видели несколько типов соединений в действии. Они перечислены в табл. 5.2. Эта таблица отражает возможности соединения, актуальные для версии 1.0.0 Kafka Streams; в будущих выпусках, возможно, что-то поменяется.

Ten slotte, laat ek jou herinner aan die belangrikste ding: jy kan gebeurtenisstrome (KStream) en strome (KTable) bywerk deur die plaaslike staat te koppel. Daarbenewens, as die grootte van die verwysingsdata nie te groot is nie, kan jy die GlobalKTable-objek gebruik. GlobalKTable repliseer alle partisies na elk van die nodusse van die Kafka Streams-toepassing, om sodoende te verseker dat alle data beskikbaar is, ongeag met watter partisie die sleutel ooreenstem.

Далее мы увидим возможность Kafka Streams, благодаря которой можно наблюдать изменения состояния без потребления данных из топика Kafka.

5.3.5. Versoekbare staat

Мы уже выполняли несколько операций с участием состояния и всегда выводили результаты в консоль (для целей разработки) или записывали их в топик (для целей промышленной эксплуатации). При записи результатов в топик приходится использовать потребитель Kafka для их просмотра.

Die lees van data uit hierdie onderwerpe kan as 'n soort gematerialiseerde sienings beskou word. Vir ons take kan ons die definisie van 'n gematerialiseerde siening van Wikipedia gebruik: "... 'n fisiese databasisobjek wat die resultate van 'n navraag bevat. Dit kan byvoorbeeld 'n plaaslike kopie van afgeleë data wees, of 'n subset van die rye en/of kolomme van 'n tabel of aansluitingsresultaat, of 'n spiltabel wat deur samevoeging verkry is” (https://en.wikipedia.org/wiki) /Materialized_view).

Kafka Streams также позволяет выполнять интерактивные запросы (interactive queries) к хранилищам состояния, что дает возможность непосредственного чтения этих материализованных представлений. Важно отметить, что запрос к хранилищу состояния носит характер операции «только для чтения». Благодаря этому вы можете не бояться случайно сделать состояние несогласованным во время обработки данных приложением.

Возможность непосредственных запросов к хранилищам состояния имеет большое значение. Она значит, что можно создавать приложения — информационные панели без необходимости сначала получать данные от потребителя Kafka. Повышает она и эффективность приложения, благодаря тому что не требуется снова записывать данные:

• as gevolg van die ligging van die data, kan hulle vinnig verkry word;
• исключается дублирование данных, поскольку они не записываются во внешнее хранилище.

Die belangrikste ding wat ek wil hê jy moet onthou, is dat jy die staat direk vanaf die aansoek kan navraag doen. Die moontlikhede wat dit vir jou bied, kan nie oorskat word nie. In plaas daarvan om data van Kafka af te verbruik en rekords in die databasis vir die toepassing te stoor, kan u die staatwinkels navraag doen met dieselfde resultaat. Om direk na staatwinkels te vra, beteken minder kode (geen verbruiker) en minder sagteware (geen behoefte aan 'n databasistabel om resultate te stoor nie).

Ons het baie inligting in hierdie hoofstuk behandel, so ons sal ons bespreking van interaktiewe navrae oor staatswinkels vir 'n oomblik stop. Maar moenie bekommerd wees nie: in Hoofstuk 9 sal ons 'n eenvoudige dashboard-toepassing met interaktiewe navrae skep. Dit sal sommige van die voorbeelde in hierdie en vorige hoofstukke gebruik om interaktiewe navrae te demonstreer en hoe om dit by Kafka Streams-toepassings te voeg.

Opsomming

• Объекты KStream олицетворяют потоки событий, сравнимые со вставками в базу данных. Объекты KTable олицетворяют потоки обновлений, они больше схожи с обновлениями в базе данных. Размер объекта KTable не растет, старые записи заменяются новыми.
• KTable-objekte word benodig vir samevoegingsbewerkings.
• Venster laat jou toe om saamgestelde data in tydbakke af te breek.
• Благодаря объектам GlobalKTable можно получить доступ к справочным данным в любой точке приложения, независимо от разбиения по секциям.
• Verbindings tussen KStream, KTable en GlobalKTable voorwerpe is moontlik.

Tot dusver het ons daarop gefokus om Kafka Streams-toepassings te bou deur die hoëvlak DSL KStream te gebruik. Alhoewel die hoëvlakbenadering jou toelaat om netjiese en bondige programme te skep, verteenwoordig die gebruik daarvan 'n sekere kompromie. Werk met DSL KStream beteken om kode meer bondig te maak ten koste van minder beheer. In die volgende hoofstuk kyk ons ​​na die lae-vlak hanteerder node API en probeer ander afwegings. Programme sal langer word as wat dit tot dusver was, maar ons sal byna enige hanteerdernodus kan skep wat ons nodig mag hê.

→ Meer besonderhede oor die boek kan gevind word by uitgewer se webwerf

→ Vir Khabrozhiteli, 'n 25% afslag op die koepon - Kafka-strome

→ По факту оплаты бумажной версии книги на e-mail высылается электронная книга.

Bron: will.com