Книга «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 по общия им ключ;
• извършваме полезни операции, като свиване и агрегиране върху тези групирани стойности.

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

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

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

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

5.3.2. Операции с прозорци

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

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

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

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

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

Такова наблюдение се състои от етапи като:

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

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

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

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

В Kafka Streams съществуват три типа прозорци:

• сесийни;
• падащи (tumbling);
• плъзгащи/«скачащи» (sliding/hopping).

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

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

Сесийни прозорци

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

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

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

Нека използваме сесийни прозорци, за да уловим информация за борсовите транзакции. Реализацията на сесийните прозорци е показана в листинг 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.

От разказаното в тази секция е важно да запомним следните ключови подробности:

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

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

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

"Тумблирущи" (tumbling) прозорци улавят събития, попадащи в определен времеви интервал. Представете си, че трябва да улавяте всички борсови транзакции на дадена компания на всеки 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 да замените ключа с името на индустрията, след което да извлечете от екземпляра Windowed обекта TransactionSummary.

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

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

Обърнете внимание, че имената на темите се описват с помощта на избираеми типове.

Сега, след като подготвихме всички компоненти, остава да напишем кода за свързване (който можете да намерите в файла src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (листинг 5.13).

Въпреки че в този код има две свързвания, те са организирани под формата на верига, тъй като отделно нито един от резултатите им не се използва. Резултатите се извеждат в края на цялата операция.

Когато стартирате горепосочената операция за свързване, ще получите резултати от следния вид:

{customer='Барни, Смит' 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

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

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