Книга "Kafka Streams в действие. Приложения и микросервизи за работа в реално време"

Здравейте, Хаброжители! Тази книга ще бъде полезна за всеки разработчик, който иска да се запознае с потоковата обработка. Разбирането на разпределеното програмиране ще помогне за по-добро изучаване на Kafka и Kafka Streams. Беше полезно да знаете и самата рамка Kafka, но това не е задължително: аз ще ви разкажа всичко, което е нужно. Опитни разработчици на Kafka, точно както и новаците, ще усвоят създаването на интересни приложения за потокова обработка с помощта на библиотеката Kafka Streams, благодарение на тази книга. Java разработчиците с междинно и напреднало ниво, вече свикнали с понятието сериализация, ще научат как да прилагат уменията си за създаване на приложения Kafka Streams. Изходният код на книгата е написан на Java 8 и значително използва синтаксиса на лямбда-изрази на Java 8, така че умението да работите с лямбда-функции (дори и на друг език за програмиране) ще ви бъде от полза.

Откъс. 5.3. Агрегиране и оконни операции

В този раздел ще преминем към изучаване на най-обещаващите части на Kafka Streams. Досега разгледахме следните аспекти на Kafka Streams:

• създаване на топология за обработка;
• използване на състояние в потокови приложения;
• изпълнение на съединения на потоци данни;
• разлики между потоците събития (KStream) и потоците актуализации (KTable).

В следващите примери ще съберем всички тези елементи заедно. Освен това ще се запознаете с оконните операции — още една забележителна възможност за потокови приложения. Първият ни пример ще бъде просто агрегиране.

5.3.1. Агрегиране на обема на продажбите на акции по сектори на индустрията

Агрегирането и групирането са жизненоважни инструменти при работа с потокови данни. Изследването на индивидуални записки в момента на получаване често се оказва недостатъчно. За извличане на допълнителна информация от данните е необходимо тяхното групиране и комбиниране.

В този пример ще влезете в ролята на дневен трейдър, който трябва да следи обемите на продажба на акции на компании в няколко индустриални сектора. По-специално, вас интересуват петте компании с най-големи обеми на продажба на акции в всеки от индустриалните сектори.

За подобно агрегиране ще са нужни няколко последващи стъпки за преобразуване на данните в необходимия формат (ако говорим в общи линии).

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

Какво е интерфейсът KGroupedStream

Методите KStream.groupBy и KStream.groupByKey връщат екземпляр KGroupedStream. KGroupedStream е междинно представяне на потока от събития след групиране по ключове. Той не е предназначен за пряка работа. Вместо това, KGroupedStream се използва за операции по агрегация, чийто резултат винаги е KTable. И тъй като резултатът от операциите по агрегация е KTable и в тях се прилага хранилище на състоянието, е възможно не всички актуализации в резултата да се изпращат по-нататък по конвейера.

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

Нека направим кратка пауза и да погледнем Рис. 5.9, който показва какво сме постигнали. Тази топология би трябвало да ви е добре известна.

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

Представеният код е кратък и извършва много действия с няколко реда. В първия параметър на метода builder.stream можете да забележите нещо ново за вас: стойността на изброимия тип AutoOffsetReset.EARLIEST (съществува и LATEST), зададена с метода Consumed.withOffsetResetPolicy. С помощта на този изброим тип можете да зададете стратегия за нулиране на офсетите за всеки от KStream или KTable, която има приоритет над параметъра за нулиране на офсетите от конфигурацията.

GroupByKey и GroupBy

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

Методът GroupByKey се прилага, когато ключовете в KStream вече са ненулеви. И най-важното, флагът "изисква повторно секциониране" никога не е бил зададен.

Методът GroupBy предполага, че сте променяли ключовете за групиране, така че флагът за повторно секциониране е зададен на true. Изпълнението след метода GroupBy на свързването, агрегирането и т.н. ще доведе до автоматично повторно секциониране.
Резюме: винаги, когато е възможно, трябва да се използва GroupByKey, а не GroupBy.

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

Методът ShareVolume.sum връща междинната сума на обема на продажбите на акции, а резултатът от цялата верига от изчисления представлява KTable. Сега разбирате каква роля играе KTable. При постъпването на обекти ShareVolume, в съответния KTable се запазва последното актуално обновление. Важно е да не забравяте, че всички актуализации се отразяват в предшестващия shareVolumeKTable, но не всички се изпращат нататък.

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

1. Изпълнете още една операция groupBy за групиране на отделните обекти ShareVolume по индустрии.
2. Приступете към сумиране на обектите ShareVolume. Този път обектът за агрегация е опашка с фиксиран приоритет. В такава опашка с фиксиран размер се запазват само петте компании с най-големи количества продадени акции.
3. Отобразете опашките от предишната точка в стринг стойност и върнете петте най-продавани компании по количество акции по индустрии.
4. Запишете резултатите в стринг формат в темата.

На рис. 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(), така че ще отделим малко време за обсъждането му.

Както помните, KTable се отличава с това, че записите с еднакви ключове се считат за актуализации. KTable замества старата запись с нова. Агрегирането става по подобен начин: агрегират се последните записи с един ключ. При постъпването на запис, той се добавя в инстанцията на класа FixedSizePriorityQueue с помощта на сумиращия функция (вторият параметър в извикването на метода aggregate), но ако вече съществува друга запись с този ключ, старата запись се изтрива с помощта на изваждащия (третият параметър в извикването на метода aggregate).

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

Научихме се да правим две важни неща:

• групираме стойностите в KTable по общия ключ;
• изпълняваме полезни операции като свиване и агрегиранe върху тези групирани стойности.

Умението да изпълняваме тези операции е важно за разбирането на смисъла на данните, преминаващи през приложението Kafka Streams, и за установяване на информацията, която те носят.

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

При изпълнение на всяка операция на свиване или агрегация е необходимо да посочите името на хранилището на състоянието. Операциите за свиване и агрегация връщат инстанция на KTable, а KTable използва хранилището на състоянието, за да замени старите резултати с нови. Както видяхте, не всички актуализации се изпращат нататък по конвейера, което е важно, тъй като операциите за агрегация са предназначени да получат крайна информация. Ако не се приложи локалното състояние, KTable ще изпрати напред всички резултати от агрегацията и свиването.

Следващото, което ще разгледаме, е изпълнението на операции, като агрегация, в рамките на определен период от време — така наречените оконни операции.

5.3.2. Оконни операции

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

Понякога подобни непрекъснати агрегации и свивания на резултати са необходими. А понякога е нужно да се извършат операции само за определен период от време. Например, да се изчисли колко борсови операции с акциите на конкретна компания са извършени през последните 10 минути. Или колко потребители са кликнали на новия рекламeн банер през последните 15 минути. Приложението може да извършва такива операции многократно, но с резултати, отнасящи се само за зададените времеви периоди (времеви прозорци).

Броене на борсовите транзакции по купувач

В следващия пример ще се занимаем с проследяване на борсовите транзакции по няколко търговци — или големи организации, или умни финансисти-индивиди.

Съществуват две възможни причини за подобно проследяване. Едната от тях е необходимостта да знаем какво купуват/продават лидерите на пазара. Ако тези големи играчи и опитни инвеститори виждат пред себе си възможности, има смисъл да следвате техните стратегии. Втората причина е желанието да се забележат всякакви възможни признаци на незаконни сделки, използващи вътрешна информация. За това ще ви е необходимо да анализирате корелацията на големите скокове в продажбите с важни прессъобщения.

Такова проследяване се състои от следните етапи:

• създаване на поток за четене от темата stock-transactions;
• групиране на входящите записи по идентификатор на купувача и борсов символ на акцията. Извикването на метода groupBy връща екземпляр от класа KGroupedStream;
• върнатият от метода KGroupedStream.windowedBy поток от данни, ограничен с времево прозорец, позволява да се извърши оконно агрегиране. В зависимост от типа на прозореца се връща или TimeWindowedKStream, или SessionWindowedKStream;
• броене на транзакциите за операцията по агрегиране. Оконният поток данни определя дали конкретен запис е взет предвид в броенето;
• записване на резултатите в тема или извеждането им в конзолата по време на разработка.

Топологията на това приложение е проста, но визуализацията ѝ няма да навреди. Нека погледнем на рис. 5.11.

След това ще разгледаме функционалността на оконните операции и съответния код.

Типове прозорци

В Kafka Streams има три типа прозорци:

• сеансови;
• „кувыркающиеся“ (tumbling);
• сколзящи/„прыгающие“ (sliding/hopping).

Кое да изберете — зависи от бизнес изискванията. „Кувыркающиеся“ и „прыгающие“ прозорци са ограничени по време, докато ограниченията на сеансовите са свързани с действията на потребителите — продължителността на сесията (-ите) се определя изцяло от това, колко активно се държи потребителят. Най-важното е да не забравяме, че всички типове прозорци се базират на времевите марки на записите, а не на системното време.

След това ще реализираме нашата топология с всеки от типовете прозорци. Целият код ще бъде представен само в първия пример; за другите типове прозорци няма да се променя нищо, освен типа на операциите с прозорци.

Сеансови прозорци

Сеансовите прозорци значително се различават от всички останали видове прозорци. Те са ограничени не толкова по време, а от активността на потребителя (или активността на същността, която искате да следите). Сеансовите прозорци се разграничават по периоди на бездействие.

Фигура 5.12 илюстрира понятието за сеансови прозорци. По-малкият сеанс ще се слее с сеанса отляво. А сеансът отдясно ще бъде отделен, тъй като следва след дълъг период на бездействие. Сеансовите прозорци са основани на действията на потребителите, но използват времеви/дата маркери от записите, за да определят с кой сеанс е свързан записа.

Използване на сеансови прозорци за проследяване на борсови транзакции

Ще използваме сеансовите прозорци, за да уловим информация за борсови транзакции. Реализацията на сеансовите прозорци е показана в листинга 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 секунди след завършването или началото на текущия сеанс в текущия (активен) сеанс.

След това посочваме каква операция на агрегация трябва да се извърши в сесийния прозорец — в този случай count. Ако входящият запис излезе извън интервала на бездействие (от която и да е страна на времевата марка), приложението създава нова сесия. Интервалът на запазване означава поддържане на сесията за определено време и позволява закъснели данни, които излизат извън периода на бездействие на сесията, но все още могат да бъдат присъединени. Освен това, началото и краят на новата сесия, получена в резултат на сливания, съответстват на най-ранната и най-късната времева марка.

Да разгледаме няколко записа от метода 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.

От разказаното в този раздел, следва да запомним следните важни нюанси:

• сесиите не са прозорци с фиксиран размер. Продължителността на сесията се определя от активността в рамките на зададения времеви интервал;
• временните маркери в данните определят дали събитието попада в съществуваща сесия или в период на бездействие.

След това ще обсъдим следващия вид прозорци — „превъртащи се“ прозорци.

„Превъртащи се“ прозорци

«Тумблиращи» (кувиращи) прозорци улавят събития, попадащи в определен интервал от време. Представете си, че трябва да улавяте всички борсови транзакции на дадена компания на всеки 20 секунди, така че събирате всички събития за този период. В края на 20-секундния интервал прозорецът «се тумблира» и преминава към нов 20-секунден интервал на наблюдение. Рисунок 5.14 илюстрира тази ситуация.

Както можете да видите, всички събития, получени за последните 20 секунди, са включени в прозореца. В края на този период се създава нов прозорец.

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

Благодарение на тази малка промяна в извикването на метода TimeWindows.of, може да се използва «тумблиращ» прозорец. В този пример няма извикване на метода until(), което означава, че ще се използва интервалът за запазване по подразбиране, равен на 24 часа.

Накрая е време да преминем към последния вариант на прозорци — «скачащи» (hopping) прозорци.

Скользящи («скачащи») прозорци

Скользящите/«скачащите» (sliding/hopping) прозорци са подобни на «тумблиращите», но с малка разлика. Скользящите прозорци не изчакват приключването на времевия интервал, преди да създадат нов прозорец за обработка на недавни събития. Те стартират нови изчисления след интервал на изчакване, по-кратък от продължителността на прозореца.

За да илюстрираме разликите между «тумблиращите» и «скачащите» прозорци, ще се върнем към примера с броенето на борсови транзакции. Нашата цел все още е да преброим броя на транзакциите, но не искаме да чакаме целия период, преди да обновим брояча. Вместо това ще обновяваме брояча на по-кратки интервали. Например, ще продължим да броим транзакциите на всеки 20 секунди, но ще обновяваме брояча на всеки 5 секунди, както е показано на рис. 5.15. В резултат имаме три прозореца с резултати с припокриващи се данни.

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

Огледалното" прозорец може да бъде преобразувано в "скачащо" чрез добавяне на извикване на метода advanceBy(). В дадения пример интервалът за запазване е 15 минути.

В този раздел видяхте как да ограничите резултатите от агрегирането с времеви прозорци. По-специално, бих искал да запомните следните три неща от този раздел:

• размерът на сесийните прозорци не се ограничава от времеви интервали, а от активността на потребителите;
• "огледалните" прозорци предоставят информация за събития в зададен период от време;
• дължината на "скачащите" прозорци е фиксирана, но те се актуализират често и могат да съдържат пресичащи се записи в различните прозорци.

След това ще научим как да преобразуваме KTable обратно в KStream за свързване.

5.3.3. Свързване на обекти KStream и KTable

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

Нека вземем данни за броя на борсовите транзакции и да ги свържем с борсови новини по съответните индустрии. Ето какво трябва да направите, за да постигнете това с вече наличния код.

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

Сега нека видим как да реализираме този план за действие.

Преобразуване на KTable в KStream

За да преобразувате KTable в KStream, трябва да направите следното.

1. Извикайте метода KTable.toStream().
2. С помощта на извикване на метода KStream.map заменете ключа с името на индустрията, след което извлечете обекта TransactionSummary от екземпляра Windowed.

Ние ще свържем тези операции в следната верига (кодът може да бъде намерен в файла src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (листинг 5.8).

Тъй като изпълняваме операция KStream.map, повторното секциониране на върнатия экземпляр KStream се извършва автоматично, когато се използва в съединение.

Завършихме процеса на трансформация, сега трябва да създадем обект KTable за четене на борсови новини.

Създаване на KTable за борсови новини

За щастие, за създаването на обект KTable е необходима само една ред код (този код може да бъде намерен в файла src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (листинг 5.9).

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

Сега можем да преминем към последната стъпка – съединението.

Съединение на актуализации на новините с данни за броя на транзакциите

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

Тази операция leftJoin е достатъчно проста. В отличие от съединенията в глава 4, методът JoinWindow не се използва, тъй като при изпълнението на съединението KStream-KTable за всеки ключ в KTable присъства само едно запис. Такова съединение не е ограничено във времето: записът или е в KTable, или го няма. Основният извод: с помощта на обекти KTable можем да обогатим KStream с по-рядко актуализирани справочни данни.

А сега ще разгледаме по-ефективен начин за обогатяване на събитията от KStream.

5.3.4. Обекти GlobalKTable

Както разбирате, има необходимост от обогатяване на потоците от събития или добавяне на контекст към тях. В глава 4 видяхте съединения на два обекта KStream, а в предишния раздел – съединение между KStream и KTable. Във всички тези случаи е необходимо повторно секциониране на потока от данни, когато ключовете се отразяват на нов тип или стойност. Понякога повторното секциониране се извършва изрично, а понякога Kafka Streams го прави автоматично. Повторното секциониране е задължително, тъй като ключовете са се променили и записите трябва да попаднат в новите секции, иначе съединението ще бъде невъзможно (това беше обсъдено в глава 4, в точката „Повторно секциониране на данните“ от подглава 4.2.4).

Повторното секциониране има своя цена

Повторното секциониране изисква разходи — допълнителни разходи за ресурси за създаване на междинни теми, запазване на дублирани данни в още една тема; то също така означава увеличение на закъснението поради запис и четене от тази тема. Освен това, ако е необходимо да се извърши свързване по повече от един аспект или измерение, трябва да се организират връзките в поредица, да се отобразят записите с нови ключове и отново да се проведе процесът на повторно секциониране.

Свързване с по-малки набори от данни

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

Екземплярите на GlobalKTable са уникални, тъй като приложението репликира всички данни на всеки от възлите. И тъй като на всеки от възлите присъстват всички данни, няма нужда да се секционира потокът от събития по ключа на справочните данни, за да бъде достъпен за всички секции. Чрез обектите GlobalKTable могат да се извършват и безключови свързвания. Нека се върнем към един от предишните примери, за да демонстрираме тази възможност.

Свързване на обекти KStream с обекти GlobalKTable

В подраздел 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 е абстрахиран в обект-функция вместо вградена lambda-израз за удобочитаемост.

Следващата стъпка е обявяването на два екземпляра на 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}

Същността не се е променила, но тези резултати изглеждат по-ясно.

Ако мислите и за глава 4, вече сте виждали няколко типа свързвания в действие. Те са изброени в табл. 5.2. Тази таблица отразява възможностите за свързване, актуални за версия 1.0.0 на Kafka Streams; в бъдещи издания може да настъпят промени.

В заключение, напомням основното: можете да свързвате потоци от събития (KStream) и потоци от актуализации (KTable) с помощта на локално състояние. Освен това, ако размерът на справочните данни не е твърде голям, можете да използвате обекта GlobalKTable. GlobalKTable репликира всички секции на всеки от възлите на приложението Kafka Streams, осигурявайки достъпност на всички данни независимо от секцията, към която съответства ключът.

По-нататък ще видим възможността на Kafka Streams, която позволява наблюдение на промените в състоянието без консумация на данни от темата в Kafka.

5.3.5. Достъпно за запитвания състояние

Вече изпълнявахме няколко операции, свързани с състоянието, и винаги извеждахме резултатите в конзолата (за цели на разработката) или ги записвахме в тема (за индустриални цели). При записването на резултатите в тема е необходимо да използвате консуматор на Kafka, за да ги прегледате.

Четенето на данни от тези теми може да се счита за разновидност на материализирани изгледи (materialized views). За нашите нужди можем да използваме определението на материализирания изглед от «Уикипедия»: «…физически обект на базата данни, съдържащ резултатите от изпълнението на заявка. Например, той може да бъде локална копия на отдалечени данни или подмножество от редове и/или колони на таблица или резултатите от свързване, или сводна таблица, получена чрез агрегация» (https://en.wikipedia.org/wiki/Materialized_view).

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

Възможността за директни заявки към съхранения в състояние е от голямо значение. Това означава, че можете да създавате приложения — информационни табла, без нуждата преди това да получавате данни от потребителя на Kafka. Тя повишава и ефективността на приложението, тъй като не е нужно отново да записвате данни:

• благодарение на локалността на данните, можете бързо да се обърнете към тях;
• изключва се дублирането на данни, тъй като те не се записват във външно хранилище.

Основното, което искам да запомните: можете директно да изпълнявате заявки към състоянието от приложението. Не може да се подцени възможностите, които това ви дава. Вместо да консумирате данни от Kafka и да записвате записи в базата данни за приложението, можете да изпълнявате заявки към съхранения в състояние с същия резултат. Директните заявки към съхранения в състояние означават по-малко количество код (отсъствие на потребител) и по-малко софтуер (отсъствие на необходимост от таблица в базата данни за съхранение на резултатите).

В настоящата глава обхванахме значителен обем информация, затова временно прекратяваме дискусията за интерактивни запитвания към хранилищата на състоянията. Но не се притеснявайте: в глава 9 ще създадем просто приложение — информационна табла с интерактивни запитвания. За демонстрация на интерактивните запитвания и възможностите за тяхната интеграция в приложения на Kafka Streams ще използваме някои от примера на тази и предишните глави.

Резюме

• Обектите KStream олицетворяват потоци от събития, сравними с вмъквания в база данни. Обектите KTable олицетворяват потоци от обновления, те повече наподобяват обновления в база данни. Размерът на обекта KTable не нараства, старите записи се заменят с нови.
• Обектите KTable са необходими за операции по агрегация.
• С помощта на оконни операции може да се разделят агрегирани данни по времеви интервали.
• Благодарение на обектите GlobalKTable може да се получи достъп до справочни данни от всяка точка на приложението, независимо от секционирането.
• Възможни са съединения между обектите KStream, KTable и GlobalKTable.

До сега се фокусирахме върху създаването на приложения на Kafka Streams с помощта на високоуровневия DSL KStream. Въпреки че високоуровневият подход позволява създаването на подредени и кратки програми, използването му представлява известен компромис. Работата с DSL KStream означава повишаване на краткостта на кода за сметка на намаляване на степента на контрол. В следващата глава ще разгледаме нискоуровневия API на обработващите възли и ще опитаме други компромиси. Програмите ще станат по-дълги, отколкото досега, но ще имаме възможността да създаваме практически всеки обработващ възел, от който можем да имаме нужда.

→ По-подробно с книгата може да се запознаете на сайта на издателството

→ За потребителите на Хаброк, отстъпка от 25% с купон — Kafka Streams

→ При реализация на плащането за хартиената версия на книгата, на e-mail ще бъде изпратена електронна книга.

Източник: habr.com