Il-ktieb “Kafka Streams in Action. Applikazzjonijiet u mikroservizzi għal xogħol f'ħin reali"

Hello, residenti Khabro! Dan il-ktieb huwa adattat għal kull żviluppatur li jrid jifhem l-ipproċessar tal-ħajt. Il-fehim tal-ipprogrammar distribwit jgħinek tifhem aħjar Kafka u Kafka Streams. Ikun sabiħ li tkun taf il-qafas Kafka innifsu, iżda dan mhux meħtieġ: ​​Ngħidlek dak kollu li għandek bżonn. Żviluppaturi Kafka b'esperjenza u novizzi bl-istess mod se jitgħallmu kif joħolqu applikazzjonijiet interessanti għall-ipproċessar tal-flussi bl-użu tal-librerija Kafka Streams f'dan il-ktieb. Iżviluppaturi Java intermedji u avvanzati diġà familjari ma 'kunċetti bħas-serialization se jitgħallmu japplikaw il-ħiliet tagħhom biex joħolqu applikazzjonijiet Kafka Streams. Il-kodiċi tas-sors tal-ktieb huwa miktub f'Java 8 u jagħmel użu sinifikanti mis-sintassi tal-espressjoni lambda ta 'Java 8, għalhekk li tkun taf kif taħdem mal-funzjonijiet lambda (anke f'lingwa ta' programmar oħra) tkun utli.

Silta. 5.3. Operazzjonijiet ta' aggregazzjoni u windowing

F'din it-taqsima, se nkomplu nesploraw l-aktar partijiet promettenti ta' Kafka Streams. S'issa koprejna l-aspetti li ġejjin ta 'Kafka Streams:

• il-ħolqien ta' topoloġija tal-ipproċessar;
• użu tal-istat fl-applikazzjonijiet ta 'streaming;
• it-twettiq ta' konnessjonijiet ta' fluss ta' data;
• differenzi bejn il-flussi tal-avvenimenti (KStream) u l-flussi tal-aġġornament (KTable).

Fl-eżempji li ġejjin se nġibu dawn l-elementi kollha flimkien. Int ser titgħallem ukoll dwar it-twieqi, karatteristika oħra kbira ta 'applikazzjonijiet ta' streaming. L-ewwel eżempju tagħna se jkun aggregazzjoni sempliċi.

5.3.1. Aggregazzjoni ta' bejgħ ta' stokk skond is-settur ta' l-industrija

L-aggregazzjoni u r-raggruppament huma għodod vitali meta taħdem ma 'streaming data. L-eżami tar-rekords individwali hekk kif jiġu riċevuti huwa spiss insuffiċjenti. Biex tiġi estratta informazzjoni addizzjonali mid-dejta, huwa meħtieġ li tiġborhom u tgħaqqadhom.

F'dan l-eżempju, int ser tilbes il-kostum ta 'negozjant ta' ġurnata li jeħtieġ li jsegwi l-volum tal-bejgħ ta 'ħażniet ta' kumpaniji f'diversi industriji. Speċifikament, inti interessat fil-ħames kumpaniji bl-akbar bejgħ ta 'sehem f'kull industrija.

Tali aggregazzjoni teħtieġ id-diversi passi li ġejjin biex tittraduċi d-dejta fil-forma mixtieqa (taħdit f'termini ġenerali).

1. Oħloq sors ibbażat fuq is-suġġett li jippubblika informazzjoni mhux maħduma dwar il-kummerċ tal-istokk. Ikollna nimmappjaw oġġett tat-tip StockTransaction għal oġġett tat-tip ShareVolume. Il-punt huwa li l-oġġett StockTransaction fih metadejta tal-bejgħ, iżda għandna bżonn biss dejta dwar in-numru ta 'ishma li qed jinbiegħu.
2. Grupp ShareData tal-volum skont is-simbolu tal-istokk. Ladarba tkun miġbura skont is-simbolu, tista' tikkollassa din id-dejta f'totali sekondarji tal-volumi tal-bejgħ tal-istokk. Ta 'min jinnota li l-metodu KStream.groupBy jirritorna istanza tat-tip KGroupedStream. U tista 'tikseb istanza ta' KTable billi tkompli ssejjaħ il-metodu KGroupedStream.reduce.

X'inhu l-interface KGroupedStream

Il-metodi KStream.groupBy u KStream.groupByKey jirritornaw istanza ta' KGroupedStream. KGroupedStream hija rappreżentazzjoni intermedja ta 'fluss ta' avvenimenti wara raggruppament minn ċwievet. Mhuwa xejn maħsub għal xogħol dirett miegħu. Minflok, KGroupedStream jintuża għal operazzjonijiet ta 'aggregazzjoni, li dejjem jirriżultaw f'KTable. U peress li r-riżultat tal-operazzjonijiet ta 'aggregazzjoni huwa KTable u jużaw maħżen tal-istat, huwa possibbli li mhux l-aġġornamenti kollha bħala riżultat jintbagħtu aktar 'l isfel fil-pipeline.

Il-metodu KTable.groupBy jirritorna KGroupedTable simili - rappreżentazzjoni intermedja tal-fluss ta' aġġornamenti, miġbura mill-ġdid b'ċavetta.

Ejja nieħdu pawża qasira u nħarsu lejn Fig. 5.9, li turi dak li ksibna. Din it-topoloġija diġà għandha tkun familjari ħafna għalik.

Ejja issa nħarsu lejn il-kodiċi għal din it-topoloġija (tista 'tinstab fil-fajl src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Elenku 5.2).

Il-kodiċi mogħti huwa distint mill-qosor tiegħu u l-volum kbir ta 'azzjonijiet imwettqa f'diversi linji. Tista 'tinnota xi ħaġa ġdida fl-ewwel parametru tal-metodu builder.stream: valur tat-tip enum AutoOffsetReset.EARLIEST (hemm ukoll LATEST), issettjat bl-użu tal-metodu Consumed.withOffsetResetPolicy. Dan it-tip ta' enumerazzjoni jista' jintuża biex jispeċifika strateġija ta' reset ta' kumpens għal kull KStream jew KTable u jieħu preċedenza fuq l-għażla ta' reset ta' kumpens mill-konfigurazzjoni.

GroupByKey u GroupBy

L-interface tal-KStream għandha żewġ metodi biex tiġbor ir-rekords: GroupByKey u GroupBy. It-tnejn jirritornaw KGroupedTable, għalhekk tista' tkun qed tistaqsi x'inhi d-differenza bejniethom u meta tuża liema waħda?

Il-metodu GroupByKey jintuża meta ċ-ċwievet fil-KStream diġà mhumiex vojta. U l-aktar importanti, il-bandiera "jeħtieġ tqassim mill-ġdid" qatt ma ġiet issettjata.

Il-metodu GroupBy jassumi li bdilt iċ-ċwievet tal-grupp, għalhekk il-bandiera tar-ripartizzjoni hija ssettjata għal vera. It-twettiq ta' tgħaqqid, aggregazzjonijiet, eċċ. wara l-metodu GroupBy jirriżulta f'qsim awtomatiku mill-ġdid.
Sommarju: Kull meta jkun possibbli, għandek tuża GroupByKey aktar milli GroupBy.

Huwa ċar x'jagħmlu l-metodi mapValues ​​u groupBy, allura ejja nagħtu ħarsa lejn il-metodu sum() (li jinsab f'src/main/java/bbejeck/model/ShareVolume.java) (Elenku 5.3).

Il-metodu ShareVolume.sum jirritorna t-total kurrenti tal-volum tal-bejgħ tal-istokk, u r-riżultat tal-katina sħiħa tal-kalkoli huwa oġġett KTable . Issa tifhem ir-rwol li għandu KTable. Meta jaslu oġġetti ShareVolume, l-oġġett KTable korrispondenti jaħżen l-aħħar aġġornament kurrenti. Huwa importanti li wieħed jiftakar li l-aġġornamenti kollha huma riflessi fis-shareVolumeKTable preċedenti, iżda mhux kollha jintbagħtu aktar.

Imbagħad nużaw din il-KTable biex naggregaw (b'numru ta 'ishma nnegozjati) biex naslu għall-ħames kumpaniji bl-ogħla volumi ta' ishma nnegozjati f'kull industrija. L-azzjonijiet tagħna f'dan il-każ se jkunu simili għal dawk għall-ewwel aggregazzjoni.

1. Wettaq operazzjoni oħra groupBy biex tiggruppa oġġetti individwali ShareVolume skont l-industrija.
2. Ibda tiġbor fil-qosor l-oġġetti ShareVolume. Din id-darba l-oġġett ta 'aggregazzjoni huwa kju ta' prijorità ta 'daqs fiss. F'dan il-kju ta' daqs fiss, jinżammu biss il-ħames kumpaniji bl-akbar ammonti ta' ishma mibjugħa.
3. Ippjana l-kjuwijiet mill-paragrafu preċedenti għal valur ta’ string u rritorna l-aqwa ħames ħażniet l-aktar negozjati skont in-numru skont l-industrija.
4. Ikteb ir-riżultati f'forma ta 'sekwenza għas-suġġett.

Fil-Fig. Figura 5.10 turi l-graff tat-topoloġija tal-fluss tad-dejta. Kif tistgħu taraw, it-tieni rawnd ta 'proċessar huwa pjuttost sempliċi.

Issa li għandna fehim ċar tal-istruttura ta 'dan it-tieni rawnd ta' pproċessar, nistgħu nduru għall-kodiċi tas-sors tiegħu (se ssibu fil-fajl src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Elenku 5.4) .

Dan initializer fih varjabbli fixedQueue. Dan huwa oġġett tad-dwana li huwa adapter għal java.util.TreeSet li jintuża biex jintraċċa l-ogħla riżultati N f'ordni dixxendenti tal-ishma nnegozjati.

Diġà rajt is-sejħiet groupBy u mapValues, għalhekk mhux se nidħlu f'dawk (qed insejħu l-metodu KTable.toStream minħabba li l-metodu KTable.print huwa deprecated). Imma għadek ma rajtx il-verżjoni KTable ta' aggregate() , għalhekk se nqattgħu ftit ħin niddiskutu dan.

Kif tiftakar, dak li jagħmel KTable differenti huwa li r-rekords bl-istess ċwievet jitqiesu bħala aġġornamenti. KTable jissostitwixxi l-entrata l-antika b'waħda ġdida. L-aggregazzjoni sseħħ b'mod simili: l-aħħar rekords bl-istess ċavetta huma aggregati. Meta jasal rekord, jiġi miżjud mal-istanza tal-klassi FixedSizePriorityQueue billi juża addder (it-tieni parametru fis-sejħa tal-metodu aggregat), iżda jekk diġà jeżisti rekord ieħor bl-istess ċavetta, allura r-rekord l-antik jitneħħa permezz ta’ subtractor (it-tielet parametru f’ is-sejħa tal-metodu aggregat).

Dan kollu jfisser li l-aggregatur tagħna, FixedSizePriorityQueue, ma jaggregax il-valuri kollha b'ċavetta waħda, iżda jaħżen somma li tiċċaqlaq tal-kwantitajiet tal-N l-aktar tipi ta 'ħażniet negozjati. Kull entrata li tidħol fiha n-numru totali ta' ishma mibjugħa s'issa. KTable jagħtik informazzjoni dwar liema ishma tal-kumpaniji huma bħalissa l-aktar negozjati, mingħajr ma teħtieġ aggregazzjoni kontinwa ta’ kull aġġornament.

Tgħallimna nagħmlu żewġ affarijiet importanti:

• grupp ta 'valuri f'KTable permezz ta' ċavetta komuni;
• twettaq operazzjonijiet utli bħal rollup u aggregazzjoni fuq dawn il-valuri raggruppati.

Li tkun taf kif twettaq dawn l-operazzjonijiet huwa importanti biex tifhem it-tifsira tad-dejta li tiċċaqlaq permezz ta 'applikazzjoni Kafka Streams u tifhem x'informazzjoni ġġorr.

Ġaqqna wkoll xi wħud mill-kunċetti ewlenin diskussi aktar kmieni f'dan il-ktieb. Fil-Kapitolu 4, iddiskutejna kif stat lokali tolleranti għall-ħsarat huwa importanti għal applikazzjoni ta 'streaming. L-ewwel eżempju f’dan il-kapitlu wera għaliex l-istat lokali huwa daqshekk importanti—jippermettilek iżżomm kont ta’ liema informazzjoni diġà rajt. L-aċċess lokali jevita dewmien fin-netwerk, u jagħmel l-applikazzjoni aktar effikaċi u reżistenti għall-iżbalji.

Meta twettaq kwalunkwe operazzjoni ta 'rollup jew aggregazzjoni, trid tispeċifika l-isem tal-maħżen tal-istat. L-operazzjonijiet ta' rollup u aggregazzjoni jirritornaw istanza ta' KTable, u KTable juża ħażna tal-istat biex jissostitwixxi riżultati qodma b'oħrajn ġodda. Kif rajt, mhux l-aġġornamenti kollha jintbagħtu fil-pipeline, u dan huwa importanti għaliex l-operazzjonijiet ta 'aggregazzjoni huma mfassla biex jipproduċu informazzjoni fil-qosor. Jekk ma tapplikax l-istat lokali, KTable jibgħat ir-riżultati kollha ta' aggregazzjoni u ġbir.

Sussegwentement, aħna ser inħarsu lejn it-twettiq ta 'operazzjonijiet bħall-aggregazzjoni f'perjodu speċifiku ta' żmien - l-hekk imsejħa operazzjonijiet ta 'twieqi.

5.3.2. Operazzjonijiet tat-tieqa

Fit-taqsima preċedenti, daħħalna konvoluzzjoni u aggregazzjoni li jiżżerżqu. L-applikazzjoni wettqet roll-up kontinwu tal-bejgħ tal-istokk segwit minn aggregazzjoni tal-ħames ħażniet l-aktar negozjati fil-borża.

Xi drabi tali aggregazzjoni kontinwa u roll-up tar-riżultati huma meħtieġa. U xi kultant ikollok bżonn twettaq operazzjonijiet biss fuq perjodu ta 'żmien partikolari. Pereżempju, ikkalkula kemm saru transazzjonijiet ta' skambju b'ishma ta' kumpanija partikolari fl-aħħar 10 minuti. Jew kemm utenti kklikkjaw fuq banner tar-reklamar ġdid fl-aħħar 15-il minuta. Applikazzjoni tista' twettaq tali operazzjonijiet diversi drabi, iżda b'riżultati li japplikaw biss għal perjodi ta' żmien speċifikati (time windows).

Għadd ta' transazzjonijiet ta' skambju mix-xerrej

Fl-eżempju li jmiss, aħna ser insegwu t-tranżazzjonijiet tal-istokk f'diversi negozjanti—jew organizzazzjonijiet kbar jew finanzjaturi individwali intelliġenti.

Hemm żewġ raġunijiet possibbli għal dan it-traċċar. Waħda minnhom hija l-ħtieġa li tkun taf liema mexxejja tas-suq qed jixtru/ibigħu. Jekk dawn il-plejers kbar u investituri sofistikati jaraw opportunità, jagħmel sens li jsegwu l-istrateġija tagħhom. It-tieni raġuni hija x-xewqa li jinstabu kwalunkwe sinjal possibbli ta 'insider trading illegali. Biex tagħmel dan, ser ikollok bżonn tanalizza l-korrelazzjoni ta 'spikes kbar tal-bejgħ ma' stqarrijiet għall-istampa importanti.

Traċċar bħal dan jikkonsisti fil-passi li ġejjin:

• il-ħolqien ta' fluss għall-qari mis-suġġett tat-tranżazzjonijiet tal-istokk;
• jiġbor ir-rekords deħlin skont l-ID tax-xerrej u s-simbolu tal-istokk. Is-sejħa tal-metodu groupBy jirritorna istanza tal-klassi KGroupedStream;
• Il-metodu KGroupedStream.windowedBy jirritorna fluss tad-dejta limitat għal tieqa tal-ħin, li tippermetti aggregazzjoni bit-twieqi. Skont it-tip ta' tieqa, jintbagħat lura jew TimeWindowedKStream jew SessionWindowedKStream;
• għadd ta' tranżazzjonijiet għall-operazzjoni ta' aggregazzjoni. Il-fluss tad-data windowed jiddetermina jekk rekord partikolari jitqiesx f'dan l-għadd;
• tikteb ir-riżultati għal suġġett jew toħroġhom fuq il-console waqt l-iżvilupp.

It-topoloġija ta 'din l-applikazzjoni hija sempliċi, iżda stampa ċara tagħha tkun ta' għajnuna. Ejja nagħtu ħarsa lejn Fig. 5.11.

Sussegwentement, aħna ser inħarsu lejn il-funzjonalità tal-operazzjonijiet tat-twieqi u l-kodiċi korrispondenti.

Tipi ta' tieqa

Hemm tliet tipi ta’ twieqi f’Kafka Streams:

• sessjoni;
• "taqtigħ" (tumbling);
• jiżżerżqu/qbiż.

Liema waħda tagħżel tiddependi fuq ir-rekwiżiti tan-negozju tiegħek. It-twieqi tat-tidwir u tal-qbiż huma limitati fiż-żmien, filwaqt li t-twieqi tas-sessjoni huma limitati mill-attività tal-utent—it-tul tas-sessjoni(jiet) huwa ddeterminat biss minn kemm hu attiv l-utent. Il-ħaġa prinċipali li għandek tiftakar hija li t-tipi kollha ta 'tieqa huma bbażati fuq it-timbri tad-data/ħin tal-entrati, mhux il-ħin tas-sistema.

Sussegwentement, nimplimentaw it-topoloġija tagħna ma 'kull wieħed mit-tipi ta' tieqa. Il-kodiċi komplut se jingħata biss fl-ewwel eżempju għal tipi oħra ta 'twieqi xejn mhu se jinbidel ħlief it-tip ta' operazzjoni tat-tieqa.

Twieqi tas-sessjoni

It-twieqi tas-sessjoni huma differenti ħafna mit-tipi l-oħra kollha ta' twieqi. Huma limitati mhux tant biż-żmien daqs kemm mill-attività tal-utent (jew l-attività tal-entità li tixtieq issegwi). It-twieqi tas-sessjoni huma delimitati minn perjodi ta' inattività.

Il-Figura 5.12 turi l-kunċett tat-twieqi tas-sessjoni. Is-sessjoni iżgħar se jingħaqad mas-sessjoni fuq ix-xellug tagħha. U s-sessjoni fuq il-lemin se tkun separata għaliex issegwi perjodu twil ta 'inattività. It-twieqi tas-sessjoni huma bbażati fuq l-attività tal-utent, iżda uża timbri tad-data/ħin mill-entrati biex tiddetermina għal liema sessjoni tappartjeni l-entrata.

Uża twieqi tas-sessjoni biex issegwi t-tranżazzjonijiet tal-istokk

Ejja nużaw twieqi tas-sessjoni biex naqbdu informazzjoni dwar tranżazzjonijiet ta 'skambju. L-implimentazzjoni tat-twieqi tas-sessjoni tidher fil-Listing 5.5 (li tista' tinstab f'src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).

Diġà rajt ħafna mill-operazzjonijiet f'din it-topoloġija, għalhekk m'hemmx għalfejn terġa' tħares lejhom hawn. Iżda hawn ukoll diversi elementi ġodda, li issa ser niddiskutu.

Kwalunkwe operazzjoni groupBy tipikament twettaq xi tip ta 'operazzjoni ta' aggregazzjoni (aggregazzjoni, rollup, jew għadd). Tista' twettaq jew aggregazzjoni kumulattiva b'total kurrenti, jew aggregazzjoni ta' tieqa, li tqis ir-rekords f'tieqa ta' żmien speċifikata.

Il-kodiċi fil-Listing 5.5 jgħodd in-numru ta 'tranżazzjonijiet fi ħdan it-twieqi tas-sessjoni. Fil-Fig. 5.13 dawn l-azzjonijiet huma analizzati pass pass.

Billi ċċempel windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) noħolqu tieqa ta' sessjoni b'intervall ta' inattività ta' 20 sekonda u intervall ta' persistenza ta' 15-il minuta. Intervall inattiv ta' 20 sekonda jfisser li l-applikazzjoni se tinkludi kwalunkwe dħul li jasal fi żmien 20 sekonda mit-tmiem jew il-bidu tas-sessjoni attwali fis-sessjoni attwali (attiva).

Sussegwentement, aħna nispeċifikaw liema operazzjoni ta 'aggregazzjoni trid titwettaq fit-tieqa tas-sessjoni - f'dan il-każ, għadd. Jekk dħul li jkun dieħel taqa 'barra mit-tieqa ta' inattività (kull naħa tat-timbru tad-data/ħin), l-applikazzjoni toħloq sessjoni ġdida. L-intervall ta' żamma jfisser iż-żamma ta' sessjoni għal ċertu ammont ta' żmien u jippermetti dejta tard li testendi lil hinn mill-perjodu ta' inattività tas-sessjoni iżda xorta tista' tiġi mehmuża. Barra minn hekk, il-bidu u t-tmiem tas-sessjoni l-ġdida li tirriżulta mill-amalgamazzjoni jikkorrispondu mat-timbru tad-data/ħin l-aktar kmieni u l-aħħar.

Ejja nħarsu lejn ftit entrati mill-metodu tal-għadd biex naraw kif jaħdmu s-sessjonijiet (Tabella 5.1).

Meta jaslu r-rekords, infittxu sessjonijiet eżistenti bl-istess ċavetta, ħin tat-tmiem inqas mit-timbru tad-data/ħin attwali - intervall ta 'inattività, u ħin tal-bidu akbar mit-timbru tad-data/ħin attwali + intervall ta' inattività. B'kont meħud ta' dan, erba' daħliet mit-tabella. 5.1 huma magħquda f'sessjoni waħda kif ġej.

1. Ir-rekord 1 jasal l-ewwel, għalhekk il-ħin tal-bidu huwa ugwali għall-ħin tat-tmiem u huwa 00:00:00.

2. Sussegwentement, tasal id-dħul 2, u nfittxu sessjonijiet li jispiċċaw mhux qabel 23:59:55 u jibdew mhux aktar tard minn 00:00:35. Insibu rekord 1 u ngħaqqdu s-sessjonijiet 1 u 2. Nieħdu l-ħin tal-bidu tas-sessjoni 1 (qabel) u l-ħin tat-tmiem tas-sessjoni 2 (aktar tard), sabiex is-sessjoni l-ġdida tagħna tibda f’00:00:00 u tispiċċa f’00: 00:15.

3. Jasal ir-rekord 3, infittxu sessjonijiet bejn 00:00:30 u 00:01:10 u ma nsibux. Żid it-tieni sessjoni għaċ-ċavetta 123-345-654,FFBE, li tibda u tispiċċa f'00:00:50.

4. Jasal ir-rekord 4 u qed infittxu sessjonijiet bejn 23:59:45 u 00:00:25. Din id-darba jinstabu ż-żewġ sessjonijiet 1 u 2 It-tliet sessjonijiet huma magħquda f'waħda, b'ħin tal-bidu ta' 00:00:00 u ħin tat-tmiem ta' 00:00:15.

Minn dak deskritt f'din it-taqsima, ta 'min jiftakar l-sfumaturi importanti li ġejjin:

• is-sessjonijiet mhumiex twieqi ta' daqs fiss. It-tul ta' sessjoni huwa determinat mill-attività f'perjodu ta' żmien partikolari;
• It-timbri tad-data/ħin fid-dejta jiddeterminaw jekk l-avveniment jaqax f'sessjoni eżistenti jew matul perjodu inattiv.

Sussegwentement niddiskutu t-tip ta 'tieqa li jmiss - twieqi "taqlib".

Twieqi "Tumbling".

Tumbling windows jaqbdu avvenimenti li jaqgħu f'ċertu perjodu ta 'żmien. Immaġina li għandek bżonn taqbad it-tranżazzjonijiet kollha tal-istokk ta 'ċerta kumpanija kull 20 sekonda, sabiex tiġbor l-avvenimenti kollha matul dak il-perjodu ta' żmien. Fl-aħħar tal-intervall ta '20 sekonda, it-tieqa tinqaleb u timxi għal intervall ġdid ta' osservazzjoni ta '20 sekonda. Il-Figura 5.14 turi din is-sitwazzjoni.

Kif tistgħu taraw, l-avvenimenti kollha riċevuti fl-aħħar 20 sekonda huma inklużi fit-tieqa. Fl-aħħar ta 'dan il-perjodu ta' żmien, tinħoloq tieqa ġdida.

Lista 5.6 turi kodiċi li juri l-użu tat-twieqi tat-tidwir biex jinqabdu tranżazzjonijiet tal-istokk kull 20 sekonda (li jinstabu f'src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

B'din il-bidla żgħira għas-sejħa tal-metodu TimeWindows.of, tista 'tuża tieqa ta' tumbling. Dan l-eżempju ma jsejjaħx il-metodu sakemm(), għalhekk se jintuża l-intervall ta 'żamma default ta' 24 siegħa.

Fl-aħħarnett, wasal iż-żmien li timxi fuq l-aħħar mill-għażliet tat-tieqa - twieqi "qbiż".

Twieqi li jiżżerżqu ("jaqbżu")

Twieqi li jiżżerżqu/qbiż huma simili għal twieqi li jduru, iżda b'differenza żgħira. Twieqi li jiżżerżqu ma jistennewx sa tmiem l-intervall tal-ħin qabel ma joħolqu tieqa ġdida biex tipproċessa avvenimenti riċenti. Jibdew kalkoli ġodda wara intervall ta' stennija inqas mit-tul tat-tieqa.

Biex turi d-differenzi bejn it-twieqi li jduru u li jaqbżu, ejja nerġgħu lura għall-eżempju tal-għadd tat-tranżazzjonijiet tal-borża. L-għan tagħna għadu li ngħoddu n-numru ta 'tranżazzjonijiet, iżda ma rridux nistennew l-ammont kollu ta' ħin qabel naġġornaw il-counter. Minflok, aħna se naġġornaw il-counter f'intervalli iqsar. Pereżempju, xorta se ngħoddu n-numru ta 'tranżazzjonijiet kull 20 sekonda, iżda naġġornaw il-counter kull 5 sekondi, kif muri fil-Fig. 5.15. F'dan il-każ, nispiċċaw bi tliet twieqi tar-riżultati b'dejta li tikkoinċidi.

Lista 5.7 turi l-kodiċi għad-definizzjoni ta 'twieqi li jiżżerżqu (li jinstabu f'src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Tieqa tat-tidwir tista 'tiġi kkonvertita għal tieqa tal-qbiż billi żżid sejħa għall-metodu advanceBy(). Fl-eżempju muri, l-intervall tal-iffrankar huwa ta '15-il minuta.

F'din it-taqsima rajt kif tillimita r-riżultati tal-aggregazzjoni għal twieqi tal-ħin. B'mod partikolari, nixtieq li tiftakar it-tliet affarijiet li ġejjin minn din it-taqsima:

• id-daqs tat-twieqi tas-sessjoni huwa limitat mhux bil-perjodu ta 'żmien, iżda mill-attività tal-utent;
• twieqi “taqlib” jipprovdu ħarsa ġenerali lejn l-avvenimenti f’perjodu ta’ żmien partikolari;
• It-tul tat-twieqi tal-qbiż huwa fiss, iżda huma aġġornati ta' spiss u jista' jkun fihom entrati li jikkoinċidu fit-twieqi kollha.

Sussegwentement, nitgħallmu kif nikkonverti KTable lura għal KStream għal konnessjoni.

5.3.3. Konnessjoni ta' oġġetti KStream u KTable

Fil-Kapitolu 4, iddiskutejna l-konnessjoni ta 'żewġ oġġetti KStream. Issa rridu nitgħallmu kif nikkonnettjaw KTable u KStream. Dan jista 'jkun meħtieġ għar-raġuni sempliċi li ġejja. KStream huwa nixxiegħa ta 'rekords, u KTable huwa nixxiegħa ta' aġġornamenti ta 'rekords, iżda xi drabi tista' tkun trid iżżid kuntest addizzjonali mal-fluss tar-rekords billi tuża aġġornamenti mill-KTable.

Ejja nieħdu dejta dwar in-numru ta 'tranżazzjonijiet tal-borża u ngħaqqduhom ma' aħbarijiet tal-borża għall-industriji rilevanti. Hawn dak li għandek bżonn tagħmel biex tikseb dan minħabba l-kodiċi li diġà għandek.

1. Ikkonverti oġġett KTable b'dejta dwar in-numru ta 'tranżazzjonijiet tal-istokk f'KStream, segwita billi tissostitwixxi ċ-ċavetta biċ-ċavetta li tindika s-settur tal-industrija li jikkorrispondi għal dan is-simbolu tal-istokk.
2. Oħloq oġġett KTable li jaqra data minn suġġett b'aħbarijiet tal-borża. Din il-KTable ġdida se tkun kategorizzata skont is-settur tal-industrija.
3. Qabbad l-aġġornamenti tal-aħbarijiet b'informazzjoni dwar in-numru ta' tranżazzjonijiet tal-borża skont is-settur tal-industrija.

Issa ejja naraw kif nimplimentaw dan il-pjan ta' azzjoni.

Ikkonverti KTable għal KStream

Biex tikkonverti KTable għal KStream għandek bżonn tagħmel dan li ġej.

1. Sejħa l-metodu KTable.toStream().
2. Billi ssejjaħ il-metodu KStream.map, ibdel iċ-ċavetta bl-isem tal-industrija, u mbagħad irkupra l-oġġett TransactionSummary mill-istanza Windowed.

Aħna se nikkatina dawn l-operazzjonijiet flimkien kif ġej (il-kodiċi jista 'jinstab fil-fajl src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Elenku 5.8).

Minħabba li qed inwettqu operazzjoni KStream.map, l-istanza KStream ritornata terġa 'tiġi diviżjonata awtomatikament meta tintuża f'konnessjoni.

Lestejna l-proċess ta 'konverżjoni, imbagħad irridu noħolqu oġġett KTable għall-qari tal-aħbarijiet tal-istokk.

Ħolqien ta' KTable għall-aħbarijiet tal-istokk

Fortunatament, il-ħolqien ta' oġġett KTable jieħu biss linja waħda ta' kodiċi (il-kodiċi jista' jinstab f'src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Elenku 5.9).

Ta 'min jinnota li l-ebda oġġett Serde ma huwa meħtieġ li jiġi speċifikat, peress li string Serdes jintużaw fis-settings. Ukoll, bl-użu ta 'l-enumerazzjoni KMIENI, it-tabella hija mimlija bir-rekords fil-bidu nett.

Issa nistgħu ngħaddu għall-aħħar pass - konnessjoni.

Tqabbad aġġornamenti tal-aħbarijiet mad-dejta tal-għadd tat-tranżazzjonijiet

Il-ħolqien ta 'konnessjoni mhuwiex diffiċli. Aħna se nużaw left join f'każ li ma jkunx hemm aħbarijiet dwar l-istokk għall-industrija rilevanti (il-kodiċi meħtieġ jista 'jinstab fil-fajl src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Elenku 5.10).

Dan l-operatur leftJoin huwa pjuttost sempliċi. B'differenza mill-joinjiet fil-Kapitolu 4, il-metodu JoinWindow ma jintużax għaliex meta twettaq KStream-KTable join, hemm entrata waħda biss fil-KTable għal kull ċavetta. Konnessjoni bħal din mhix limitata fiż-żmien: ir-rekord huwa jew fil-KTable jew assenti. Il-konklużjoni ewlenija: bl-użu ta 'oġġetti KTable tista' tarrikkixxi KStream b'dejta ta 'referenza aġġornata inqas frekwenti.

Issa se nħarsu lejn mod aktar effiċjenti biex tarrikkixxi l-avvenimenti minn KStream.

5.3.4. Oġġetti GlobalKTable

Kif tistgħu taraw, hemm bżonn li tarrikkixxi l-flussi tal-avvenimenti jew li żżid il-kuntest magħhom. Fil-Kapitolu 4 rajt il-konnessjonijiet bejn żewġ oġġetti KStream, u fit-taqsima preċedenti rajt il-konnessjoni bejn KStream u KTable. F'dawn il-każijiet kollha, huwa meħtieġ li terġa 'taqsam il-fluss tad-dejta meta jiġu mmappjati ċ-ċwievet għal tip jew valur ġdid. Xi drabi t-tqassim mill-ġdid isir b'mod espliċitu, u xi drabi Kafka Streams jagħmel dan awtomatikament. Il-qsim mill-ġdid huwa meħtieġ minħabba li ċ-ċwievet inbidlu u r-rekords iridu jispiċċaw f'taqsimiet ġodda, inkella l-konnessjoni tkun impossibbli (dan ġie diskuss fil-Kapitolu 4, fit-taqsima "Dejta mill-ġdid" fis-subtaqsima 4.2.4).

It-tqassim mill-ġdid għandu spiża

It-tqassim mill-ġdid jeħtieġ spejjeż - spejjeż ta 'riżorsi addizzjonali għall-ħolqien ta' suġġetti intermedji, il-ħażna tad-dejta duplikata f'suġġett ieħor; tfisser ukoll aktar latenza minħabba kitba u qari minn dan is-suġġett. Barra minn hekk, jekk għandek bżonn tingħaqad f'aktar minn aspett jew dimensjoni waħda, trid tikkatina l-joints, timmappa r-rekords b'ċwievet ġodda, u terġa 'tmexxi l-proċess ta' tqassim mill-ġdid.

Konnessjoni ma' settijiet ta' data iżgħar

F'xi każijiet, il-volum ta 'data ta' referenza li għandha tiġi konnessa huwa relattivament żgħir, għalhekk kopji kompluti tagħha jistgħu faċilment joqogħdu lokalment fuq kull nodu. Għal sitwazzjonijiet bħal dawn, Kafka Streams jipprovdi l-klassi GlobalKTable.

L-istanzi tal-GlobalKTable huma uniċi minħabba li l-applikazzjoni tirreplika d-dejta kollha għal kull wieħed min-nodi. U peress li d-dejta kollha hija preżenti fuq kull node, m'hemmx għalfejn tinqasam il-fluss tal-avvenimenti b'ċavetta tad-dejta ta 'referenza sabiex tkun disponibbli għall-ħitan kollha. Tista 'wkoll tagħmel konnessjonijiet mingħajr ċavetta billi tuża oġġetti GlobalKTable. Ejja mmorru lura għal wieħed mill-eżempji preċedenti biex nuru din il-karatteristika.

Konnessjoni ta' oġġetti KStream ma' oġġetti GlobalKTable

Fis-subsezzjoni 5.3.2, għamilna aggregazzjoni ta 'tieqa ta' transazzjonijiet ta 'skambju mix-xerrejja. Ir-riżultati ta 'din l-aggregazzjoni dehru xi ħaġa bħal din:

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

Filwaqt li dawn ir-riżultati servew l-iskop, kien ikun aktar utli li kieku l-isem tal-klijent u l-isem sħiħ tal-kumpanija kienu jintwerew ukoll. Biex iżżid l-isem tal-klijent u l-isem tal-kumpanija, tista 'tagħmel tingħaqad normali, iżda ser ikollok bżonn tagħmel żewġ mappings ewlenin u tqassim mill-ġdid. Bil-GlobalKTable tista 'tevita l-ispiża ta' operazzjonijiet bħal dawn.

Biex tagħmel dan, aħna ser nużaw l-oġġett countStream mil-Listing 5.11 (il-kodiċi korrispondenti jista 'jinstab f'src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) u qabbadha ma' żewġ oġġetti GlobalKTable.

Dan diġà ddiskutejna qabel, għalhekk mhux se nirrepetih. Imma ninnota li l-kodiċi fil-funzjoni toStream().map huwa estratt f'oġġett ta 'funzjoni minflok espressjoni lambda inline għall-fini ta' leġġibbiltà.

Il-pass li jmiss huwa li tiddikjara żewġ każijiet ta 'GlobalKTable (il-kodiċi muri jista' jinstab fil-fajl src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (Elenku 5.12).

Jekk jogħġbok innota li l-ismijiet tas-suġġetti huma deskritti billi jintużaw tipi enumerati.

Issa li għandna l-komponenti kollha lesti, kulma jibqa 'huwa li tikteb il-kodiċi għall-konnessjoni (li tista' tinstab fil-fajl src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (Elenku 5.13).

Għalkemm hemm żewġ tingħaqad f'dan il-kodiċi, huma kkatinati minħabba li l-ebda riżultat tagħhom ma jintuża separatament. Ir-riżultati jintwerew fl-aħħar tal-operazzjoni kollha.

Meta tmexxi l-operazzjoni ta' tingħaqad ta' hawn fuq, ikollok riżultati bħal dawn:

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

L-essenza ma nbidlitx, iżda dawn ir-riżultati jidhru aktar ċari.

Jekk tgħodd sal-Kapitolu 4, diġà rajt diversi tipi ta' konnessjonijiet fl-azzjoni. Huma elenkati fit-tabella. 5.2. Din it-tabella tirrifletti l-kapaċitajiet ta' konnettività mill-verżjoni 1.0.0 ta' Kafka Streams; Xi ħaġa tista' tinbidel fir-rilaxxi futuri.

Biex tikkonkludi l-affarijiet, ejja nġabru l-affarijiet bażiċi: tista 'tqabbad flussi ta' avvenimenti (KStream) u taġġorna flussi (KTable) billi tuża l-istat lokali. Inkella, jekk id-daqs tad-dejta ta 'referenza ma jkunx kbir wisq, tista' tuża l-oġġett GlobalKTable. GlobalKTables jirreplika l-diviżorji kollha għal kull nodu tal-applikazzjoni Kafka Streams, u jiżgura li d-dejta kollha tkun disponibbli irrispettivament minn liema partizzjoni tikkorrispondi għaliha ċ-ċavetta.

Sussegwentement se naraw il-karatteristika Kafka Streams, li bis-saħħa tagħha nistgħu nosservaw bidliet fl-istat mingħajr ma nikkunsmaw dejta minn suġġett Kafka.

5.3.5. Stat mistoqsija

Aħna diġà wettaqna diversi operazzjonijiet li jinvolvu l-istat u dejjem joħorġu r-riżultati lill-console (għal skopijiet ta 'żvilupp) jew iktebhom għal suġġett (għal skopijiet ta' produzzjoni). Meta tikteb ir-riżultati għal suġġett, trid tuża konsumatur Kafka biex tarahom.

Il-qari tad-dejta minn dawn is-suġġetti jista’ jitqies bħala tip ta’ fehmiet materjali. Għall-iskopijiet tagħna, nistgħu nużaw id-definizzjoni ta' veduta mmaterjalizzata mill-Wikipedija: “...oġġett ta' database fiżika li fih ir-riżultati ta' mistoqsija. Pereżempju, tista’ tkun kopja lokali ta’ data remota, jew subsett tar-ringieli u/jew kolonni ta’ tabella jew riżultati ta’ tingħaqad, jew tabella sommarja miksuba permezz ta’ aggregazzjoni” (https://en.wikipedia.org/wiki /Materjalized_view).

Kafka Streams jippermettilek ukoll li tmexxi mistoqsijiet interattivi fuq ħwienet tal-istat, li jippermettulek taqra direttament dawn l-opinjonijiet immaterjalizzati. Huwa importanti li wieħed jinnota li l-mistoqsija lill-maħżen tal-istat hija operazzjoni li tinqara biss. Dan jiżgura li ma jkollokx għalfejn tinkwieta dwar li aċċidentalment tagħmel l-istat inkonsistenti waqt li l-applikazzjoni tiegħek tkun qed tipproċessa d-dejta.

Il-ħila li tinvestiga direttament il-ħwienet tal-istat hija importanti. Dan ifisser li inti tista 'toħloq applikazzjonijiet dashboard mingħajr ma jkollok l-ewwel iġġib id-dejta mill-konsumatur Kafka. Iżid ukoll l-effiċjenza tal-applikazzjoni, minħabba l-fatt li m'hemmx bżonn li terġa 'tikteb id-data:

• grazzi għal-lokalità tad-dejta, jistgħu jiġu aċċessati malajr;
• id-duplikazzjoni tad-data hija eliminata, peress li mhix miktuba fuq ħażna esterna.

Il-ħaġa ewlenija li rrid li tiftakar hija li tista 'titlob l-istat direttament minn ġewwa l-applikazzjoni tiegħek. L-opportunitajiet li dan jagħtik ma jistgħux jiġu eżaġerati. Minflok ma tikkonsma dejta minn Kafka u taħżen ir-rekords f'database għall-applikazzjoni, tista 'tfittex ħwienet tal-istat bl-istess riżultat. Mistoqsijiet diretti lill-ħwienet tal-istat ifissru inqas kodiċi (l-ebda konsumatur) u inqas softwer (l-ebda ħtieġa għal tabella tad-database biex taħżen ir-riżultati).

F'dan il-kapitolu koprejna pjuttost ftit, għalhekk aħna nħallu d-diskussjoni tagħna dwar mistoqsijiet interattivi kontra ħwienet tal-istat għalissa. Imma tinkwetax: fil-Kapitolu 9, aħna ser noħolqu applikazzjoni dashboard sempliċi b'mistoqsijiet interattivi. Se tuża wħud mill-eżempji minn dan il-kapitoli u dawk preċedenti biex turi mistoqsijiet interattivi u kif tista 'żżidhom mal-applikazzjonijiet Kafka Streams.

Sommarju

• L-oġġetti KStream jirrappreżentaw flussi ta' avvenimenti, komparabbli ma' inserzjonijiet f'database. L-oġġetti KTable jirrappreżentaw flussi ta' aġġornamenti, aktar bħal aġġornamenti għal database. Id-daqs tal-oġġett KTable ma jikberx, ir-rekords qodma jiġu sostitwiti b'oħrajn ġodda.
• L-oġġetti KTable huma meħtieġa għal operazzjonijiet ta' aggregazzjoni.
• Bl-użu ta 'operazzjonijiet ta' windowing, tista 'taqsam id-dejta aggregata f'bramel tal-ħin.
• Grazzi għall-oġġetti GlobalKTable, tista 'taċċessa dejta ta' referenza kullimkien fl-applikazzjoni, irrispettivament mill-qsim.
• Konnessjonijiet bejn oġġetti KStream, KTable u GlobalKTable huma possibbli.

S'issa, konna ffukajna fuq il-bini ta' applikazzjonijiet Kafka Streams bl-użu ta' KStream DSL ta' livell għoli. Għalkemm l-approċċ ta 'livell għoli jippermettilek toħloq programmi puliti u konċiżi, l-użu tiegħu jirrappreżenta kompromess. Li taħdem ma 'DSL KStream tfisser li żżid il-konċiżjoni tal-kodiċi tiegħek billi tnaqqas il-grad ta' kontroll. Fil-kapitolu li jmiss, aħna ser inħarsu lejn l-API ta 'livell baxx ta' handler node u nippruvaw kompromessi oħra. Il-programmi se jkunu itwal milli kienu qabel, iżda se nkunu nistgħu noħolqu kważi kull node ta 'handler li jista' jkun hemm bżonn.

→ Aktar dettalji dwar il-ktieb jistgħu jinstabu fuq websajt tal-pubblikatur

→ Għal Habrozhiteli 25% skont bl-użu tal-kupun - Kafka Streams

→ Mal-ħlas għall-verżjoni stampata tal-ktieb, jintbagħat ktieb elettroniku bl-e-mail.

Sors: www.habr.com