Книгата „Кафка потоци во акција. Апликации и микросервиси за работа во реално време“

Здраво, жители на Хабро! Оваа книга е погодна за секој развивач кој сака да ја разбере обработката на нишки. Разбирањето на дистрибуираното програмирање ќе ви помогне подобро да ги разберете Кафка и Кафка струи. Би било убаво да се знае самата рамка на Кафка, но тоа не е неопходно: ќе ви кажам сè што ви треба. Искусните програмери на Кафка и почетниците ќе научат како да креираат интересни апликации за обработка на стрим користејќи ја библиотеката Кафка струи во оваа книга. Среден и напреден Java програмери кои веќе се запознаени со концепти како што е серијализацијата ќе научат да ги применуваат своите вештини за да креираат апликации на Kafka Streams. Изворниот код на книгата е напишан во Java 8 и значително ја користи синтаксата на изразување Java 8 ламбда, така што ќе ви помогне да знаете како да работите со ламбда функции (дури и на друг програмски јазик).

Извадок. 5.3. Агрегација и операции на прозорци

Во овој дел, ќе продолжиме да ги истражуваме најперспективните делови на Кафка потоци. Досега ги опфативме следните аспекти на Кафка потоци:

• создавање топологија за обработка;
• користење на состојба во стриминг апликации;
• вршење на поврзувања со поток на податоци;
• разлики помеѓу преносите на настани (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: вредност од типот enum AutoOffsetReset.EARLIEST (има и НАЈНОВО), поставена со методот Consumed.withOffsetResetPolicy. Овој тип на набројување може да се користи за одредување стратегија за ресетирање на поместување за секој KStream или KTable и има предност пред опцијата за ресетирање поместување од конфигурацијата.

GroupByKey и GroupBy

Интерфејсот KStream има два методи за групирање записи: GroupByKey и GroupBy. И двајцата враќаат KGroupedTable, па можеби се прашувате која е разликата помеѓу нив и кога да се користи кој?

Методот GroupByKey се користи кога клучевите во KStream веќе не се празни. И што е најважно, знамето „бара повторна партиција“ никогаш не беше поставено.

Методот GroupBy претпоставува дека сте ги смениле клучевите за групирање, така што знамето за повторна партиција е поставено на точно. Вршењето спојувања, агрегации итн. по методот GroupBy ќе резултира со автоматско повторно партиционирање.
Резиме: Секогаш кога е можно, треба да користите GroupByKey наместо GroupBy.

Јасно е што прават методите mapValues ​​и groupBy, па ајде да го разгледаме методот sum() (се наоѓа во src/main/java/bbejeck/model/ShareVolume.java) (Список 5.3).

Методот ShareVolume.sum го враќа вкупниот број на продажби на акции, а резултатот од целиот синџир на пресметки е објект KTable . Сега ја разбирате улогата што ја игра KTable. Кога ќе пристигнат објектите ShareVolume, соодветниот објект KTable го складира најновото тековно ажурирање. Важно е да се запамети дека сите ажурирања се рефлектираат во претходната shareVolumeKTable, но не сите се испраќаат понатаму.

Потоа ја користиме оваа табела за агрегација (по број на акции со кои се тргува) за да дојдеме до петте компании со најголем обем на акции со кои се тргува во секоја индустрија. Нашите акции во овој случај ќе бидат слични на оние за првата агрегација.

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

Сето ова значи дека нашиот агрегатор, FixedSizePriorityQueue, не ги собира сите вредности со еден клуч, туку складира движечка сума од количините на N најтргуваните типови акции. Секој влезен запис го содржи вкупниот број на продадени акции досега. KTable ќе ви даде информации за акциите на компаниите кои моментално се најтргувани, без да се бара тркалачка агрегација на секое ажурирање.

Научивме да правиме две важни работи:

• групирајте ги вредностите во KTable со заеднички клуч;
• врши корисни операции како што се собирање и собирање на овие групирани вредности.

Знаењето како да се извршат овие операции е важно за да се разбере значењето на податоците што се движат низ апликацијата Кафка стримс и да се разберат какви информации тие носат.

Исто така, собравме некои од клучните концепти дискутирани претходно во оваа книга. Во поглавје 4, разговаравме за тоа колку е важна локалната состојба толерантна за грешки за апликација за стриминг. Првиот пример во ова поглавје покажа зошто локалната држава е толку важна - ви овозможува да ги следите информациите што веќе сте ги виделе. Локалниот пристап ги избегнува доцнењата на мрежата, што ја прави апликацијата пофункционална и поотпорна на грешки.

Кога вршите каква било операција за собирање или собирање, мора да го наведете името на државната продавница. Операциите за собирање и собирање враќаат примерок на KTable, а KTable користи складирање на состојби за да ги замени старите резултати со нови. Како што видовте, не се испраќаат сите ажурирања по цевководот, и ова е важно затоа што операциите за агрегација се дизајнирани да продуцираат резиме информации. Ако не ја примените локалната состојба, KTable ќе ги проследи сите резултати од собирање и собирање.

Следно, ќе го разгледаме извршувањето на операции како што е агрегација во одреден временски период - таканаречени операции за прозорци.

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

Во претходниот дел, воведовме лизгачка конволуција и агрегација. Апликацијата изврши континуирано собирање на продажбите на акции проследено со агрегација на петте најтргувани акции на берзата.

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

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

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

Постојат две можни причини за ова следење. Еден од нив е потребата да се знае што купуваат/продаваат лидерите на пазарот. Ако овие големи играчи и софистицирани инвеститори видат можност, има смисла да се следи нивната стратегија. Втората причина е желбата да се забележат какви било можни знаци на нелегално тргување со инсајдери. За да го направите ова, ќе треба да ја анализирате корелацијата на големите продажни скокови со важни соопштенија за печатот.

Таквото следење се состои од следниве чекори:

• креирање стрим за читање од темата берзански трансакции;
• групирање на дојдовните записи по ID на купувачот и симбол на акции. Повикувањето на методот groupBy враќа инстанца од класата KGroupedStream;
• Методот KGroupedStream.windowedBy враќа проток на податоци ограничен на временски прозорец, што овозможува агрегација со прозорци. Во зависност од типот на прозорецот, се враќа или TimeWindowedKStream или SessionWindowedKStream;
• број на трансакции за операцијата за агрегација. Прозорецот на протокот на податоци одредува дали одреден запис е земен предвид во оваа бројка;
• пишување резултати на тема или нивно емитување на конзолата за време на развојот.

Топологијата на оваа апликација е едноставна, но јасната слика за неа би била корисна. Ајде да погледнеме на Сл. 5.11.

Следно, ќе ја разгледаме функционалноста на операциите на прозорецот и соодветниот код.

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

Постојат три типа на прозорци во Кафка потоци:

• сесиски;
• „превртување“ (превртување);
• лизгање/рипнување.

Кој да изберете зависи од вашите деловни барања. Прозорците што се превртуваат и скокаат се временски ограничени, додека прозорците на сесиите се ограничени од активноста на корисникот - времетраењето на сесиите се одредува само од тоа колку е активен корисникот. Главната работа што треба да се запамети е дека сите типови на прозорци се засноваат на ознаките за датум/време на записите, а не на времето на системот.

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

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

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

Слика 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:XNUMX часот.

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 часот.

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

• сесиите не се прозорци со фиксна големина. Времетраењето на сесијата се одредува според активноста во даден временски период;
• Ознаките за датум/време во податоците одредуваат дали настанот спаѓа во постоечка сесија или во период на мирување.

Следно, ќе разговараме за следниот тип на прозорец - прозорци „превртливи“.

„Провртени“ прозорци

Прозорците што паѓаат снимаат настани што спаѓаат во одреден временски период. Замислете дека треба да ги снимите сите трансакции со акции на одредена компанија на секои 20 секунди, па да ги соберете сите настани во тој временски период. На крајот од интервалот од 20 секунди, прозорецот се превртува и се префрла на нов интервал на набљудување од 20 секунди. Слика 5.14 ја илустрира оваа ситуација.

Како што можете да видите, сите настани примени во последните 20 секунди се вклучени во прозорецот. На крајот од овој временски период, се креира нов прозорец.

Описот 5.6 покажува код што ја демонстрира употребата на паѓачки прозорци за снимање на трансакциите на акции на секои 20 секунди (се наоѓа во src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Со оваа мала промена на повикот на методот TimeWindows.of, можете да користите ролна прозорец. Овој пример не го повикува методот до(), така што ќе се користи стандардниот интервал на задржување од 24 часа.

Конечно, време е да се префрлиме на последната од опциите на прозорецот - „прескокнување“ на прозорци.

Лизгачки ("скокачки") прозорци

Лизгачките/подскокнувачки прозорци се слични на прозорците што се вртат, но со мала разлика. Лизгачките прозорци не чекаат до крајот на временскиот интервал пред да се создаде нов прозорец за обработка на неодамнешните настани. Тие започнуваат нови пресметки по интервал на чекање помал од времетраењето на прозорецот.

За да ги илустрираме разликите помеѓу прозорите што се тркалаат и скокаат, да се вратиме на примерот на броење берзански трансакции. Нашата цел е сè уште да го броиме бројот на трансакции, но не сакаме да чекаме цело време пред да го ажурираме бројачот. Наместо тоа, ќе го ажурираме бројачот на пократки интервали. На пример, сепак ќе го броиме бројот на трансакции на секои 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. Исто така, со користење на НАЈРАНО набројување, табелата се пополнува со записи на самиот почеток.

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

Поврзување на новости со податоци за бројот на трансакции

Создавањето врска не е тешко. Ќе користиме лево приклучување во случај да нема новости за акции за соодветната индустрија (потребниот код може да се најде во датотеката src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Список 5.10).

Овој лево-Приклучи оператор е прилично едноставен. За разлика од спојувањата во Поглавје 4, методот JoinWindow не се користи бидејќи при изведување на спојување KStream-KTable, има само еден запис во KTable за секое копче. Таквата врска не е временски ограничена: записот е или во KTable или отсутен. Главниот заклучок: користејќи објекти на KTable можете да го збогатите KStream со поретко ажурирани референтни податоци.

Сега ќе погледнеме поефикасен начин за збогатување на настани од KStream.

5.3.4. ГлобалКТабела објекти

Како што можете да видите, постои потреба да се збогатат преносите на настани или да се додаде контекст на нив. Во поглавје 4 ги видовте врските помеѓу два објекти на KStream, а во претходниот дел ја видовте врската помеѓу KStream и KTable. Во сите овие случаи, неопходно е повторно да се подели протокот на податоци при мапирање на клучевите на нов тип или вредност. Понекогаш репартиционирањето се прави експлицитно, а понекогаш Кафка стримс го прави тоа автоматски. Повторната партиција е неопходна затоа што клучевите се променети и записите мора да завршат во нови делови, инаку врската ќе биде невозможна (ова беше дискутирано во Поглавје 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='Barney, Smith' company="Exxon", transactions= 17}

Суштината не е променета, но овие резултати изгледаат појасни.

Ако одбројувате до Поглавје 4, веќе сте виделе неколку типови на врски во акција. Тие се наведени во табелата. 5.2. Оваа табела ги одразува можностите за поврзување од верзијата 1.0.0 на Kafka Streams; Нешто може да се промени во идните изданија.

За да ги завршиме работите, да ги повториме основите: можете да поврзувате преноси на настани (KStream) и да ги ажурирате преносите (KTable) користејќи локална состојба. Алтернативно, ако големината на референтните податоци не е премногу голема, можете да го користите објектот GlobalKTable. GlobalKTables ги реплицира сите партиции на секој апликативен јазол на Kafka Streams, осигурувајќи дека сите податоци се достапни без разлика на која партиција одговара клучот.

Следно ќе ја видиме функцијата Кафка текови, благодарение на која можеме да ги набљудуваме промените на состојбата без да трошиме податоци од темата на Кафка.

5.3.5. Состојба што може да се праша

Веќе извршивме неколку операции кои вклучуваат состојба и секогаш ги изнесуваме резултатите на конзолата (за развојни цели) или ги пишуваме на тема (за производствени цели). Кога пишувате резултати на тема, треба да користите потрошувач на Кафка за да ги видите.

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

Kafka Streams исто така ви овозможува да извршувате интерактивни прашања во државните продавници, овозможувајќи ви директно да ги читате овие материјализирани погледи. Важно е да се забележи дека барањето до државната продавница е операција само за читање. Ова осигурува дека нема да морате да се грижите за случајно да ја направите состојбата неконзистентна додека вашата апликација обработува податоци.

Способноста директно да се прашуваат државните продавници е важна. Ова значи дека можете да креирате апликации на контролната табла без претходно да преземате податоци од потрошувачот на Кафка. Исто така, ја зголемува ефикасноста на апликацијата, поради фактот што нема потреба повторно да се пишуваат податоци:

• благодарение на локалитетот на податоците, тие можат брзо да се пристапат;
• Умножувањето на податоците е елиминирано, бидејќи не е запишано на надворешно складирање.

Главната работа што сакам да ја запомните е дека можете директно да ја побарате состојбата од вашата апликација. Можностите што ви ги дава ова не можат да бидат преценети. Наместо да трошите податоци од Кафка и да складирате записи во базата на податоци за апликацијата, можете да ги прашате државните продавници со истиот резултат. Директните барања до државните продавници значат помалку код (без потрошувач) и помалку софтвер (нема потреба од табела со база на податоци за складирање на резултатите).

Опфативме доста работи во ова поглавје, па засега ќе ја оставиме нашата дискусија за интерактивни прашања против државните продавници. Но, не грижете се: во Поглавје 9, ќе создадеме едноставна апликација за контролна табла со интерактивни прашања. Ќе користи некои од примерите од ова и претходните поглавја за да демонстрира интерактивни прашања и како можете да ги додадете во апликациите на Kafka Streams.

Краток преглед

• Објектите на KStream претставуваат текови на настани, споредливи со вметнувањата во базата на податоци. Објектите на KTable претставуваат текови за ажурирање, повеќе како ажурирања на база на податоци. Големината на објектот KTable не расте, старите записи се заменуваат со нови.
• Објектите на KTable се потребни за операциите на агрегација.
• Користејќи ги операциите за прозорци, можете да ги поделите збирните податоци во временски кофи.
• Благодарение на објектите на GlobalKTable, можете да пристапите до референтните податоци каде било во апликацијата, без оглед на партиционирањето.
• Можни се поврзувања помеѓу објектите KStream, KTable и GlobalKTable.

Досега, се фокусиравме на создавање апликации на Kafka Streams користејќи го DSL на високо ниво KStream. Иако пристапот на високо ниво ви овозможува да креирате уредни и концизни програми, неговото користење претставува компромис. Работата со DSL KStream значи зголемување на концизноста на вашиот код со намалување на степенот на контрола. Во следното поглавје, ќе го разгледаме API-то на јазолот за ракувач на ниско ниво и ќе се обидеме со други компромиси. Програмите ќе бидат подолги отколку што беа претходно, но ќе можеме да создадеме речиси секој јазол на управувач што можеби ни треба.

→ Повеќе детали за книгата можете да најдете на веб-страница на издавачот

→ За Хаброжители 25% попуст со користење на купон - Кафка потоци

→ По уплата на хартиената верзија на книгата, електронска книга ќе биде испратена по е-пошта.

Извор: www.habr.com