Књига „Кафкини токови у акцији. Апликације и микросервисе за рад у реалном времену"

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

Извод. 5.3. Операције агрегације и прозора

У овом одељку ћемо прећи на истраживање делова Кафкиних токова који највише обећавају. До сада смо покрили следеће аспекте Кафкиних токова:

• креирање топологије обраде;
• коришћење стања у апликацијама за стриминг;
• извођење веза токова података;
• разлике између токова догађаја (КСтреам) и токова ажурирања (КТабле).

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

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

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

У овом примеру ћете обући костим дневног трговца који треба да прати обим продаје акција компанија у неколико индустрија. Конкретно, заинтересовани сте за пет компанија са највећом продајом удела у свакој индустрији.

Такво агрегирање ће захтевати неколико следећих корака да би се подаци превели у жељени облик (уопштено говорећи).

1. Креирајте извор заснован на темама који објављује сирове информације о трговању акцијама. Мораћемо да мапирамо објекат типа СтоцкТрансацтион у објекат типа СхареВолуме. Поента је да објекат СтоцкТрансацтион садржи метаподатке о продаји, али су нам потребни само подаци о броју акција које се продају.
2. Групирајте податке о обиму акција према симболу акције. Када се групишу по симболу, ове податке можете сажмити у међузбирове обима продаје залиха. Вреди напоменути да метода КСтреам.гроупБи враћа инстанцу типа КГроупедСтреам. И можете добити КТабле инстанцу даљим позивањем методе КГроупедСтреам.редуце.

Шта је интерфејс КГроупедСтреам

Методе КСтреам.гроупБи и КСтреам.гроупБиКеи враћају инстанцу КГроупедСтреам. КГроупедСтреам је посредни приказ тока догађаја након груписања по кључевима. Уопште није намењен директном раду са њим. Уместо тога, КГроупедСтреам се користи за операције агрегације, чији је резултат увек КТабле. А пошто је резултат операција агрегације КТабле и они користе складиште стања, могуће је да се сва ажурирања као резултат не шаљу даље низ цевовод.

Метода КТабле.гроупБи враћа сличну КГроупедТабле - посредни приказ тока ажурирања, прегруписаних по кључу.

Хајде да направимо кратку паузу и погледамо Сл. 5.9, који показује шта смо постигли. Ова топологија би вам већ требала бити позната.

Погледајмо сада код за ову топологију (може се наћи у датотеци срц/маин/јава/ббејецк/цхаптер_5/АггрегатионсАндРедуцингЕкампле.јава) (Листинг 5.2).

Дати код се одликује краткоћом и великим обимом радњи које се изводе у неколико редова. Можда ћете приметити нешто ново у првом параметру методе буилдер.стреам: вредност типа енума АутоОффсетРесет.ЕАРЛИЕСТ (постоји и ЛАТЕСТ), постављена помоћу методе Цонсумед.витхОффсетРесетПолици. Овај тип набрајања се може користити за спецификацију стратегије ресетовања помака за сваки КСтреам или КТабле и има предност над опцијом ресетовања померања из конфигурације.

ГроупБиКеи и ГроупБи

КСтреам интерфејс има две методе за груписање записа: ГроупБиКеи и ГроупБи. Оба враћају КГроупедТабле, па се можда питате која је разлика између њих и када да користите коју?

Метод ГроупБиКеи се користи када кључеви у КСтреам-у већ нису празни. И што је најважније, заставица „захтева поновно партиционисање“ никада није постављена.

Метод ГроупБи претпоставља да сте променили кључеве за груписање, тако да је заставица репартиције постављена на труе. Извођење спајања, агрегирања итд. након методе ГроупБи ће резултирати аутоматским поновним партиционирањем.
Резиме: Кад год је могуће, требало би да користите ГроупБиКеи уместо ГроупБи.

Јасно је шта раде методе мапВалуес и гроупБи, па хајде да погледамо метод сум() (који се налази у срц/маин/јава/ббејецк/модел/СхареВолуме.јава) (листинг 5.3).

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

Затим користимо ову КТтабелу да агрегирамо (према броју тргованих акција) да бисмо дошли до пет компанија са највећим обимом акција којима се тргује у свакој индустрији. Наше акције у овом случају ће бити сличне онима за прво здруживање.

1. Извршите још једну операцију гроупБи да бисте груписали појединачне СхареВолуме објекте по делатностима.
2. Започните сумирање СхареВолуме објеката. Овог пута објекат агрегације је ред са приоритетом фиксне величине. У овом реду фиксне величине задржава се само пет компанија са највећим бројем продатих акција.
3. Мапирајте редове из претходног пасуса на вредност низа и вратите првих пет акција које се највише тргују по броју по делатностима.
4. Запишите резултате у облику низа на тему.

На сл. Слика 5.10 приказује граф топологије тока података. Као што видите, други круг обраде је прилично једноставан.

Сада када имамо јасно разумевање структуре овог другог круга обраде, можемо се обратити његовом изворном коду (наћи ћете га у датотеци срц/маин/јава/ббејецк/цхаптер_5/АггрегатионсАндРедуцингЕкампле.јава) (Листинг 5.4) .

Овај иницијализатор садржи променљиву фикедКуеуе. Ово је прилагођени објекат који је адаптер за јава.утил.ТрееСет који се користи за праћење првих Н резултата у опадајућем редоследу акција којима се тргује.

Већ сте видели позиве гроупБи и мапВалуес, тако да нећемо улазити у њих (позивамо методу КТабле.тоСтреам јер је метода КТабле.принт застарела). Али још увек нисте видели КТабле верзију аггрегате(), па ћемо потрошити мало времена на дискусију о томе.

Као што се сећате, оно што чини КТабле другачијим је то што се записи са истим кључевима сматрају ажурирањима. КТабле замењује стари унос новим. Агрегација се дешава на сличан начин: агрегирају се најновији записи са истим кључем. Када запис стигне, он се додаје у инстанцу класе ФикедСизеПриоритиКуеуе помоћу сабирача (други параметар у позиву агрегатне методе), али ако већ постоји други запис са истим кључем, стари запис се уклања помоћу одузимања (трећи параметар у позив агрегатне методе).

Све ово значи да наш агрегатор, ФикедСизеПриоритиКуеуе, не агрегира све вредности једним кључем, већ чува покретни збир количина Н најтргованијих врста акција. Сваки улазни унос садржи укупан број до сада продатих акција. КТабле ће вам дати информације о томе којим се акцијама компанија тренутно највише тргује, без потребе за сталном агрегацијом сваког ажурирања.

Научили смо да радимо две важне ствари:

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

Познавање начина на који се обављају ове операције важно је за разумевање значења података који се крећу кроз Кафка Стреамс апликацију и разумевање које информације оне носе.

Такође смо спојили неке од кључних концепата о којима смо раније говорили у овој књизи. У поглављу 4, расправљали смо о томе колико је локално стање отпорно на грешке важно за апликацију за стриминг. Први пример у овом поглављу показао је зашто је локална држава толико важна—омогућава вам да пратите које информације сте већ видели. Локални приступ избегава кашњења на мрежи, чинећи апликацију ефикаснијом и отпорнијом на грешке.

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

Затим ћемо погледати извођење операција као што је агрегација у одређеном временском периоду - такозване операције прозора.

5.3.2. Прозорске операције

У претходном одељку увели смо клизну конволуцију и агрегацију. Апликација је извршила континуирано прикупљање продаје акција праћено агрегацијом пет најтргованијих акција на берзи.

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

Рачунање мењачких трансакција по купцу

У следећем примеру, пратићемо трансакције акција код више трговаца — било великих организација или паметних појединачних финансијера.

Постоје два могућа разлога за ово праћење. Једна од њих је потреба да се зна шта тржишни лидери купују/продају. Ако ови велики играчи и софистицирани инвеститори виде прилику, има смисла пратити њихову стратегију. Други разлог је жеља да се уоче сви могући знаци илегалне трговине инсајдерима. Да бисте то урадили, мораћете да анализирате корелацију великих скокова продаје са важним саопштењима за штампу.

Такво праћење се састоји од следећих корака:

• креирање тока за читање из теме берзанских трансакција;
• груписање улазних записа према ИД-у купца и симболу акције. Позивање методе гроупБи враћа инстанцу класе КГроупедСтреам;
• Метода КГроупедСтреам.виндоведБи враћа ток података ограничен на временски прозор, који омогућава агрегацију у прозорима. У зависности од типа прозора, враћа се или ТимеВиндоведКСтреам или СессионВиндоведКСтреам;
• број трансакција за операцију агрегације. Прозорски ток података одређује да ли је одређени запис узет у обзир у овом бројању;
• писање резултата у тему или њихово слање на конзолу током развоја.

Топологија ове апликације је једноставна, али би јасна слика о њој била од помоћи. Хајде да погледамо Сл. 5.11.

Затим ћемо погледати функционалност прозорских операција и одговарајући код.

Врсте прозора

Постоје три типа прозора у Кафка Стреамс:

• сессионал;
• „тумблинг” (тумбање);
• клизање/скакање.

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

Затим имплементирамо нашу топологију са сваким од типова прозора. Комплетан код ће бити дат само у првом примеру за друге типове прозора ништа се неће променити осим типа операције прозора.

Прозори сесије

Прозори сесије се веома разликују од свих других типова прозора. Оне су ограничене не толико временом колико активношћу корисника (или активношћу ентитета који желите да пратите). Прозори сесија су ограничени периодима неактивности.

Слика 5.12 илуструје концепт прозора сесије. Мања сесија ће се спојити са сесијом са леве стране. А седница са десне стране биће одвојена јер прати дуг период неактивности. Прозори сесије су засновани на активности корисника, али користе датум/време из уноса да би одредили којој сесији тај унос припада.

Коришћење прозора сесије за праћење трансакција акција

Хајде да користимо прозоре сесије да ухватимо информације о трансакцијама размене. Имплементација прозора сесије је приказана у Листингу 5.5 (који се може наћи у срц/маин/јава/ббејецк/цхаптер_5/ЦоунтингВиндовингАндКТаблеЈоинЕкампле.јава).

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

Било која операција гроупБи обично обавља неку врсту операције агрегације (агрегације, скупљања или бројања). Можете да извршите или кумулативно агрегацију са текућим укупним износом, или агрегацију прозора, која узима у обзир записе унутар одређеног временског периода.

Код у Листингу 5.5 броји број трансакција унутар прозора сесије. На сл. 5.13 ове акције се анализирају корак по корак.

Позивањем виндоведБи(СессионВиндовс.витх(твентиСецондс).унтил(фифтеенМинутес)) креирамо прозор сесије са интервалом неактивности од 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,ФФБЕ, која почиње и завршава се у 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 секунди (налази се у срц/маин/јава/ббејецк/цхаптер_5/ЦоунтингВиндовингАндКтаблеЈоинЕкампле.јава).

Са овом малом променом у позиву методе ТимеВиндовс.оф, можете користити прозор који се окреће. Овај пример не позива методу унтил(), тако да ће се користити подразумевани интервал задржавања од 24 сата.

Коначно, време је да пређемо на последњу од опција прозора – „скакање“ прозора.

Клизни ("скакање") прозори

Клизни/скокови прозори су слични прозорима који се преврћу, али са малом разликом. Клизни прозори не чекају до краја временског интервала пре креирања новог прозора за обраду недавних догађаја. Они почињу нове прорачуне након интервала чекања мањег од трајања прозора.

Да бисмо илустровали разлике између превртања и скакања прозора, вратимо се на пример бројања берзанских трансакција. Наш циљ је и даље да пребројимо број трансакција, али не желимо да чекамо читаво време пре ажурирања бројача. Уместо тога, ажурираћемо бројач у краћим интервалима. На пример, и даље ћемо бројати број трансакција сваких 20 секунди, али ажурирати бројач сваких 5 секунди, као што је приказано на Сл. 5.15. У овом случају, на крају имамо три прозора резултата са подацима који се преклапају.

Листинг 5.7 показује код за дефинисање клизних прозора (налази се у срц/маин/јава/ббејецк/цхаптер_5/ЦоунтингВиндовингАндКтаблеЈоинЕкампле.јава).

Прозор који се преврће може се конвертовати у прозор за скакање додавањем позива методу адванцеБи(). У приказаном примеру, интервал чувања је 15 минута.

Видели сте у овом одељку како да ограничите резултате агрегације на временске прозоре. Посебно желим да запамтите следеће три ствари из овог одељка:

• величина прозора сесије није ограничена временским периодом, већ активностима корисника;
• „завртљиви“ прозори пружају преглед догађаја у датом временском периоду;
• Трајање прозора за прескакање је фиксно, али се они често ажурирају и могу садржати уносе који се преклапају у свим прозорима.

Затим ћемо научити како да конвертујемо КТабле назад у КСтреам за везу.

5.3.3. Повезивање КСтреам и КТабле објеката

У 4. поглављу смо разговарали о повезивању два КСтреам објекта. Сада морамо да научимо како да повежемо КТабле и КСтреам. Ово може бити потребно из следећег једноставног разлога. КСтреам је ток записа, а КТабле је ток ажурирања записа, али понекад ћете можда желети да додате додатни контекст у ток записа користећи ажурирања из КТабле.

Узмимо податке о броју берзанских трансакција и комбинујмо их са берзанским новостима за релевантне индустрије. Ево шта треба да урадите да бисте то постигли с обзиром на код који већ имате.

1. Конвертујте КТабле објекат са подацима о броју трансакција акција у КСтреам, након чега следи замена кључа са кључем који означава индустријски сектор који одговара овом симболу акције.
2. Креирајте КТабле објекат који чита податке из теме са вестима са берзе. Овај нови КТабле ће бити категорисан по индустријским секторима.
3. Повежите ажуриране вести са информацијама о броју берзанских трансакција по индустријским секторима.

Сада да видимо како да применимо овај акциони план.

Конвертујте КТабле у КСтреам

Да бисте конвертовали КТабле у КСтреам, потребно је да урадите следеће.

1. Позовите метод КТабле.тоСтреам().
2. Позивањем методе КСтреам.мап, замените кључ именом индустрије, а затим преузмите објекат ТрансацтионСуммари из Виндовед инстанце.

Ове операције ћемо повезати заједно на следећи начин (код се може наћи у датотеци срц/маин/јава/ббејецк/цхаптер_5/ЦоунтингВиндовингАндКтаблеЈоинЕкампле.јава) (Листинг 5.8).

Пошто изводимо операцију КСтреам.мап, враћена КСтреам инстанца се аутоматски поново партиционише када се користи у вези.

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

Креирање КТабле-а за вести о акцијама

На срећу, креирање КТабле објекта захтева само једну линију кода (код се може наћи у срц/маин/јава/ббејецк/цхаптер_5/ЦоунтингВиндовингАндКтаблеЈоинЕкампле.јава) (листинг 5.9).

Вреди напоменути да се не морају специфицирати Серде објекти, пошто се стринг Сердес користи у подешавањима. Такође, коришћењем НАЈРАНИЈЕ енумерације, табела се попуњава записима на самом почетку.

Сада можемо да пређемо на последњи корак - повезивање.

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

Стварање везе није тешко. Користићемо леви спој у случају да нема вести о акцијама за релевантну индустрију (потребан код се може наћи у датотеци срц/маин/јава/ббејецк/цхаптер_5/ЦоунтингВиндовингАндКтаблеЈоинЕкампле.јава) (Листинг 5.10).

Овај оператор лефтЈоин је прилично једноставан. За разлику од спајања у поглављу 4, ЈоинВиндов метода се не користи јер када се врши спајање КСтреам-КТабле, постоји само један унос у КТабле за сваки кључ. Таква веза није временски ограничена: запис је или у КТабле или га нема. Главни закључак: коришћењем КТабле објеката можете обогатити КСтреам са мање често ажурираним референтним подацима.

Сада ћемо погледати ефикаснији начин да обогатимо догађаје из КСтреам-а.

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

Као што видите, постоји потреба за обогаћивањем токова догађаја или додавањем контекста у њих. У поглављу 4 видели сте везе између два КСтреам објекта, ау претходном одељку видели сте везу између КСтреам-а и КТабле-а. У свим овим случајевима, потребно је поново партиционисати ток података приликом мапирања кључева на нови тип или вредност. Понекад се репартиционирање врши експлицитно, а понекад Кафка Стреамс то ради аутоматски. Поновно партиционисање је неопходно јер су се кључеви променили и записи морају завршити у новим секцијама, иначе ће повезивање бити немогуће (о томе је било речи у поглављу 4, у одељку „Поновно партиционисање података“ у пододељку 4.2.4).

Поновно партиционисање има цену

Поновно партиционисање захтева трошкове – додатне трошкове ресурса за креирање међутема, чување дупликата података у другој теми; значи и повећано кашњење због писања и читања из ове теме. Поред тога, ако треба да се придружите више од једног аспекта или димензије, морате уланчати спојеве, мапирати записе са новим кључевима и поново покренути процес поновног партиционисања.

Повезивање са мањим скуповима података

У неким случајевима, обим референтних података који се повезују је релативно мали, тако да се њихове комплетне копије лако могу уклопити локално на сваком чвору. За овакве ситуације, Кафка Стреамс обезбеђује класу ГлобалКТабле.

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

Повезивање КСтреам објеката са ГлобалКТабле објектима

У пододељку 5.3.2 извршили смо прозор агрегације мењачких трансакција по купцима. Резултати ове агрегације изгледали су отприлике овако:

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

Иако су ови резултати служили сврси, било би корисније да су име купца и пуни назив компаније такође били приказани. Да бисте додали име клијента и име компаније, можете да урадите уобичајена спајања, али ћете морати да урадите два мапирања кључева и поновно партиционисање. Са ГлобалКТабле можете избећи трошкове таквих операција.

Да бисмо то урадили, користићемо објекат цоунтСтреам из Листинга 5.11 (одговарајући код се може наћи у срц/маин/јава/ббејецк/цхаптер_5/ГлобалКТаблеЕкампле.јава) и повезати га са два ГлобалКТабле објекта.

О томе смо већ разговарали, па нећу да понављам. Али примећујем да је код у функцији тоСтреам().мап апстрахован у објекат функције уместо у инлине ламбда израз ради читљивости.

Следећи корак је декларисање две инстанце ГлобалКТабле (приказан код се може наћи у датотеци срц/маин/јава/ббејецк/цхаптер_5/ГлобалКТаблеЕкампле.јава) (Листинг 5.12).

Имајте на уму да су називи тема описани помоћу набројаних типова.

Сада када имамо све компоненте спремне, остаје само да напишемо код за везу (који се може наћи у датотеци срц/маин/јава/ббејецк/цхаптер_5/ГлобалКТаблеЕкампле.јава) (Листинг 5.13).

Иако постоје два споја у овом коду, они су уланчани јер се ниједан њихов резултат не користи засебно. Резултати се приказују на крају целе операције.

Када покренете горњу операцију придруживања, добићете следеће резултате:

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

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

Ако одбројате до 4. поглавља, већ сте видели неколико врста веза у акцији. Они су наведени у табели. 5.2. Ова табела одражава могућности повезивања од верзије 1.0.0 Кафка Стреамс-а; Нешто се може променити у будућим издањима.

Да завршимо ствари, хајде да поновимо основе: можете да повежете токове догађаја (КСтреам) и токове ажурирања (КТабле) користећи локално стање. Алтернативно, ако величина референтних података није превелика, можете користити ГлобалКТабле објекат. ГлобалКТаблес реплицира све партиције на сваки Кафка Стреамс апликациони чвор, осигуравајући да су сви подаци доступни без обзира на то којој партицији кључ одговара.

Затим ћемо видети функцију Кафка Стреамс, захваљујући којој можемо да посматрамо промене стања без конзумирања података из Кафкине теме.

5.3.5. Упитно стање

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

Читање података из ових тема може се сматрати врстом материјализованих погледа. За наше потребе можемо користити дефиницију материјализованог погледа са Википедије: „... физички објекат базе података који садржи резултате упита. На пример, то може бити локална копија удаљених података, или подскуп редова и/или колона табеле или резултата спајања, или збирна табела добијена агрегацијом“ (хттпс://ен.википедиа.орг/вики /Материјализовани_поглед).

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

Важна је могућност директног постављања упита у складишта стања. То значи да можете креирати апликације за контролну таблу без потребе да прво преузимате податке од Кафка потрошача. Такође повећава ефикасност апликације, због чињенице да нема потребе за поновним писањем података:

• захваљујући локалности података, њима се може брзо приступити;
• дуплирање података је елиминисано, јер се не уписују у екстерну меморију.

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

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

Резиме

• КСтреам објекти представљају токове догађаја, упоредиви са уметањима у базу података. КТабле објекти представљају токове ажурирања, више као ажурирања базе података. Величина објекта КТабле не расте, стари записи се замењују новим.
• КТабле објекти су потребни за операције агрегације.
• Користећи операције прозора, можете поделити агрегиране податке у временске сегменте.
• Захваљујући ГлобалКТабле објектима, можете приступити референтним подацима било где у апликацији, без обзира на партиционисање.
• Везе између КСтреам, КТабле и ГлобалКТабле објеката су могуће.

До сада смо се фокусирали на изградњу Кафка Стреамс апликација користећи КСтреам ДСЛ високог нивоа. Иако вам приступ високог нивоа омогућава да креирате уредне и концизне програме, његово коришћење представља компромис. Рад са ДСЛ КСтреам-ом значи повећање сажетости вашег кода смањењем степена контроле. У следећем поглављу ћемо погледати АПИ чвора руковаоца ниског нивоа и испробати друге компромисе. Програми ће бити дужи него што су били раније, али ћемо моћи да креирамо скоро сваки чвор за руковање који нам може затребати.

→ Више детаља о књизи можете пронаћи на веб-сајт издавача

→ За Хаброзхители 25% попуста користећи купон - Кафка Стреамс

→ По уплати папирне верзије књиге, електронска књига ће бити послата на е-маил.

Извор: ввв.хабр.цом