🥇 Libro nga "Kafka Streams in Action. Mga aplikasyon ug microservices para sa real-time nga trabaho"

Kumusta, mga residente sa Khabro! Kini nga libro angay alang sa bisan kinsa nga developer nga gusto nga makasabut sa pagproseso sa thread. Ang pagsabut sa gipang-apod-apod nga programa makatabang kanimo nga mas masabtan ang Kafka ug Kafka Streams. Nindot nga mahibal-an ang balangkas sa Kafka mismo, apan dili kini kinahanglan: Isulti ko kanimo ang tanan nga imong kinahanglan. Ang mga eksperyensiyado nga mga developer ug mga bag-o sa Kafka magkat-on unsaon paghimo og makapaikag nga mga aplikasyon sa pagproseso sa sapa gamit ang librarya sa Kafka Streams niini nga libro. Ang mga intermediate ug advanced Java developers nga pamilyar na sa mga konsepto sama sa serialization makakat-on sa paggamit sa ilang mga kahanas sa paghimo sa mga aplikasyon sa Kafka Streams. Ang source code sa libro gisulat sa Java 8 ug naghimo sa mahinungdanong paggamit sa Java 8 lambda expression syntax, busa ang pagkahibalo unsaon pagtrabaho sa mga function sa lambda (bisan sa laing programming language) magamit.

Kinutlo. 5.3. Mga operasyon sa paghugpong ug pag-windowing

Niini nga seksyon, magpadayon kita sa pag-usisa sa labing maayong mga bahin sa Kafka Streams. Sa pagkakaron nasakpan na namo ang mosunod nga mga aspeto sa Kafka Streams:

paghimo sa usa ka pagproseso topology;
paggamit sa estado sa streaming nga mga aplikasyon;
pagbuhat sa data stream koneksyon;
mga kalainan tali sa mga sapa sa panghitabo (KStream) ug mga sapa sa pag-update (KTable).

Sa mosunod nga mga pananglitan atong dad-on ang tanan niini nga mga elemento. Makakat-on ka usab bahin sa pag-windowing, usa pa ka maayo nga bahin sa streaming nga mga aplikasyon. Ang atong unang pananglitan mao ang usa ka yano nga aggregation.

5.3.1. Paghiusa sa pagbaligya sa stock sa sektor sa industriya

Ang pagtipon ug paggrupo hinungdanon nga mga himan kung nagtrabaho kauban ang streaming data. Ang pagsusi sa indibidwal nga mga rekord ingon nga kini nadawat kasagaran dili igo. Aron makuha ang dugang nga impormasyon gikan sa datos, gikinahanglan ang paggrupo ug paghiusa niini.

Sa kini nga pananglitan, imong isul-ob ang costume sa usa ka day trader nga kinahanglan nga masubay ang gidaghanon sa pagbaligya sa mga stock sa mga kompanya sa daghang mga industriya. Sa partikular, interesado ka sa lima ka mga kompanya nga adunay pinakadako nga pagbaligya sa bahin sa matag industriya.

Ang ingon nga panagsama nanginahanglan sa mosunod nga daghang mga lakang aron mahubad ang datos sa gusto nga porma (pagsulti sa kinatibuk-ang termino).

Paghimo usa ka gigikanan nga nakabase sa hilisgutan nga nagpatik sa hilaw nga impormasyon sa trading sa stock. Kinahanglan namon nga mapa ang usa ka butang nga tipo nga StockTransaction sa usa ka butang nga tipo nga ShareVolume. Ang punto mao nga ang butang nga StockTransaction adunay metadata sa pagbaligya, apan kinahanglan ra namon ang datos bahin sa gidaghanon sa mga bahin nga gibaligya.
Grupo nga ShareVolume nga datos pinaagi sa simbolo sa stock. Sa higayon nga ma-grupo pinaagi sa simbolo, mahimo nimong i-collapse kini nga data ngadto sa mga subtotal sa stock sales volume. Angay nga hinumdoman nga ang KStream.groupBy nga pamaagi nagbalik usa ka pananglitan sa tipo nga KGroupedStream. Ug mahimo ka makakuha usa ka KTable nga pananglitan pinaagi sa dugang nga pagtawag sa KGroupedStream.reduce nga pamaagi.

Unsa ang interface sa KGroupedStream

Ang KStream.groupBy ug KStream.groupByKey nga mga pamaagi nagbalik ug instance sa KGroupedStream. Ang KGroupedStream usa ka intermediate nga representasyon sa usa ka stream sa mga panghitabo human sa paggrupo pinaagi sa mga yawe. Dili gyud kini gituyo alang sa direktang pagtrabaho niini. Hinunoa, ang KGroupedStream gigamit alang sa mga operasyon sa aggregation, nga kanunay moresulta sa usa ka KTable. Ug tungod kay ang resulta sa mga operasyon sa aggregation usa ka KTable ug gigamit nila ang usa ka tindahan sa estado, posible nga dili tanan nga mga pag-update ingon usa ka sangputanan gipadala sa ubos sa pipeline.

Ang KTable.groupBy nga pamaagi nagbalik sa susamang KGroupedTable - usa ka intermediate nga representasyon sa stream sa mga update, gi-group pag-usab pinaagi sa yawe.

Mopahulay ta ug kadiyot ug tan-awon ang Fig. 5.9, nga nagpakita kung unsa ang atong nakab-ot. Kini nga topology kinahanglan nga pamilyar kaayo kanimo.

Atong tan-awon karon ang code alang niini nga topology (makita kini sa file src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Listing 5.2).

Ang gihatag nga kodigo gipalahi sa kamubo niini ug sa daghang gidaghanon sa mga aksyon nga gihimo sa daghang linya. Mahimo nimong mamatikdan ang usa ka butang nga bag-o sa unang parameter sa builder.stream nga pamaagi: usa ka bili sa enum type AutoOffsetReset.EARLIEST (adunay usab usa ka LATEST), gibutang gamit ang Consumed.withOffsetResetPolicy nga pamaagi. Kini nga matang sa enumeration mahimong gamiton aron itakda ang offset reset nga estratehiya para sa matag KStream o KTable ug mag-una sa offset reset nga opsyon gikan sa configuration.

GroupByKey ug GroupBy

Ang interface sa KStream adunay duha ka pamaagi sa paggrupo sa mga rekord: GroupByKey ug GroupBy. Ang duha nagbalik sa usa ka KGroupedTable, mao nga tingali nahibulong ka kung unsa ang kalainan tali kanila ug kanus-a gamiton kung hain?

Ang GroupByKey nga pamaagi gigamit kung ang mga yawe sa KStream wala nay sulod. Ug labaw sa tanan, ang bandila nga "nanginahanglan pag-re-partition" wala gyud itakda.

Ang GroupBy nga pamaagi nagtuo nga imong giusab ang mga yawe sa paggrupo, mao nga ang repartition nga bandila gibutang sa tinuod. Ang paghimo sa mga pag-apil, mga aggregation, ug uban pa pagkahuman sa GroupBy nga pamaagi moresulta sa awtomatik nga pagbahinbahin pag-usab.
Sumaryo: Kung mahimo, kinahanglan nimo gamiton ang GroupByKey kaysa sa GroupBy.

Klaro kung unsa ang gibuhat sa mga pamaagi sa mapValues ug groupBy, busa atong tan-awon ang sum() nga pamaagi (makita sa src/main/java/bbejeck/model/ShareVolume.java) (Listing 5.3).

Ang ShareVolume.sum nga pamaagi nagbalik sa nagdagan nga total sa stock sales volume, ug ang resulta sa tibuok nga kutay sa mga kalkulasyon usa ka KTable nga butang . Karon nasabtan na nimo ang papel sa KTable. Kung ang mga butang sa ShareVolume moabut, ang katugbang nga butang nga KTable nagtipig sa labing bag-o nga bag-ong update. Mahinungdanon nga hinumdoman nga ang tanan nga mga pag-update gipakita sa miaging shareVolumeKTable, apan dili tanan gipadala pa.

Sunod, gamit kini nga KTable, among gihugpong (sa gidaghanon sa mga bahin nga gibaligya) aron makaabot sa lima ka mga kompanya nga adunay labing taas nga gidaghanon sa mga bahin nga gibaligya sa matag industriya. Ang among mga aksyon sa kini nga kaso mahimong parehas sa una nga panagsama.

Paghimo og laing grupoBy nga operasyon sa paggrupo sa indibidwal nga ShareVolume nga mga butang sa industriya.
Sugdi ang pag-summarize sa mga butang sa ShareVolume. Niining higayona ang aggregation object usa ka fixed-size priority queue. Niining pirmi nga gidak-on nga pila, ang lima lang ka kompanya nga adunay labing kadaghan nga bahin nga gibaligya ang gipabilin.
Mapa ang mga pila gikan sa miaging paragraph ngadto sa usa ka string value ug ibalik ang nag-unang lima ka pinaka-trade nga stocks pinaagi sa numero sa industriya.
Isulat ang mga resulta sa hugpong nga porma sa hilisgutan.

Sa Fig. Ang Figure 5.10 nagpakita sa data flow topology graph. Sama sa imong nakita, ang ikaduhang hugna sa pagproseso yano ra.

Karon nga kita adunay usa ka tin-aw nga pagsabut sa istruktura niining ikaduhang hugna sa pagproseso, mahimo namong ablihan ang source code niini (imong makita kini sa file src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (Listing 5.4) .

Kini nga initializer adunay usa ka fixedQueue variable. Kini usa ka kostumbre nga butang nga usa ka adaptor alang sa java.util.TreeSet nga gigamit sa pagsubay sa kinatas-ang N nga resulta sa pagkanaog nga han-ay sa mga bahin nga gibaligya.

Nakita na nimo ang groupBy ug mapValues nga mga tawag, busa dili kami moadto sa mga (gitawag namo ang KTable.toStream nga pamaagi tungod kay ang KTable.print nga pamaagi wala na gigamit). Apan wala pa nimo makita ang KTable nga bersyon sa aggregate() pa, mao nga mogahin kami og gamay nga panahon sa paghisgot niana.

Sama sa imong nahinumduman, ang nakapalahi sa KTable mao nga ang mga rekord nga adunay parehas nga mga yawe giisip nga mga update. Gipulihan sa KTable ang daan nga entry og bag-o. Ang aggregation mahitabo sa susamang paagi: ang pinakabag-o nga mga rekord nga adunay parehas nga yawe giipon. Kung moabut ang usa ka rekord, idugang kini sa pananglitan sa klase nga FixedSizePriorityQueue gamit ang usa ka adder (ikaduha nga parameter sa aggregate method call), apan kung adunay lain nga rekord nga adunay parehas nga yawe, nan ang daan nga rekord gikuha gamit ang usa ka subtractor (ikatulo nga parameter sa ang aggregate method call).

Kining tanan nagpasabot nga ang atong aggregator, ang FixedSizePriorityQueue, wala mag-aggregate sa tanang values sa usa ka yawe, apan nagtipig ug naglihok nga kantidad sa gidaghanon sa N nga pinaka-trade nga matang sa stocks. Ang matag umaabot nga entry naglangkob sa kinatibuk-ang gidaghanon sa mga bahin nga gibaligya sa pagkakaron. Hatagan ka sa KTable og kasayuran bahin sa kung unsang mga bahin sa mga kompanya ang karon ang labing gipamaligya, nga wala kinahanglana ang rolling aggregation sa matag update.

Natun-an namon ang paghimo sang duha ka importante nga butang:

mga kantidad sa grupo sa KTable pinaagi sa usa ka komon nga yawe;
paghimo sa mapuslanon nga mga operasyon sama sa rollup ug paghugpong sa mga gigrupo nga kantidad.

Ang pagkahibalo unsaon pagbuhat niini nga mga operasyon importante aron masabtan ang kahulogan sa datos nga nagalihok pinaagi sa aplikasyon sa Kafka Streams ug masabtan kung unsa nga impormasyon ang gidala niini.

Gihiusa usab namo ang pipila sa mga mahinungdanong konsepto nga gihisgutan sa sayo pa niini nga libro. Sa Kapitulo 4, among gihisgutan kung unsa ka importante ang fault-tolerant, lokal nga estado alang sa usa ka streaming nga aplikasyon. Ang unang pananglitan niini nga kapitulo nagpakita kon nganong ang lokal nga estado importante kaayo—kini naghatag kanimo ug abilidad sa pagsubay sa impormasyon nga imong nakita na. Ang lokal nga pag-access naglikay sa mga paglangan sa network, nga naghimo sa aplikasyon nga labi ka pasundayag ug dili makasugakod sa sayup.

Kung maghimo sa bisan unsang rollup o aggregation nga operasyon, kinahanglan nimo nga ipiho ang ngalan sa tindahan sa estado. Ang rollup ug aggregation operations nagbalik ug KTable nga instance, ug ang KTable naggamit sa state storage aron ilisan ang daan nga mga resulta og mga bag-o. Sama sa imong nakita, dili tanan nga mga pag-update gipadala sa pipeline, ug kini hinungdanon tungod kay ang mga operasyon sa aggregation gidesinyo aron makahimo og summary nga impormasyon. Kung dili nimo i-apply ang lokal nga estado, ipasa sa KTable ang tanan nga mga resulta sa aggregation ug rollup.

Sunod, atong tan-awon ang pagbuhat sa mga operasyon sama sa aggregation sulod sa usa ka piho nga yugto sa panahon - gitawag nga windowing operations.

5.3.2. Mga operasyon sa bintana

Sa miaging seksyon, among gipaila ang sliding convolution ug aggregation. Ang aplikasyon nagpahigayon sa usa ka padayon nga roll-up sa stock sales volume, gisundan sa aggregation sa lima ka pinaka-trade stocks sa exchange.

Usahay ang ingon nga padayon nga pagtipon ug pag-roll-up sa mga resulta gikinahanglan. Ug usahay kinahanglan nimo nga himuon ang mga operasyon sa usa ka gihatag nga yugto sa panahon. Pananglitan, kuwentaha kung pila ang mga transaksyon sa pagbinayloay nga gihimo sa mga bahin sa usa ka partikular nga kompanya sa miaging 10 minuto. O pila ka mga tiggamit ang nag-klik sa usa ka bag-ong banner sa advertising sa miaging 15 minuto. Ang usa ka aplikasyon mahimong mohimo sa ingon nga mga operasyon sa daghang mga higayon, apan adunay mga resulta nga magamit lamang sa piho nga mga yugto sa panahon (mga bintana sa oras).

Pag-ihap sa mga transaksyon sa pagbaylo sa pumapalit

Sa sunod nga pananglitan, atong bantayan ang mga transaksyon sa stock sa daghang mga negosyante—mahimo nga dagkong organisasyon o maalamon nga indibidwal nga mga financier.

Adunay duha ka posibleng rason alang niini nga pagsubay. Usa niini mao ang panginahanglan nga mahibal-an kung unsa ang gipalit / gibaligya sa mga lider sa merkado. Kung kining mga dagkong magdudula ug sopistikado nga mga tigpamuhunan makakita og oportunidad, makatarunganon nga sundon ang ilang estratehiya. Ang ikaduhang rason mao ang tinguha nga makit-an ang bisan unsang posibleng mga timailhan sa ilegal nga insider trading. Aron mahimo kini, kinahanglan nimo nga analisahon ang correlation sa dagkong mga spike sa pagpamaligya nga adunay hinungdanon nga mga pagpagawas sa balita.

Ang maong pagsubay naglangkob sa mosunod nga mga lakang:

paghimo og usa ka sapa alang sa pagbasa gikan sa hilisgutan sa stock-transactions;
paggrupo sa umaabot nga mga rekord pinaagi sa buyer ID ug stock simbolo. Ang pagtawag sa grupoBy nga pamaagi nagbalik sa usa ka pananglitan sa klase sa KGroupedStream;
Ang KGroupedStream.windowedBy nga pamaagi nagbalik ug data stream nga limitado sa usa ka time window, nga nagtugot sa windowed aggregation. Depende sa matang sa bintana, ang TimeWindowedKStream o ang SessionWindowedKStream ibalik;
ihap sa transaksyon alang sa aggregation operation. Ang windowed data flow nagtino kung ang usa ka partikular nga rekord gikonsiderar niini nga ihap;
pagsulat sa mga resulta sa usa ka hilisgutan o pag-output niini sa console sa panahon sa pag-uswag.

Ang topology niini nga aplikasyon yano ra, apan ang usa ka tin-aw nga hulagway niini makatabang. Atong tan-awon ang Fig. 5.11.

Sunod, atong tan-awon ang pagpaandar sa mga operasyon sa bintana ug ang katugbang nga code.

Mga tipo sa bintana

Adunay tulo ka matang sa mga bintana sa Kafka Streams:

sesyonal;
"pagtumba";
pag-slide / paglukso.

Hain ang pilion nagdepende sa imong mga kinahanglanon sa negosyo. Ang pag-tumbling ug paglukso nga mga bintana gilimitahan sa panahon, samtang ang mga bintana sa sesyon limitado sa kalihokan sa user—ang gidugayon sa (mga) sesyon gitino lamang kung unsa ka aktibo ang user. Ang panguna nga butang nga hinumdoman mao nga ang tanan nga mga tipo sa bintana gibase sa mga selyo sa petsa / oras sa mga entri, dili ang oras sa sistema.

Sunod, gipatuman namon ang among topology sa matag usa sa mga tipo sa bintana. Ang kompleto nga code ihatag lamang sa unang pananglitan; alang sa ubang mga matang sa windows walay mausab gawas sa matang sa operasyon sa bintana.

Mga bintana sa sesyon

Ang mga bintana sa sesyon lahi kaayo sa tanan nga mga lahi sa mga bintana. Sila limitado dili kaayo sa panahon sama sa kalihokan sa user (o sa kalihokan sa entidad nga gusto nimong sundan). Ang mga bintana sa sesyon gilimitahan sa mga panahon sa pagkadili aktibo.

Ang Figure 5.12 naghulagway sa konsepto sa session windows. Ang mas gamay nga sesyon maghiusa sa sesyon sa wala niini. Ug ang sesyon sa tuo magbulag tungod kay nagsunod kini sa usa ka taas nga panahon sa pagkadili aktibo. Ang mga bintana sa sesyon gibase sa kalihokan sa gumagamit, apan gamita ang mga selyo sa petsa/oras gikan sa mga entri aron mahibal-an kung unsang sesyon ang nahisakop.

Paggamit sa mga bintana sa sesyon aron masubay ang mga transaksyon sa stock

Atong gamiton ang mga bintana sa sesyon aron makuha ang kasayuran bahin sa mga transaksyon sa pagbinayloay. Ang pagpatuman sa session windows gipakita sa Listing 5.5 (nga makita sa src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).

Nakita na nimo ang kadaghanan sa mga operasyon niini nga topology, busa dili na kinahanglan nga tan-awon kini pag-usab dinhi. Apan adunay usab daghang mga bag-ong elemento dinhi, nga atong hisgutan karon.

Bisan unsa nga grupoBy nga operasyon kasagarang nagpahigayon sa usa ka matang sa aggregation nga operasyon (paghiusa, rollup, o pag-ihap). Mahimo nimong buhaton ang bisan unsang cumulative aggregation nga adunay nagdagan nga total, o window aggregation, nga gikonsiderar ang mga rekord sa sulod sa usa ka piho nga window sa oras.

Ang code sa Listing 5.5 nag-ihap sa gidaghanon sa mga transaksyon sulod sa mga bintana sa sesyon. Sa Fig. 5.13 kini nga mga aksyon gianalisa sa matag lakang.

Pinaagi sa pagtawag sa windowedBy(SessionWindows.with(twentySeconds).hangtod(fifteenMinutes)) maghimo mig session window nga adunay inactivity interval nga 20 seconds ug usa ka persistence interval nga 15 minutes. Ang usa ka idle interval nga 20 segundos nagpasabut nga ang aplikasyon maglakip sa bisan unsang entry nga moabut sa sulod sa 20 segundos sa katapusan o pagsugod sa karon nga sesyon sa karon (aktibo) nga sesyon.

Sunod, among gitino kung unsang aggregation nga operasyon ang kinahanglan nga himuon sa window sa sesyon - sa kini nga kaso, ihap. Kung ang umaabot nga entry mahulog sa gawas sa inactivity window (bisan asa nga kilid sa date/time stamp), ang aplikasyon maghimo ug bag-ong session. Ang agwat sa pagpadayon nagpasabut sa pagpadayon sa usa ka sesyon sa usa ka piho nga gidugayon sa oras ug gitugotan ang ulahi nga datos nga molapas sa panahon nga dili aktibo sa sesyon apan mahimo gihapon nga ilakip. Dugang pa, ang pagsugod ug pagtapos sa bag-ong sesyon nga resulta sa paghiusa katumbas sa pinakauna ug pinakabag-o nga selyo sa petsa/oras.

Atong tan-awon ang pipila ka mga entry gikan sa pamaagi sa pag-ihap aron makita kung giunsa ang mga sesyon molihok (Table 5.1).

Kung moabut ang mga rekord, mangita kami sa mga naglungtad nga mga sesyon nga adunay parehas nga yawe, usa ka oras sa pagtapos nga mas gamay kaysa sa karon nga selyo sa petsa/oras - agwat sa dili aktibo, ug usa ka oras sa pagsugod nga labi pa sa karon nga selyo sa petsa/oras + agwat sa dili aktibo. Sa pagkonsiderar niini, upat ka entries gikan sa lamesa. 5.1 gihiusa ngadto sa usa ka sesyon sama sa mosunod.

1. Ang rekord 1 unang moabot, mao nga ang oras sa pagsugod katumbas sa oras sa pagtapos ug 00:00:00.

2. Sunod, ang entry 2 moabot, ug kita mangita sa mga sesyon nga matapos dili sayo kay sa 23:59:55 ug magsugod sa dili molapas sa 00:00:35. Among pangitaon ang rekord 1 ug gihiusa ang mga sesyon 1 ug 2. Among gikuha ang oras sa pagsugod sa sesyon 1 (mas sayo) ug ang oras sa pagtapos sa sesyon 2 (sa ulahi), aron ang among bag-ong sesyon magsugod sa 00:00:00 ug matapos sa 00: 00:15.

3. Pag-abot sa record 3, nangita kami og mga sesyon tali sa 00:00:30 ug 00:01:10 ug wala'y nakit-an. Pagdugang ug ikaduhang sesyon para sa yawe 123-345-654,FFBE, sugod ug matapos sa 00:00:50.

4. Nag-abot ang rekord 4 ug nangita kami og mga sesyon tali sa 23:59:45 ug 00:00:25. Niining higayona makit-an ang duha ka sesyon 1 ug 2. Ang tanan nga tulo ka sesyon gihiusa sa usa, nga adunay oras sa pagsugod sa 00:00:00 ug oras sa pagtapos sa 00:00:15.

Gikan sa gihulagway niini nga seksyon, angay nga hinumdoman ang mosunod nga importante nga mga nuances:

ang mga sesyon dili fixed-size nga mga bintana. Ang gidugayon sa usa ka sesyon gitino pinaagi sa kalihokan sulod sa gihatag nga yugto sa panahon;
Ang mga selyo sa petsa/oras sa datos nagtino kung ang panghitabo naa ba sulod sa usa ka kasamtangan nga sesyon o sa panahon nga walay trabaho.

Sunod atong hisgutan ang sunod nga matang sa bintana - "tumbling" nga mga bintana.

Mga bintana nga "nagkurog".

Ang mga tumbling windows nagkuha sa mga panghitabo nga nahulog sa usa ka piho nga yugto sa panahon. Hunahunaa nga kinahanglan nimo nga makuha ang tanan nga mga transaksyon sa stock sa usa ka kompanya matag 20 segundos, aron imong kolektahon ang tanan nga mga panghitabo sa kana nga yugto sa panahon. Sa pagtapos sa 20-segundos nga agwat, ang bintana moligid ug mobalhin ngadto sa bag-ong 20-segundos nga agwat sa obserbasyon. Ang Figure 5.14 naghulagway niini nga sitwasyon.

Sama sa imong nakita, ang tanan nga mga panghitabo nga nadawat sa katapusan nga 20 segundos gilakip sa bintana. Sa katapusan niini nga yugto sa panahon, usa ka bag-ong bintana ang gihimo.

Ang listahan sa 5.6 nagpakita sa code nga nagpakita sa paggamit sa tumbling windows aron makuha ang mga transaksyon sa stock matag 20 segundos (makita sa src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Uban niining gamay nga pagbag-o sa TimeWindows.of method call, mahimo nimong gamiton ang usa ka tumbling window. Kini nga pananglitan wala magtawag sa until() nga pamaagi, mao nga ang default retention interval nga 24 oras ang gamiton.

Sa katapusan, panahon na nga magpadayon sa katapusan sa mga kapilian sa bintana - "paglukso" nga mga bintana.

Pag-slide ("paglukso") nga mga bintana

Ang mga sliding/hopping nga mga bintana susama sa nangatumba nga mga bintana, apan adunay gamay nga kalainan. Ang mga sliding windows dili maghulat hangtod sa katapusan sa agwat sa oras sa dili pa maghimo usa ka bag-ong bintana aron maproseso ang bag-ong mga panghitabo. Nagsugod sila og bag-ong mga kalkulasyon human sa agwat sa paghulat nga ubos pa sa gidugayon sa bintana.

Sa pag-ilustrar sa mga kalainan tali sa tumbling ug jumping windows, mobalik kita sa pananglitan sa pag-ihap sa mga transaksyon sa stock exchange. Ang among tumong mao ang pag-ihap sa gidaghanon sa mga transaksyon, apan dili kami gusto nga maghulat sa tibuok nga gidugayon sa panahon sa dili pa i-update ang counter. Hinuon, among i-update ang counter sa mas mugbo nga mga agwat. Pananglitan, mag-ihap gihapon kami sa gidaghanon sa mga transaksyon matag 20 segundos, apan i-update ang counter matag 5 segundos, sama sa gipakita sa Fig. 5.15. Sa kini nga kaso, kami adunay tulo nga mga bintana sa resulta nga adunay nagsapaw nga datos.

Ang listahan sa 5.7 nagpakita sa code alang sa pagtino sa sliding windows (makita sa src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

Ang usa ka tumbling window mahimong makombertir sa usa ka hopping window pinaagi sa pagdugang sa usa ka tawag sa advanceBy() nga pamaagi. Sa gipakita nga panig-ingnan, ang sal-ang sa pagluwas mao ang 15 minuto.

Nakita nimo sa kini nga seksyon kung giunsa limitahan ang mga resulta sa panagsama sa mga bintana sa oras. Sa partikular, gusto ko nga imong hinumdoman ang mosunod nga tulo ka mga butang gikan niini nga seksyon:

ang gidak-on sa mga bintana sa sesyon limitado dili sa yugto sa panahon, apan sa kalihokan sa user;
Ang "tumbling" nga mga bintana naghatag usa ka kinatibuk-ang pagtan-aw sa mga panghitabo sa sulod sa gihatag nga yugto sa panahon;
Ang gidugayon sa paglukso sa mga bintana gitakda, apan kini kanunay nga gi-update ug mahimong adunay nagsapaw nga mga entry sa tanan nga mga bintana.

Sunod, mahibal-an namon kung giunsa ang pag-convert sa usa ka KTable balik sa usa ka KStream para sa usa ka koneksyon.

5.3.3. Pagkonektar sa KStream ug KTable nga mga butang

Sa Kapitulo 4, among gihisgutan ang pagkonektar sa duha ka butang sa KStream. Karon kinahanglan naton mahibal-an kung giunsa ang pagkonektar sa KTable ug KStream. Mahimong gikinahanglan kini tungod sa mosunod nga yanong rason. Ang KStream usa ka sapa sa mga rekord, ug ang KTable usa ka sapa sa mga update sa rekord, apan usahay mahimo nimong idugang ang dugang nga konteksto sa stream sa rekord gamit ang mga update gikan sa KTable.

Atong kuhaon ang datos sa gidaghanon sa mga transaksyon sa stock exchange ug isagol kini sa mga balita sa stock exchange alang sa may kalabutan nga mga industriya. Ania ang kinahanglan nimong buhaton aron makab-ot kini nga gihatag ang code nga naa na nimo.

Pag-convert sa usa ka butang nga KTable nga adunay datos sa gidaghanon sa mga transaksyon sa stock ngadto sa usa ka KStream, gisundan sa pag-ilis sa yawe sa yawe nga nagpakita sa sektor sa industriya nga katumbas niini nga simbolo sa stock.
Paghimo usa ka butang nga KTable nga nagbasa sa datos gikan sa usa ka hilisgutan nga adunay balita sa stock exchange. Kining bag-ong KTable i-categorize sa sektor sa industriya.
Ikonektar ang mga update sa balita sa impormasyon sa gidaghanon sa mga transaksyon sa stock exchange sa sektor sa industriya.

Karon atong tan-awon kon unsaon pagpatuman niini nga plano sa aksyon.

I-convert ang KTable sa KStream

Aron mabag-o ang KTable sa KStream kinahanglan nimo buhaton ang mosunud.

Tawga ang KTable.toStream() nga pamaagi.
Pinaagi sa pagtawag sa pamaagi sa KStream.map, ilisan ang yawe sa ngalan sa industriya, ug dayon kuhaa ang TransactionSummary nga butang gikan sa Windowed nga pananglitan.

Atong i-chain kini nga mga operasyon sama sa mosunod (ang code makita sa file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.8).

Tungod kay nagpahigayon kami og operasyon sa KStream.map, ang gibalik nga instance sa KStream awtomatik nga gibahin pag-usab kung kini gigamit sa usa ka koneksyon.

Nakompleto na namo ang proseso sa pagkakabig, sunod kinahanglan namong maghimo ug KTable nga butang para sa pagbasa sa stock news.

Pagmugna sa KTable alang sa stock nga balita

Maayo na lang, ang paghimo og KTable nga butang nagkinahanglan lang og usa ka linya sa code (ang code makita sa src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.9).

Angay nga matikdan nga walay Serde nga mga butang ang gikinahanglan nga espesipiko, tungod kay ang string Serdes gigamit sa mga setting. Usab, pinaagi sa paggamit sa UNA nga enumeration, ang lamesa napuno sa mga rekord sa sinugdanan.

Karon makapadayon na kita sa katapusang lakang - koneksyon.

Pagkonektar sa mga update sa balita sa datos sa ihap sa transaksyon

Ang paghimo og koneksyon dili lisud. Maggamit kami og left join kung walay stock news para sa may kalabutan nga industriya (ang gikinahanglan nga code makita sa file src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.10).

Kining leftJoin operator kay simple ra. Dili sama sa mga pag-apil sa Kapitulo 4, ang JoinWindow nga pamaagi wala gigamit tungod kay sa pagbuhat sa usa ka KStream-KTable nga pag-apil, adunay usa lamang ka entry sa KTable alang sa matag yawe. Ang ingon nga koneksyon dili limitado sa oras: ang rekord naa sa KTable o wala. Ang panguna nga konklusyon: gamit ang KTable nga mga butang mahimo nimong mapauswag ang KStream nga adunay dili kaayo kanunay nga gi-update nga datos sa pakisayran.

Karon atong tan-awon ang usa ka mas episyente nga paagi aron mapauswag ang mga panghitabo gikan sa KStream.

5.3.4. GlobalKTable nga mga butang

Sama sa imong nakita, adunay panginahanglan sa pagpauswag sa mga sapa sa panghitabo o pagdugang sa konteksto niini. Sa Kapitulo 4 imong nakita ang koneksyon tali sa duha ka KStream nga butang, ug sa miaging seksyon imong nakita ang koneksyon tali sa KStream ug KTable. Sa tanan niini nga mga kaso, gikinahanglan nga i-partition pag-usab ang data stream sa dihang magmapa sa mga yawe ngadto sa bag-ong tipo o bili. Usahay ang repartitioning gihimo nga klaro, ug usahay ang Kafka Streams awtomatiko kini. Ang re-partitioning gikinahanglan tungod kay ang mga yawe nausab ug ang mga rekord kinahanglan nga mahuman sa bag-ong mga seksyon, kung dili ang koneksyon mahimong imposible (kini gihisgutan sa Kapitulo 4, sa seksyon nga "Re-partitioning data" sa subsection 4.2.4).

Ang re-partition adunay gasto

Ang pagbahin pag-usab nanginahanglan mga gasto - dugang nga gasto sa kapanguhaan alang sa paghimo sa mga intermediate nga mga hilisgutan, pagtipig sa doble nga datos sa lain nga hilisgutan; nagpasabut usab kini nga dugang nga latency tungod sa pagsulat ug pagbasa gikan niini nga hilisgutan. Dugang pa, kung kinahanglan nimo nga moapil sa labaw sa usa ka aspeto o dimensyon, kinahanglan nimo nga kadena ang mga pagdugtong, mapa ang mga rekord gamit ang bag-ong mga yawe, ug ipadagan pag-usab ang proseso sa pagbahinbahin.

Pagkonektar sa gagmay nga mga dataset

Sa pipila ka mga kaso, ang gidaghanon sa reference data nga konektado medyo gamay, mao nga ang kompleto nga mga kopya niini dali nga mohaum sa lokal sa matag node. Alang sa mga sitwasyon nga sama niini, ang Kafka Streams naghatag sa GlobalKTable nga klase.

Talagsaon ang mga higayon sa GlobalKTable tungod kay gikopya sa aplikasyon ang tanan nga datos sa matag usa sa mga node. Ug tungod kay ang tanan nga datos anaa sa matag node, dili kinahanglan nga partition ang stream sa panghitabo pinaagi sa reference data key aron kini magamit sa tanan nga mga partisyon. Mahimo ka usab nga maghimo mga keyless nga pag-apil gamit ang GlobalKTable nga mga butang. Balikan nato ang usa sa miaging mga pananglitan aron ipakita kini nga bahin.

Pagkonektar sa mga butang sa KStream sa mga butang nga GlobalKTable

Sa subsection 5.3.2, among gihimo ang window aggregation sa exchange transactions sa mga pumapalit. Ang mga resulta niini nga aggregation ingon niini:

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

Samtang kini nga mga resulta nagsilbi sa katuyoan, mas mapuslanon kung ang ngalan sa kustomer ug ang tibuuk nga ngalan sa kompanya gipakita usab. Aron idugang ang ngalan sa kustomer ug ngalan sa kompanya, mahimo nimo ang normal nga pag-apil, apan kinahanglan nimo nga buhaton ang duha nga hinungdanon nga mga mapa ug pagbahin pag-usab. Uban sa GlobalKTable imong malikayan ang gasto sa maong mga operasyon.

Aron mahimo kini, atong gamiton ang countStream nga butang gikan sa Listing 5.11 (ang katugbang nga code makita sa src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) ug ikonektar kini sa duha ka GlobalKTable nga mga butang.

Gihisgotan na nato kini kaniadto, mao nga dili na nako kini balikon. Apan akong namatikdan nga ang code sa toStream().map function kay abstracted ngadto sa function object imbes sa usa ka inline nga lambda expression para sa pagkabasa.

Ang sunod nga lakang mao ang pagdeklara sa duha ka mga higayon sa GlobalKTable (ang gipakita nga code makita sa file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (Listing 5.12).

Palihug timan-i nga ang mga ngalan sa topiko gihulagway gamit ang enumerated type.

Karon nga andam na ang tanan nga sangkap, ang nahabilin mao ang pagsulat sa code alang sa koneksyon (nga makita sa file src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (Listing 5.13).

Bisan tuod adunay duha ka mga sumpay niini nga kodigo, sila gikadena tungod kay walay usa sa ilang mga resulta gigamit nga gilain. Ang mga resulta gipakita sa katapusan sa tibuok nga operasyon.

Kung gipadagan nimo ang operasyon sa pag-apil sa ibabaw, makakuha ka mga resulta nga sama niini:

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

Ang esensya wala mausab, apan kini nga mga resulta tan-awon nga mas klaro.

Kung mag-ihap ka hangtod sa Kapitulo 4, nakakita ka na daghang mga klase sa koneksyon nga naglihok. Gilista sila sa lamesa. 5.2. Kini nga lamesa nagpakita sa mga kapabilidad sa pagkonektar sa bersyon 1.0.0 sa Kafka Streams; Adunay mahimong mausab sa umaabot nga pagpagawas.

Aron maputos ang mga butang, atong i-recap ang mga sukaranan: mahimo nimong ikonektar ang mga sapa sa panghitabo (KStream) ug i-update ang mga sapa (KTable) gamit ang lokal nga estado. Sa laing bahin, kung ang gidak-on sa reference data dili kaayo dako, mahimo nimong gamiton ang GlobalKTable nga butang. Gikopya sa GlobalKTables ang tanan nga partisyon sa matag node sa aplikasyon sa Kafka Streams, pagsiguro nga magamit ang tanan nga datos bisan unsa nga partisyon ang katumbas sa yawe.

Sunod atong makita ang bahin sa Kafka Streams, salamat nga atong maobserbahan ang mga pagbag-o sa estado nga wala mag-usik sa datos gikan sa usa ka hilisgutan sa Kafka.

5.3.5. Pangutana nga kahimtang

Nakahimo na kami og daghang mga operasyon nga naglambigit sa estado ug kanunay nga nagpagawas sa mga resulta ngadto sa console (alang sa mga katuyoan sa pagpalambo) o isulat kini sa usa ka hilisgutan (alang sa mga katuyoan sa produksyon). Kung nagsulat sa mga resulta sa usa ka hilisgutan, kinahanglan nimo nga mogamit usa ka konsumedor sa Kafka aron makita kini.

Ang pagbasa sa datos gikan niini nga mga hilisgutan mahimong isipon nga usa ka matang sa materyal nga mga panglantaw. Alang sa among mga katuyoan, magamit namon ang kahulugan sa usa ka materyal nga pagtan-aw gikan sa Wikipedia: "...usa ka pisikal nga butang sa database nga adunay sulud nga mga sangputanan sa usa ka pangutana. Pananglitan, kini mahimong usa ka lokal nga kopya sa hilit nga datos, o usa ka subset sa mga laray ug/o mga kolum sa usa ka lamesa o mga resulta sa pag-apil, o usa ka summary table nga nakuha pinaagi sa aggregation” (https://en.wikipedia.org/wiki /Materialized_view).

Gitugotan ka usab sa Kafka Streams sa pagpadagan sa mga interactive nga pangutana sa mga tindahan sa estado, nga gitugotan ka nga direkta nga mabasa kini nga mga materyal nga pagtan-aw. Importante nga hinumdoman nga ang pangutana sa tindahan sa estado usa ka read-only nga operasyon. Gisiguro niini nga dili ka kinahanglan mabalaka bahin sa aksidente nga paghimo sa estado nga dili managsama samtang ang imong aplikasyon nagproseso sa datos.

Importante ang abilidad sa direktang pagpangutana sa mga tindahan sa estado. Nagpasabot kini nga makahimo ka og mga aplikasyon sa dashboard nga dili kinahanglan nga magkuha una og datos gikan sa konsumidor sa Kafka. Gidugangan usab niini ang kaepektibo sa aplikasyon, tungod sa kamatuoran nga dili na kinahanglan nga isulat pag-usab ang datos:

salamat sa lokalidad sa datos, dali silang ma-access;
Ang pagdoble sa datos giwagtang, tungod kay wala kini gisulat sa external storage.

Ang panguna nga butang nga gusto nako nga imong hinumdoman mao nga mahimo nimo direkta nga mangutana sa estado gikan sa sulod sa imong aplikasyon. Ang mga oportunidad nga gihatag niini kanimo dili mahimong palabihon. Imbis nga mag-usik sa datos gikan sa Kafka ug magtipig sa mga rekord sa usa ka database alang sa aplikasyon, mahimo nimong pangutana ang mga tindahan sa estado nga adunay parehas nga sangputanan. Ang direkta nga mga pangutana sa mga tindahan sa estado nagpasabut nga gamay nga code (walay konsumedor) ug gamay nga software (dili kinahanglan ang usa ka lamesa sa database aron matipigan ang mga resulta).

Natabonan na namo ang gamay nga bahin niini nga kapitulo, mao nga ibilin namo ang among diskusyon sa interactive nga mga pangutana batok sa mga tindahan sa estado sa pagkakaron. Apan ayaw kabalaka: sa Kapitulo 9, maghimo kami usa ka yano nga aplikasyon sa dashboard nga adunay mga interactive nga pangutana. Gamiton niini ang pipila ka mga pananglitan gikan niini ug sa miaging mga kapitulo aron ipakita ang mga interactive nga pangutana ug kung giunsa nimo kini idugang sa mga aplikasyon sa Kafka Streams.

Sumaryo

Ang mga butang sa KStream nagrepresentar sa mga sapa sa mga panghitabo, ikatandi sa mga pagsal-ot sa usa ka database. Ang mga butang nga KTable nagrepresentar sa mga sapa sa pag-update, nga mas sama sa mga update sa usa ka database. Ang gidak-on sa butang nga KTable dili motubo, ang daan nga mga rekord gipulihan sa mga bag-o.
Ang mga butang nga KTable gikinahanglan alang sa mga operasyon sa aggregation.
Pinaagi sa paggamit sa mga operasyon sa windowing, mahimo nimong bahinon ang mga aggregated data sa mga balde sa oras.
Salamat sa GlobalKTable nga mga butang, mahimo nimong ma-access ang reference data bisan asa sa aplikasyon, bisan unsa pa ang partitioning.
Ang mga koneksyon tali sa KStream, KTable ug GlobalKTable nga mga butang posible.

Hangtod karon, naka-focus kami sa pagtukod sa mga aplikasyon sa Kafka Streams gamit ang taas nga lebel nga KStream DSL. Bisan kung ang taas nga lebel nga pamaagi nagtugot kanimo sa paghimo og hapsay ug mubo nga mga programa, ang paggamit niini nagrepresentar sa usa ka trade-off. Ang pagtrabaho kauban ang DSL KStream nagpasabut sa pagdugang sa pagkamubu sa imong code pinaagi sa pagkunhod sa lebel sa pagkontrol. Sa sunod nga kapitulo, atong tan-awon ang low-level handler node API ug sulayan ang ubang mga trade-off. Ang mga programa mahimong mas taas kaysa kaniadto, apan makahimo kami sa paghimo sa halos bisan unsang handler node nga kinahanglan namon.

→ Dugang nga mga detalye bahin sa libro makita sa website sa magmamantala

→ Para sa Habrozhiteli 25% nga diskwento gamit ang kupon - Mga sapa sa Kafka

→ Sa pagbayad alang sa papel nga bersyon sa libro, usa ka elektronik nga libro ipadala pinaagi sa e-mail.

Source: www.habr.com

Ang libro nga "Kafka Streams in Action. Mga aplikasyon ug microservices para sa real-time nga trabaho"