Книгата „Кафка Потоци в действие. Приложения и микроуслуги за работа в реално време"

Здравейте Habrites! Тази книга е подходяща за всеки разработчик, който иска да разбира стрийминг. Разбирането на разпределеното програмиране ще ви помогне да разберете по-добре 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 с помощта на разширител (вторият параметър в извикването на агрегатния метод), но ако вече съществува друг запис със същия ключ, тогава старият запис се премахва с помощта на subtractor (третият параметър в извикването на агрегатния метод).

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Видове прозорци

Има три вида прозорци в Kafka Streams:

• сесия;
• "тумблинг" (тумблинг);
• плъзгане / "скачане" (плъзгане / подскачане).

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

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

Прозорци на сесии

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

Фигура 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 секунди след края или началото на текущата сесия в текущата (активна) сесия.

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

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

Най-накрая е време да преминем към последната от опциите на прозореца, прескачащи прозорци.

Плъзгащи се ("скачащи") прозорци

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

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

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

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

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

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

След това ще научим как да конвертираме 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 се използва в настройките. Също така, благодарение на използването на НАЙ-РАННОТО изброяване, таблицата се попълва със записи в самото начало.

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

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

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

Този оператор 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-5 (съответният код може да бъде намерен в src/main/java/bbejeck/chapter_XNUMX/GlobalKTableExample.java) и ще го свържем с два обекта GlobalKTable.

Обсъждали сме това и преди, така че няма да се повтарям. Но имайте предвид, че кодът във функцията toStream().map е абстрахиран във функционален обект за четимост вместо вграден ламбда израз.

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

Имайте предвид, че имената на темите са описани с помощта на изброени типове.

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

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

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

{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 потребител, за да ги видите.

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

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

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

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

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

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

Обобщение

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

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

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

→ За Khabrozhiteli 25% отстъпка от купона - Потоци на Кафка

→ При заплащане на хартиената версия на книгата, на електронната поща се изпраща електронна книга.

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