Raamat "Kafka Streams tegevuses. Rakendused ja mikroteenused reaalajas töötamiseks"

Tere, Habrojekte! See raamat sobib igale arendajale, kes soovib voogude töötlemisse süveneda. Jaotatud programmeerimise mõistmine aitab paremini mõista Kafka ja Kafka Streams'i. Oluline oleks tunda ka Kafka raamistikku, kuid see pole hädavajalik: ma selgitan teile kõik, mida peate teadma. Kogenud Kafka arendajad, nagu ka algajad, saavad selle raamatu kaudu meisterdada huvitavaid voogude töötlemise rakendusi, kasutades Kafka Streams'i raamatukogu. Kesk- ja kõrgema taseme Java arendajad, kes on juba tuttavad selliste mõisted nagu serialiseerimine, õpivad, kuidas rakendada oma oskusi Kafka Streams'i rakenduste loomisel. Raamatu lähtekood on kirjutatud Java 8-s ja kasutab ulatuslikult Java 8 lambda-väljendite süntaksit, seega tuleb kasuks oskused lambda-funktsioonide töötluseks (isegi teises programmeerimiskeeles).

Lõik. 5.3. Aggregeerimine ja aknoperatsioonid

Selles jaotises viime läbi uurimise kõige lubavamatest osadest Kafka Streams'ist. Kuni praeguseni oleme käsitlenud järgmisi Kafka Streams'i aspekte:

• töötlemise topoloogia loomine;
• oleku kasutamine voogude rakendustes;
• andmevoogude ühendamine;
• erinevused sündmusvoogude (KStream) ja uuenduste voogude (KTable) vahel.

Järgmistes näidetes koondame kõik need elemendid kokku. Samuti tutvute aknoperatsioonidega — veel ühe suurepärase võimalusega voogude rakendustes. Meie esimene näide on lihtne aggregeerimine.

5.3.1. Aktsiate müügimahtude aggregeerimine tööstusharu kaupa

Aggregeerimine ja grupeerimine on eluliselt vajalikud tööriistad voogude andmete töötlemisel. Üksikute rekordite uurimine reaalajas osutub tihti ebapiisavaks. Andmetest lisainfo väljavõtmiseks on vajalikud gruppeerimine ja kombineerimine.

Selles näites peate end intraday kaupleja kostüümile panema, kes peab jälgima aktsiate müügimahu mahtu mitmetes tööstusharudes. Eriti huvitavad teid viis ettevõtet, mille aktsiate müügimaht on igas tööstusharus suurim.

Selle aggregeerimise jaoks on vaja mitmeid järgmised samme andmete vajaliku vormi tõlkimiseks (üldiselt öeldes).

1. Loo source põhined teemarikkal, mis avaldab aktsiate kauplemise kohta toorinfot. Me peame kuvama StockTransaction tüüpi objekti ShareVolume tüübiks. Asi on selles, et StockTransaction objekt sisaldab müügimetainfot, kuid me vajame ainult andmeid müüdud aktsiate arvu kohta.
2. Klasifitseerime ShareVolume andmeid aktsia sümbolite kaupa. Pärast sümbolite järgi grupeerimist saame need andmed kokku pöörata aktsiate müügi vahekokkuvõteteks. Tuleb märkida, et meetod KStream.groupBy tagastab KGroupedStream tüüpi eseme. KTable'i saame võtta, kutsudes edasi KGroupedStream.reduce meetodit.

Mis on KGroupedStream liides?

Meetodid KStream.groupBy ja KStream.groupByKey tagastavad KGroupedStream tüüpi eseme. KGroupedStream on vahepealne esitus sündmuste voost pärast grupeerimist võtmete järgi. See ei ole mõeldud otseseks töötamiseks. Selle asemel kasutatakse KGroupedStream'i kogumise toiminguteks, mille tulemuseks on alati KTable. Kuna kogumise tegemise tulemuseks on KTable ja see katab olekuhoidla, võivad mitte kõik uuendused liikuda edasi torustiku kaudu.

Meetod KTable.groupBy tagastab sarnase KGroupedTable'i — vahepealse esitluse värskenduste voost, mis on uuesti grupeeritud võtme järgi.

Teeme väikese pausi ja vaatame joonist 5.9, mis näitab, mida me saavutasime. See topoloogia peaks teile juba tuttav olema.

Vaadakem nüüd koodi selle topoloogia jaoks (seda saab leida failist src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (loetelu 5.2).

Esitatud koodi iseloomustab lühidus ja suur hulk tehteid, mis tehakse mõne reaga. Meetodi builder.stream esimeses parameetris võite märgata midagi uut: loetletud tüübi väärtus AutoOffsetReset.EARLIEST (on olemas ka LATEST), mis on määratud meetodi Consumed.withOffsetResetPolicy abil. Selle loetletud tüübi abil saate määrata nihkestrateegia iga KStream või KTable jaoks, see on eeliseks nihkemeetodi seadistusele konfiguratsioonis.

GroupByKey ja GroupBy

KStream liideses on kaks meetodit kirje gruppeeringuks: GroupByKey ja GroupBy. Mõlemad tagastavad KGroupedTable'i, seega võite küsida: milline on nende vahe ja millal kumbagi kasutada?

GroupByKey meetodit rakendatakse, kui KStream'i võti on juba olemas. Pealegi ei ole lipp 'nõuab ümberjaotamist' kunagi seadistatud.

GroupBy meetod eeldab, et olete muutnud rühmitamise võtmeid, mistõttu on ümberjaotamise lipp seadistatud väärtusele true. GroupBy järel teostatavad ühendused, agregatsioonid jne viivad automaatse ümberjaotamiseni.
Kokkuvõtteks: kui vähegi võimalik, tuleks kasutada GroupByKey meetodit, mitte GroupBy'd.

Kuna mapValues ja groupBy meetodite funktsioone on selge, vaatame sum() meetodit (see on leitav failist src/main/java/bbejeck/model/ShareVolume.java) (loetelu 5.3).

ShareVolume.sum meetod tagastab aktsiate müügimahtude vahepealse summa ning kogu arvutuste ahela tulemus on objekt KTable. Nüüd mõistate, millist rolli KTable mängib. Kui ShareVolume objekte lisatakse, salvestatakse vastavas KTable objektis viimane asjakohane uuendus. On oluline mitte unustada, et kõik uuendused kajastuvad eelnevas shareVolumeKTable'is, kuid mitte kõik saadetakse edasi.

Edasi, kasutades seda KTable'i, teeme agregatsiooni (müüdud aktsiate arvu alusel), et saada viis ettevõtet, kellel on iga tööstusharu suurimad aktsiate müügivoldid. Meie tegevus on sarnane esimese agregatsiooniga.

1. Tehke veel üks groupBy operatsioon, et rühmitada eraldi ShareVolume objekte tööstusharude järgi.
2. Alustage ShareVolume objektide summeerimisega. Seekord on agregatsiooni objectiks fikseeritud suurusega prioriteediköit. Sellises fikseeritud suurusega köites hoitakse ainult viit ettevõtet, kellel on kõige rohkem müüdud aktsiaid.
3. Muuda eelneva punkti järjekorrad stringiväärtuseks ja tagasta viie kõige paremini müüdud aktsia põhjal nende tööstusharude järgi.
4. Kanna tulemused stringivormis teemasse.

Joonisel 5.10 on kujutatud andmevoogude topoloogia graafik. Nagu näete, on teise töötlemise tsükkel üsna lihtne.

Nüüd, kui olete saanud selge arusaama selle teise töötlemise tsükli struktuurist, võite tutvuda selle lähtekoodiga (leiate selle failist src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (loetelu 5.4).

Selles initsialisaatoris on muutuja fixedQueue. See on kohandatud objekt — adapter java.util.TreeSet'ile, mida kasutatakse, et jälgida N suurimat müügihulka kahanevas järjestuses.

Te olete juba tuttavad groupBy ja mapValues väljakutsetega, seega ei jääme neile pikemalt peatuma (me kutsume KTable.toStream meetodit, kuna KTable.print meetod on vananenud). Kuid te pole veel näinud KTable versiooni meetodist aggregate(), seega kulutame veidi aega selle arutamiseks.

Nagu te meenute, erineb KTable sellest, et sama võtmega kirjeid peetakse uuendusteks. KTable asendab vana kirje uuega. Aggregeerimine toimub sarnaselt: aggregeeritakse viimased kirjed sama võtmega. Kirje saabudes lisatakse see FixedSizePriorityQueue klassi eksemplarisse koos summeerijaga (teine parameeter aggregate meetodi üleskutses), kuid kui sama võtmega on juba olemas teine kirje, siis vana kirje eemaldatakse vähendajaga (kolmas parameeter aggregate meetodi üleskutses).

See kõik tähendab, et meie aggregeerija, FixedSizePriorityQueue, ei aggregeeri kõiki sama võtmega väärtusi, vaid hoiab N kõige enam müüdud aktsiate liikide libisevat summat. Igas saabunud kirjes on üldine müüdud aktsiate hulk seni. KTable annab teile teavet selle kohta, milliste ettevõtete aktsiaid müüakse praegu rohkem, libisev aggregeerimine iga uuenduse puhul ei ole vajalik.

Oleme õppinud tegema kahte olulist asja:

• gruppima KTable's väärtusi nende ühise võtme järgi;
• teostama nendele rühmitatud väärtustele kasulikke toiminguid, nagu kokkuvõte ja aggregeerimine.

Neid toiminguid osata on oluline, et mõista andmete tähendust, mis liigub läbi Kafka Streams rakenduse, ja välja selgitada, millist teavet nad kannavad.

Me oleme samuti kokku viinud mõned võtmekontseptsioonid, millest oleme varem selles raamatus rääkinud. 4. peatükis rääkisime, kui oluline on voogedastusrakenduse jaoks talletatud, lokaalne olek. Selles peatükis toodud esimene näide demonstreeris, miks lokaalne olek on nii oluline – see võimaldab jälgida, millist teavet olete juba näinud. Lokaalne ligipääs aitab vältida võrgulisi viivitusi, muutes rakenduse tootlikumaks ja vigu taluvamaks.

Iga koondamis- või agregatsioonitegevuse läbiviimisel tuleb täpsustada olekutehnika nime. Koondamis- ja agregatsioonitegevused tagastavad KTable'i koopia, samal ajal kui KTable kasutab olekutehnikat, et asendada vanad tulemused uutega. Nagu olete näinud, ei saadeta kaugeltki kõiki värskendusi torust edasi, ja see on oluline, kuna agregatsioonitegevused on mõeldud lõpptulemuse saamiseks. Ilma lokaalse olekuta saadab KTable edasi kõik agregatsiooni ja koondamise tulemused.

Järgmises vaates vaatame läbi selliseid tegevusi nagu agregatsioon kindla ajavahemiku jooksul – nii nimetatud aktsioonitegevusi (windowing operations).

5.3.2. Aktsioonitegevused

Eelnevas peatükis tutvusime "libiseva" koondamise ja agregatsiooniga. Rakendus tegi pidevat koondamist aktsiate müügimahu kohta, järgides viie kõige rohkem müüdud aktsia koondamist.

Mõnikord on sarnased pidevad agregatsioonid ja koondamised vajalikud. Kuid mõnikord tuleb teostada tegevusi ainult määratud ajavahemiku jooksul. Näiteks välja arvutada, kui palju börsil tehinguid on tehtud konkreetse ettevõtte aktsiatega viimase 10 minuti jooksul. Või kui palju kasutajaid klõpsas uuel reklaambänneril viimase 15 minuti jooksul. Rakendus võib neid tegevusi teha korduvalt, kuid tulemustega, mis on seotud ainult määratud ajavahemikega (ajaliste akendega).

Börsil tehingute arvestamine ostja järgi

Järgmises näites meie teema on börsil tehingute jälgimine mitme kaupleja puhul – kas suurte ettevõtete või nutikate individuaalsete finantsteadlaste puhul.

On olemas kaks võimalikku põhjust sarnase jälgimise jaoks. Üks neist on vajadus teada, mida ostavad/müüvad turuliidrid. Kui need suured mängijad ja kogenud investorid näevad endale avanevaid võimalusi, on mõistlik järgida nende strateegiat. Teine põhjus on soov märgata võimalikke ilmsi ebaseaduslikke tehingutega, mis kasutavad siseteavet. Selleks peate analüüsima suurte müügivoolude ja oluliste pressiteadete korrelatsiooni.

Selline jälgimine hõlmab järgmisi etappe:

• loome voolu lugemiseks teemast stock-transactions;
• grupime sissetulevad kirjed ostja ID ja aktsia sümboli järgi. GroupBy meetodi kutsumine tagastab KGroupedStream klassi eksemplari;
• KGroupedStream.windowedBy meetodi abil tagastatakse ajavahemiku piiramiseks andmevoog, mis võimaldab teostada akna agregatsiooni. Olenemata akna tüübist tagastatakse kas TimeWindowedKStream või SessionWindowedKStream;
• tehingute arvestamine agregatsiooni toimingu jaoks. Aken määrab, kas konkreetne kande arvesse võtmine arvutamisel toimub;
• tulemuste salvestamine teemasse või nende väljastamine konsooli arenduse käigus.

Rakenduse topoloogia on lihtne, kuid visuaalne esitus ei teeks kahju. Vaatame joonist 5.11.

Edasi vaatleme akna toimingute funktsionaalsust ja vastavat koodi.

Akna tüübid

Kafka Streams'is on kolm tüüpi aknaid:

• seanssi;
• „rulluvad” (tumbling);
• libisevad/„hüppavad” (sliding/hopping).

Mille valida — sõltub ärinõuetest. „Rulluvad” ja „hüppavad” aknad piirduvad ajaga, samas kui seansside piirangud on seotud kasutajate tegevustega — seansi (seansside) kestus määratakse ainuilt selle järgi, kui aktiivselt käitub kasutaja. Peamine on mitte unustada, et kõik akna tüübid põhinevad kannete kuupäeva/aegade märgistel, mitte süsteemi ajast.

Edasi rakendame meie topoloogiat iga akna tüübi korral. Täielik kood toodakse ainult esimeses näites, teiste akna tüüpide jaoks ei muutu midagi peale akna toimingu tüübi.

Seanssi aknad

Seanside aknad erinevad oluliselt kõigist teistest aknatüüpidest. Need on piiratud mitte niivõrd ajaga, vaid kasutaja tegevusega (või selle üksuse tegevusega, mida soovite jälgida). Seanside aknad eraldatakse tegevusetuse perioodide kaupa.

Joonisel 5.12 illustreeritakse seanside akna mõistet. Väiksem seanss sulandub temast vasakul asuva seansi sisse. Paremal asuv seanss on eraldi, kuna järgneb pikale tegevusetuse perioodile. Seanside aknad põhinevad kasutajate tegevusel, kuid kasutavad kuupäeva/kellaaja märke, et määrata, mille seansi alla vastav kirje kuulub.

Seanside aknade kasutamine börsitehingute jälgimiseks

Kasutame seanside aknaid börsitehingute teabe kogumiseks. Seanside akna rakendamine on näidatud loendis 5.5 (mida võib leida failist src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).

Enamik selle topoloogia operatsioone on teile juba tuttavad, nii et pole vajalik neid siin uuesti käsitleda. Aga siin on ka mõned uued elemendid, millest me praegu räägime.

Iga groupBy operatsiooni puhul tehakse tavaliselt mingisugune agregatsioonioperatsioon (agregatsioon, kokkusurumine või arvu lugemine). Saame teostada kas kumulatiivset agregatsiooni jooksva kogusummaga või akna agregatsiooni, kus arvesse võetakse kirjeid määratud ajavahemiku jooksul.

Loendi 5.5 kood loendab tehingute arvu seanside akende piires. Joonisel 5.13 analüüsitakse neid toiminguid samm-sammult.

Kasutades funktsiooni windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)), loome seanside akna 20-sekundilise tegevusetuse intervalliga ja 15-minutilise säilitamisintervalliga. 20-sekundiline tegevusetuse intervall tähendab, et rakendus hõlmab igat kirjet, mis saabub 20 sekundi jooksul pärast praeguse seansi lõppu või algust, praegusesse (aktiivsesse) seanssi.

Edasi anname teada, millist agregatsiooni toimingut tuleb seanssi aknas teostada - antud juhul count. Kui sisenev kirje ületab tegevusetuse intervalli (mõlemalt poolt ajatempli), siis rakendus loob uue seansi. Salvestamise intervall tähendab seansi hoidmist teatud aja jooksul ja võimaldab hilinenud andmeid, mis ületavad seansi tegevusetuse perioodi, kuid võivad siiski liituda. Lisaks on uue seansi algus ja lõpp, mis tulenevad ühendamisest, vastavalt varasemale ja hilisemale ajatempli.

Vaadakem mõningaid kirjeid count meetodist, et näha, kuidas seansid toimivad (tabel 5.1).

Kirjete saabumise korral otsime juba olemasolevaid seansse sama võtme jaoks, mille lõpp aeg on vähem kui praegune ajatempli - tegevusetuse intervall ja mille algusaeg on rohkem kui praegune ajatempli + tegevusetuse intervall. Selle järel ühendatakse neli kirjet tabelist 5.1 ühte seanssi järgmiselt.

1. Esimesena saabub kirje 1, nii et algusaeg on sama, mis lõppaeg ja on 00:00:00.

2. Järgmisena saabub kirje 2, ja me otsime seansse, mis lõpevad hiljemalt 23:59:55 ja algavad hiljemalt 00:00:35. Leiame kirje 1 ja ühendame seansid 1 ja 2. Võtame seansi 1 algusaja (varasem) ja seansi 2 lõppaaja (hilisem), nii et meie uus seanss algab 00:00:00 ja lõpeb 00:00:15.

3. Saabub kirje 3, me otsime seansse ajavahemikus 00:00:30 kuni 00:01:10 ja ei leia ühtegi. Lisame teise seansi võtmega 123-345-654,FFBE, algusega ja lõpuga 00:00:50.

4. Saabub kirje 4, ja me otsime seansse ajavahemikus 23:59:45 kuni 00:00:25. Seekord leitud mõlemad seansid - 1 ja 2. Kõik kolm seanssi ühendatakse üheks, algusajaga 00:00:00 ja lõppaeg 00:00:15.

Käesolevas jaotises on mõningaid olulisi nüansse, mida tuleks meeles pidada:

• seansid ei ole fikseeritud suurusega aknad. Seansi kestus määratakse aktiveerituse järgi määratud ajavahemikus;
• ajatempli andmed määravad, kas sündmus kuulub olemasolevasse seanssi või tegevusetuse intervalli.

Edasi arutame järgmist akna tüüpi - „veerevad” aknad.

„Veerevad” aknad

„Veerlevad“ (tumbling) aknad haaravad sündmusi, mis toimuvad kindlas ajavahemikus. Kujutlege, et peate salvestama kõiki mingisuguse ettevõtte börsitehinguid iga 20 sekundi tagant, seega kogute kõik sündmused selle ajavahemiku jooksul. 20-sekundilise intervalli lõppedes „veerleb“ aken ja läheb uude 20-sekundilise vaatlusperioodi. Joonis 5.14 illustreerib seda olukorda.

Kuidas näha, on kõik viimase 20 sekundi jooksul saabunud sündmused aknas. Selle ajavahemiku lõppedes luuakse uus aken.

Loendis 5.6 on toodud kood, mis demonstreerib „veerlevate“ akende kasutamist börsitehingute igakuiseks salvestamiseks iga 20 sekundi tagant (seda võib leida failist src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Selle väikese muudatuse tõttu meetodi TimeWindows.of kutsumisel on võimalik kasutada „veerlevat“ akent. Antud näites ei ole meetodi until() kutsumist, mistõttu kasutatakse vaikimisi salvestusintervalli, mis on 24 tundi.

Lõpuks on aeg liikuda viimase aknatüübi juurde – „hüppavad“ (hopping) aknad.

Libisevad („hüppavad“) aknad

Libisevad/„hüppavad“ (sliding/hopping) aknad on sarnased „veerlevatele“, kuid väikese erinevusega. Libisevad aknad ei oota ajavahemiku lõppu enne uue akna loomist hiljutiste sündmuste töötlemiseks. Nad käivitavad uued arvutused pärast ooteaega, mis on lühem akna kestusest.

Erinevuste illustreerimiseks „veerlevate“ ja „hüppavate“ akende vahel pöördume tagasi börsitehingute arvestuse näite juurde. Meie eesmärk on endiselt arvestada tehingute arvu, kuid me ei soovi oodata kogu ajavahemikku enne arvu värskendamist. Selle asemel värskendame arvu lühemate ajavahemike tagant. Näiteks loeme endiselt tehingute arvu iga 20 sekundi tagant, kuid värskendame arvu iga 5 sekundi tagant, nagu on näidatud joonisel 5.15. Selle tulemusena on meil kolm tulemuste akent, millel on kattuvad andmed.

Loendis 5.7 on toodud kood libisevate akende seadistamiseks (seda võib leida failist src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

„Veerev” akent saab muuta „hüppavaks” funktsiooni advanceBy() kutsumisega. Antud näites on salvestusintervall 15 minutit.

Selles osas olete näinud, kuidas piirata tulemusi ajavahemikega. Eriti tasub meeles pidada kolm asja, mida sellelt, jaotiselt kaasa võtta:

• seanss akna suurus ei ole piiratud aja jooksul, vaid kasutajate aktiivsusega;
• „veerlevad” aknad annavad ülevaate sündmustest antud ajavahemiku jooksul;
• „hüppavate” akende kestus on kindel, kuid need uuendatakse sageli ja võivad kõikides akendes sisaldada kattuvaid kirjeid.

Seejärel õppime, kuidas KTable tagasi KStreami muuta ühendamiseks.

5.3.3. KStreami ja KTable'i ühendamine

Kapis 4 arutasime kahe KStreami objekti ühendamist. Nüüd õpime KTable'i ja KStreami ühendamist. Seda võib vaja minna järgmise lihtsa põhjuse tõttu. KStream on kirjevoog, KTable aga kirjeuuenduste voog, kuid mõnikord võib olla vajalik lisada voogudele konteksti KTable'i uuenduste kaudu.

Võtame andmed börsitehingute arvu kohta ja ühendame need börsiuudistega vastavate tööstusharude järgi. Siin on, mida tuleb teha, et saavutada seda olemasoleva koodiga.

1. Muutke börsitehingute arvu KTable objekt KStreamiks, asendades võtme nii, et see tähendaks tööstusharu, mis vastab antud aktsiate sümbolile.
2. Looge KTable objekt, mis loeb andmeid börsiuudiste teemast. See uus KTable klassifitseeritakse tööstusharude järgi.
3. Ühendage uudiste uuendused börsitehingute arvu teabega vastavates tööstusharudes.

Nüüd vaatame, kuidas seda tegevuskava ellu viia.

KTable'i muutmine KStreamiks

KTable'i KStreamiks muutmiseks tuleb teha järgmist.

1. Kutsuda välja meetod KTable.toStream().
2. Kasutades meetodi KStream.map kutset, asendada võti tööstusharu nimega ja seejärel välja võtta TransactionSummary objekti.

Seome need toimingud ahelaks järgmiselt (koodi leiate failist src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (loend 5.8).

Kuna me teeme operatsiooni KStream.map, toimub KStreami tagastatud eksemplari ümberjaotamine automaatselt, kui seda kasutatakse ühenduses.

Oleme lõpetanud muundamisprotsessi, nüüd peame looma KTable objekti börsiuudiste lugemiseks.

KTable loomine börsiuudiste jaoks

Õnneks piisab KTable objekti loomiseks ühest koodireast (seda koodi saab leida failist src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (loetelu 5.9).

Oluline on märkida, et Serde objekte ei ole vaja määrata, kuna seadistustes kasutatakse stringide Serde-d. Samuti täidab tabel tänu EARLIEST loetlemise kasutamisele kirjeid alguses.

Nüüd saame liikuda viimase sammu juurde — ühendamine.

Uudiste värskenduste ja tehingute arvu andmete ühendamine

Ühenduse loomine ei ole keeruline. Kasutame vasakut ühendust juhuks, kui vastavas tööstusharus pole börsiuudiseid (vajalik kood on saadaval failis src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (loetelu 5.10).

See leftJoin operaator on üsna lihtne. Erinevalt 4. peatükist, kus kasutati JoinWindow meetodit, ei kasutata KStream-KTable ühendamisel, kuna KTable'is on iga võtme kohta ainult üks kirje. Selline ühendamine ei ole ajaliselt piiratud: kirje kas on KTable'is või pole. Peamine järeldus: KTable objektide abil saab KStreami rikastada harvemini uuendatud viidatud andmetega.

Nüüd vaatame tõhusamat viisi KStreami sündmuste rikastamiseks.

5.3.4. GlobalKTable objektid

Kuidas te nüüd aru saite, on vaja rikastada sündmuste vooge või lisada neile konteksti. 4. peatükis nägite kahe KStream objekti ühendamist, ja eelmises jaos KStreami ja KTable'i ühendamist. Kõigil nendel juhtudel on andmevoo ümberjaotamine vajalik, kui võtmed muudetakse uue tüübi või väärtuse jaoks. Mõnikord toimub ümberjaotamine selgelt, mõnikord teeb Kafka Streams seda automaatselt. Ümberjaotamine on vajalik, kuna võtmed on muutunud ja kirjed peavad olema uuteks sektsioonideks, vastasel juhul ei saa ühendust teha (sellest arutati 4. peatükis jaotises "Andmete ümberjaotamine" alajaos 4.2.4).

Uuesti jaotamine on tasuline

Uuesti jaotamine nõuab kulutusi — täiendavaid ressursikulu vahepealsete teemade loomisele, duplikaatandmete säilitamisele veel ühes teemas; see toob kaasa ka viivituste suurenemise seoses andmete lugemise ja kirjutamisega sellesse teemasse. Lisaks, kui on vajalik teostada ühendust rohkem kui ühe aspekti või mõõtme alusel, tuleb ühendused organiseerida ahelana, kuvada kirjed uute võtmete järgi ja uuesti viia läbi uuesti jaotamise protsess.

Ühendamine väiksemate andmekogumitega

Mõnes olukorras on viidatud andmete maht, millega plaanitakse ühendust, suhteliselt väike, nii et nende täiendavad koopiad mahuvad igasse sõlme kohapeal. Selle jaoks on Kafka Streams'is olemas klass GlobalKTable.

GlobalKTable'i eksemplarid on ainulaadsed, kuna rakendus replitseerib kõik andmed igasse sõlmesse. Kuna igas sõlmes on kõik andmed, ei ole vaja sündmuste voogu jaotada viidatud andmete võtme alusel, et need oleksid kergesti ligipääsetavad kõikidele sektsioonidele. GlobalKTable objektidega on samuti võimalik teostada võtmeta ühendusi. Naaseme tagasi ühe eelneva näite juurde, et demonstreerida seda võimalust.

KStream objektide ühendamine GlobalKTable objektidega

Alajaotises 5.3.2 viisime läbi akna agregatsiooni börsitehingute üle ostjate lõikes. Selle agregatsiooni tulemused nägid välja umbes järgmised:

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

Kuigi need tulemused vastasid seatud eesmärgile, oleks mugavam, kui seal oleks ka kliendi nimi ja ettevõtte täielik nimi. Kliendi nime ja ettevõtte nime lisamiseks on võimalik teha tavalisi ühendusi, kuid see nõuab kahe võtme kaardistamist ja uuesti jaotamist. GlobalKTable abil saab selliseid toimingute kulutusi vältida.

Selleks kasutame objekti countStream näiteks 5.11 (vastav kood asub failis src/main/java/bbejeck/chapter_5/GlobalKTableExample.java), ühendades selle kahe GlobalKTable objektiga.

Oleme seda juba varem arutanud, seega ei hakka ma seda kordama. Kuid mainin, et funktsiooni toStream().map kood on loetavuse huvides abstraheeritud objekti-funktsiooni, mitte sisseehitatud lambdana.

Järgmine etapp on kahe GlobalKTable instantsi deklareerimine (toodud kood on saadaval failis src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (loetelu 5.12).

Pange tähele, et teema nimed on määratletud loendatavate tüüpide abil.

Nüüd, kui oleme kõik komponendid ette valmistanud, jääb kirjutada kood ühendamiseks (mida saab leida failist src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (loetelu 5.13).

Kuigi selles koodis on kaks ühendust, on need korraldatud ahelaks, kuna eraldi ei kasutata ühtegi nende tulemusi. Tulemused väljastatakse kogu operatsiooni lõpus.

Käivitades ülaltoodud ühenduse operatsiooni, saate järgmiste sõnumite tulemusi:

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

Sisu pole muutunud, kuid need tulemused näevad välja arusaadavamad.

Kui arvestada peatükki 4, olete juba näinud mitmeid tüüpe ühendusi tegevuses. Need on loetletud tabelis 5.2. See tabel kajastab ühenduste võimalusi, mis on seotud Kafka Streams versiooniga 1.0.0; tulevastes väljalasemetes võivad mõned asjad muutuda.

Kokkuvõtteks tahan meelde tuletada põhipunkti: saate ühendada sündmuste vooge (KStream) ja värskenduste vooge (KTable) kohaliku oleku abil. Lisaks, kui kaasandmete maht pole liiga suur, saab kasutada objekti GlobalKTable. GlobalKTable kopeerib kõik osad iga Kafka Streams rakenduse sõlme, tagades seeläbi kõigi andmete kättesaadavuse sõltumata sellest, millisele osale vastab võti.

Edasi vaatame Kafka Streams'i võimalust, mis võimaldab jälgida oleku muutusi ilma andmeid Kafka teemast tarbimata.

5.3.5. Küsitav olek

Oleme juba teinud mitmeid olekuga seotud operatsioone ja alati oleme tulemused konsooli väljastanud (arenduse jaoks) või salvestanud need teemasse (tööstuslikeks otstarveteks). Tulemuste teemasse kirjutamisel tuleb neid vaadata Kafka tarbija abil.

Nende teemade andmete lugemist võib pidada materialiseeritud vaadete (materialized views) üheks vormiks. Meie jaoks võime kasutada Wikipedia määratlust materialiseeritud vaate kohta: „… füüsiline andmebaasi objekt, mis sisaldab päringu täitmise tulemusi. Näiteks võib see olla kohaliku koopia kaugandmetest, või ridade ja/või veergude alamhulk tabelist või ühendamise tulemustest, või kogumisega saadud kokkuvõtte tabel.” (https://en.wikipedia.org/wiki/Materialized_view).

Kafka Streams võimaldab samuti interaktiivsete päringute (interactive queries) tegemist olekuhoidlatest, mis annab võimaluse neid materialiseeritud vaateid otse lugeda. Oluline on märkida, et päring olekuhoidlas on ainult lugemisoperatsioon. See tähendab, et saate olla kindel, et rakenduse andmete töötlemise käigus ei muuda te olekut ebatäpsuseks.

Otseste päringute võimalus olekuhoidlatesse on äärmiselt oluline. See tähendab, et saab luua rakendusi — informatiivseid paneele, ilma et peaksite esialgu andmeid Kafka tarbijalt hankima. Samuti suurendab see rakenduse efektiivsust, kuna andmeid ei pea uuesti salvestama:

• andmete lokaliseerimise tõttu on neile kiire juurdepääs;
• välditakse andmete dubleerimist, kuna neid ei salvestata välisesse andmehoidlasse.

Peamine, mida ma tahaksin, et te meeles peaksite: saate otse pärida olekust rakendusest. Selle võimaluse olulisust on raske üle hinnata. Selle asemel, et tarbida andmeid Kafka-st ja salvestada kirjeid andmebaasi rakenduse jaoks, saate teha päringuid olekuhoidlatele sama tulemusega. Otsesed päringud olekuhoidlatele tähendavad vähem koodi (tarbijat pole) ja vähem tarkvara (andmete tulemuste salvestamiseks pole vajadust andmebaasitabeli järele).

Oleme selles peatükis käsitlenud suurt hulka teavet, seega teeme nüüd väikese pausi interaktiivsete päringute arutelust olekutehul. Ära muretse: peatükis 9 loome lihtsa rakenduse - interaktiivsete päringutega teabelehe. Sellel on mõned näited käesolevast ja eelmise peatüki interaktiivsetest päringutest ja nende lisamisest Kafka Streams rakendustesse.

Elulookirjeldus

• KStream objektid kehastavad sündmuste vooge, mis sarnanevad andmebaasi lisamisele. KTable objektid kehastavad uuenduste vooge, nad sarnanevad rohkem andmebaasi uuendustega. KTable objekti suurus ei kasva, vanad kirjeldused asendatakse uutega.
• KTable objektid on vajalikud agregatsioonitegevuste jaoks.
• Aknategevuste abil saab agreggeeritud andmed jagada ajavahemike kaupa.
• GlobalKTable objektide kaudu on võimalik juurdepääs viidatud andmetele rakenduse igas punktis sõltumata osade jaotumisest.
• KStream, KTable ja GlobalKTable objektide vahel on lubatud ühendused.

Kuni praeguseni oleme keskendunud Kafka Streams rakenduste loomisele KStream kõrgetasemelise DSL-i abil. Kuigi kõrgetasemeline lähenemine võimaldab luua puhtaid ja lühikesi programme, tähendab selle kasutamine teatud kompromissi. Töö DSL KStreamiga tähendab koodi lühendamise suurendamist kontrollitaseme alandamise arvelt. Järgmises peatükis vaatame madalama taseme API-d sõlme töötlejatest ja proovime teisi kompromisse. Programmid saavad olema pikemad kui nad enne olid, aga meil on võimalus luua praktiliselt igasuguseid töötlejaid, mis meil kunagi vajalikud võivad olla.

→ Rohkem teavet raamatu kohta leiate aadressilt kirjastuse veebisaidil

→ Habr Residentide jaoks allahindlus 25% kupongiga — Kafka Streams

→ Pärast trükitud raamatuga maksmist saadetakse elektrooniline raamat e-posti teel.

Allikas: habr.com