{"id":32632,"date":"2019-10-31T21:48:08","date_gmt":"2019-10-31T18:48:08","guid":{"rendered":"https:\/\/prohoster.info\/blog\/ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo\/"},"modified":"2019-10-31T21:48:08","modified_gmt":"2019-10-31T18:48:08","slug":"ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo","status":"publish","type":"post","link":"https:\/\/prohoster.info\/sq\/blog\/administrirovanie\/ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo","title":{"rendered":"Jo vet\u00ebm p\u00ebrpunimi: Si e transformuam Kafka Streams n\u00eb nj\u00eb baz\u00eb t\u00eb shp\u00ebrndar\u00eb t\u00eb t\u00eb dh\u00ebnave dhe \u00e7far\u00eb doli nga kjo","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

P\u00ebrsh\u00ebndetje, Habr!<\/p>\n

Ju kujtojm\u00eb se pas librit p\u00ebr Kafka<\/a><\/noindex> ne kemi nxjerr\u00eb nj\u00eb vep\u00ebr t\u00eb tjera interesante p\u00ebr bibliotek\u00ebn Kafka Streams API<\/a><\/noindex>. <\/p>\n

\"Jo
\n
\nNd\u00ebrkoh\u00eb q\u00eb komuniteti po zbret n\u00eb kufijt\u00eb e mund\u00ebsive t\u00eb k\u00ebtij mjeti t\u00eb fuqish\u00ebm. S\u00eb fundmi, u publikua nj\u00eb artikull, me p\u00ebrkthimin e t\u00eb cilit d\u00ebshirojm\u00eb t'ju njohim. Autori, p\u00ebrmes p\u00ebrvoj\u00ebs s\u00eb tij, tregon se si t\u00eb krijosh nj\u00eb depo t\u00eb shp\u00ebrndar\u00eb t\u00eb t\u00eb dh\u00ebnave nga Kafka Streams. Lexim t\u00eb k\u00ebndsh\u00ebm!
\n<\/a><\/noindex>
\nBiblioteka Apache Kafka Streams<\/a><\/noindex> p\u00ebrdoret n\u00eb mbar\u00eb bot\u00ebn n\u00eb nd\u00ebrmarrje p\u00ebr procesimin e shp\u00ebrndar\u00eb t\u00eb flukseve mbi Apache Kafka. Nj\u00eb nga aspektet e n\u00ebnvler\u00ebsuara t\u00eb k\u00ebtij framework-u \u00ebsht\u00eb se ai lejon ruajtjen e nj\u00eb gjendjeje lokale, e cila prodhohet n\u00eb baz\u00eb t\u00eb p\u00ebrpunimit t\u00eb flukseve.<\/p>\n

N\u00eb k\u00ebt\u00eb artikull, do t\u00eb flas p\u00ebr m\u00ebnyr\u00ebn sesi n\u00eb kompanin\u00eb ton\u00eb arrit\u00ebm t\u00eb p\u00ebrfitojm\u00eb nga kjo mund\u00ebsi gjat\u00eb zhvillimit t\u00eb nj\u00eb produkti p\u00ebr sigurin\u00eb e aplikacioneve n\u00eb re. Me ndihm\u00ebn e Kafka Streams, ne krijuam mikrosh\u00ebrbime me gjendje t\u00eb ndar\u00eb, secili prej t\u00eb cil\u00ebve na sh\u00ebrben si nj\u00eb burim i q\u00ebndruesh\u00ebm dhe me disponueshm\u00ebri t\u00eb lart\u00eb t\u00eb informacionit t\u00eb besuesh\u00ebm mbi gjendjen e objekteve n\u00eb sistem. P\u00ebr ne, kjo \u00ebsht\u00eb nj\u00eb p\u00ebrpar\u00ebsi sa n\u00eb aspektin e q\u00ebndrueshm\u00ebris\u00eb, ashtu edhe n\u00eb p\u00ebrmir\u00ebsimin e mb\u00ebshtetjes.<\/p>\n

N\u00ebse jeni t\u00eb interesuar p\u00ebr nj\u00eb qasje alternative q\u00eb lejon p\u00ebrdorimin e nj\u00eb baze t\u00eb dh\u00ebnash qendrore p\u00ebr t\u00eb mb\u00ebshtetur gjendjen formale t\u00eb objekteve tuaj \u2013 lexoni, do t\u00eb jet\u00eb interesante\u2026 <\/p>\n

Pse e konsideruam se ka ardhur koha p\u00ebr t\u00eb ndryshuar qasjet tona n\u00eb pun\u00ebn me gjendjen e ndar\u00eb <\/b><\/p>\n

Na duheshin mb\u00ebshtetje p\u00ebr gjendjen e objekteve t\u00eb ndryshme, duke u mb\u00ebshtetur n\u00eb raportet e agen\u00ebve (p.sh.: a \u00ebsht\u00eb sulmuar site-i)? Para kalimit n\u00eb Kafka Streams, shpesh kemi varur menaxhimin e gjendjes nga nj\u00eb baz\u00eb t\u00eb dh\u00ebnash qendrore (+ API sh\u00ebrbimi). Ky qasje ka disa disavantazhe: n\u00eb situatat e datenintensiv<\/a><\/noindex> mb\u00ebshtetje p\u00ebr konsistenc\u00ebn dhe sinkronizimin b\u00ebhet nj\u00eb sfid\u00eb e v\u00ebrtet\u00eb. Baza e t\u00eb dh\u00ebnave mund t\u00eb b\u00ebhet nyja e ngusht\u00eb, ose t\u00eb ndodhet n\u00eb gjendje garash<\/a><\/noindex> dhe t\u00eb p\u00ebrbalet me pap\u00ebrcaktueshm\u00ebrin\u00eb.<\/p>\n

\"Jo
\n
\nIlustrimi 1: nj\u00eb skenar tipik me ndarje t\u00eb gjendjes, q\u00eb ndodhi para kalimit n\u00eb
\nKafka dhe Kafka Streams: agent\u00ebt raportojn\u00eb p\u00ebrmbledhjet e tyre p\u00ebrmes API-s\u00eb, gjendja e azhurnuar llogaritet p\u00ebrmes nj\u00eb baze t\u00eb dh\u00ebnash qendrore <\/i><\/p>\n

Njoftohuni me Kafka Streams \u2013 tani \u00ebsht\u00eb b\u00ebr\u00eb e leht\u00eb t\u00eb krijoni mikros\u00ebrvime me gjendje t\u00eb ndar\u00eb <\/b><\/p>\n

Rreth nj\u00eb vit m\u00eb par\u00eb, ne vendos\u00ebm t\u00eb rishikojm\u00eb me kujdes skenar\u00ebt tan\u00eb p\u00ebr menaxhimin e gjendjes s\u00eb ndar\u00eb p\u00ebr t\u00eb trajtuar disa probleme. Menj\u00ebher\u00eb vendos\u00ebm t\u00eb provonim Kafka Streams \u2013 dihet se sa e shkall\u00ebzueshme, e disponueshme dhe e q\u00ebndrueshme \u00ebsht\u00eb, si dhe sa e pasur \u00ebsht\u00eb me funksionalitete p\u00ebr p\u00ebrpunim rrjedhash (transformime, p\u00ebrfshir\u00eb ato me ruajtjen e gjendjes). Pik\u00ebrisht ajo q\u00eb na nevojitej, pa p\u00ebrmendur se sa e pjekur dhe e besueshme \u00ebsht\u00eb sistemi i d\u00ebrgimit t\u00eb mesazheve n\u00eb Kafka.<\/p>\n

\u00c7do nj\u00eb nga sh\u00ebrbimet mikro q\u00eb krijuam me ruajtjen e gjendjes u nd\u00ebrtua mbi nj\u00eb instanc\u00eb t\u00eb Kafka Streams me nj\u00eb topologji mjaft t\u00eb thjesht\u00eb. Ajo p\u00ebrb\u00ebhej nga 1) burimi 2) procesori me nj\u00eb magazin\u00eb t\u00eb vazhdueshme t\u00eb \u00e7el\u00ebsave dhe vlerave 3) rrjedha:<\/p>\n

\"Jo
\n
\nIlustrimi 2: topologjia e paracaktuar e instancave tona t\u00eb rrjedhave p\u00ebr sh\u00ebrbimet mikro me ruajtjen e gjendjes. Vini re: k\u00ebtu ka gjithashtu nj\u00eb magazin\u00eb ku ruhen metadat p\u00ebr planifikimin. <\/i><\/p>\n

Me k\u00ebt\u00eb qasje t\u00eb re, agjent\u00ebt formulojn\u00eb mesazhe q\u00eb d\u00ebrgohen n\u00eb tem\u00ebn e origjin\u00ebs, nd\u00ebrsa konsumator\u00ebt \u2013 le t\u00eb themi, sh\u00ebrbimi i njoftimeve p\u00ebrmes post\u00ebs \u2013 marrin gjendjen e llogaritur t\u00eb ndar\u00eb p\u00ebrmes koleksionit (tem\u00ebs s\u00eb daljes).<\/p>\n

\"Jo
\n
\nIlustrimi 3: nj\u00eb shembuj i ri i fluksit t\u00eb detyrave p\u00ebr skenar\u00eb me mikrosh\u00ebrbime t\u00eb ndara: 1) agjenti krijon nj\u00eb mesazh q\u00eb arrin n\u00eb tem\u00ebn e origjin\u00ebs Kafka; 2) mikrosh\u00ebrbimi me gjendje t\u00eb ndar\u00eb (duke p\u00ebrdorur Kafka Streams) e proceson at\u00eb dhe regjistron gjendjen e llogaritur n\u00eb tem\u00ebn p\u00ebrfundimtare Kafka; pas k\u00ebsaj, 3) konsumator\u00ebt marrin gjendjen e re.<\/i><\/p>\n

Hej, dhe kjo magazin\u00eb e integruar e \u00e7el\u00ebsave dhe vlerave \u00ebsht\u00eb v\u00ebrtet shum\u00eb e dobishme!<\/b><\/p>\n

Si\u00e7 u p\u00ebrmend m\u00eb sip\u00ebr, topologjia jon\u00eb me gjendje t\u00eb ndar\u00eb p\u00ebrmban nj\u00eb magazin\u00eb \u00e7el\u00ebsash dhe vlerash. Ne gjet\u00ebm disa m\u00ebnyra p\u00ebr ta p\u00ebrdorur at\u00eb, dhe dy prej tyre jan\u00eb p\u00ebrshkruar m\u00eb posht\u00eb.<\/p>\n

M\u00ebnyra #1: p\u00ebrdorimi i magazin\u00ebs s\u00eb \u00e7el\u00ebsave dhe vlerave gjat\u00eb llogaritjeve.<\/b><\/i><\/p>\n

Depozita jon\u00eb e par\u00eb e \u00e7el\u00ebsave dhe vlerave p\u00ebrmbante t\u00eb dh\u00ebna ndihm\u00ebse q\u00eb na duhej p\u00ebr llogaritjet. P\u00ebr shembull, n\u00eb disa raste, gjendja e ndar\u00eb u p\u00ebrcaktua sipas principit t\u00eb \"shumic\u00ebs s\u00eb votave\". N\u00eb depo mund t\u00eb mbaheshin t\u00eb gjitha raportet m\u00eb t\u00eb fundit t\u00ebagjent\u00ebve n\u00eb lidhje me nj\u00eb objekt t\u00eb caktuar. M\u00eb pas, duke marr\u00eb nj\u00eb raport t\u00eb ri nga ndonj\u00eb agent, ne mund t\u00eb ruanim at\u00eb, t\u00eb nxirrnim nga depo raportet e gjith\u00eb agjent\u00ebve t\u00eb tjer\u00eb p\u00ebr gjendjen e t\u00eb nj\u00ebjtit objekt dhe t\u00eb p\u00ebrs\u00ebrisnim llogaritjen.
\nM\u00eb posht\u00eb n\u00eb ilustrimin 4 tregohet se si ne hap\u00ebm aksesin n\u00eb depozitat e \u00e7el\u00ebsave dhe vlerave p\u00ebr metod\u00ebn e p\u00ebrpunimit t\u00eb procesorit, k\u00ebshtu q\u00eb m\u00eb pas mund t\u00eb p\u00ebrpunonim nj\u00eb mesazh t\u00eb ri.<\/p>\n

\"Jo
\n
\nIlustrimi 4: hapja e aksesit n\u00eb depozitat e \u00e7el\u00ebsave dhe vlerave p\u00ebr metod\u00ebn e p\u00ebrpunimit t\u00eb procesorit (pas k\u00ebsaj, n\u00eb \u00e7do skenar q\u00eb punon me gjendjen e ndar\u00eb, nevojitet t\u00eb implementohet metoda doProcess<\/code>)<\/i><\/p>\n

Opcioni #2: krijimi i nj\u00eb API CRUD mbi Kafka Streams<\/b><\/i><\/p>\n

Pas pasi p\u00ebr t\u00eb nd\u00ebrtuar fluxin ton\u00eb baz\u00eb t\u00eb detyrave, filluam t\u00eb provonim t\u00eb shkruanim nj\u00eb RESTful CRUD API p\u00ebr mikrosh\u00ebrbimet tona me gjendje t\u00eb ndar\u00eb. Ne donim q\u00eb t\u00eb mund t\u00eb nxirrnim gjendjen e disa ose t\u00eb gjitha objekteve, si dhe t\u00eb vendosnim ose hiqnim gjendjen e nj\u00eb objekti (kjo \u00ebsht\u00eb e dobishme p\u00ebr mb\u00ebshtetje n\u00eb an\u00ebn serverike).<\/p>\n

P\u00ebr t\u00eb mb\u00ebshtetur t\u00eb gjith\u00eb API-n\u00eb Get State, sa her\u00eb na nevojitej t\u00eb llogarisnim p\u00ebrs\u00ebri gjendjen gjat\u00eb p\u00ebrpunimit, ne e ruanim at\u00eb p\u00ebr nj\u00eb koh\u00eb t\u00eb gjat\u00eb n\u00eb ruajtjen e integruar t\u00eb \u00e7el\u00ebsave dhe vlerave. N\u00eb k\u00ebt\u00eb rast, \u00ebsht\u00eb mjaft e thjesht\u00eb t\u00eb implementoni nj\u00eb API t\u00eb till\u00eb duke p\u00ebrdorur nj\u00eb instanc\u00eb t\u00eb vetme t\u00eb Kafka Streams, si\u00e7 \u00ebsht\u00eb treguar n\u00eb listimin m\u00eb posht\u00eb:<\/p>\n

\"Jo
\n
\nIlustrimi 5: p\u00ebrdorimi i ruajtjes s\u00eb integruar t\u00eb \u00e7el\u00ebsave dhe vlerave p\u00ebr t\u00eb marr\u00eb gjendjen e parakalkuluar t\u00eb objektit <\/i><\/p>\n

P\u00ebrdit\u00ebsimi i gjendjes s\u00eb objektit p\u00ebrmes API-s\u00eb gjithashtu nuk \u00ebsht\u00eb e v\u00ebshtir\u00eb p\u00ebr t'u realizuar. N\u00eb parim, p\u00ebr k\u00ebt\u00eb vet\u00ebm duhet t\u00eb krijoni nj\u00eb prodhues Kafka, dhe me t\u00eb t\u00eb b\u00ebni nj\u00eb shkronj\u00eb, n\u00eb t\u00eb cil\u00ebn p\u00ebrmban gjendjen e re. K\u00ebshtu garanton q\u00eb t\u00eb gjitha mesazhet e gjeneruara p\u00ebrmes API-s\u00eb do t\u00eb p\u00ebrpunohen n\u00eb t\u00eb nj\u00ebjt\u00ebn m\u00ebnyr\u00eb si ato q\u00eb vijn\u00eb nga prodhues t\u00eb tjer\u00eb (p.sh. agjent\u00ebt).<\/p>\n

\"Jo
\n
\nIlustrimi 6: mund t\u00eb p\u00ebrcaktojm\u00eb gjendjen e objektit p\u00ebrmes produesit Kafka <\/i><\/p>\n

Nj\u00eb komplikim i vog\u00ebl: Kafka ka shum\u00eb parti<\/b><\/p>\n

M\u00eb pas donim t\u00eb shp\u00ebrndanim ngarkes\u00ebn e p\u00ebrpunimit dhe t\u00eb p\u00ebrmir\u00ebsonim disponueshm\u00ebrin\u00eb, duke ofruar p\u00ebr \u00e7do skenar nj\u00eb klaster mikrosh\u00ebrbimesh me gjendje t\u00eb p\u00ebrbashk\u00ebt. Konfigurimi doli t\u00eb ishte mjaft i thjesht\u00eb: pasi konfiguruam t\u00eb gjith\u00eb instancat q\u00eb t\u00eb punonin me t\u00eb nj\u00ebjtin ID t\u00eb aplikacionit (dhe me t\u00eb nj\u00ebjtat server\u00eb t\u00eb ngarkes\u00ebs fillestare), pothuajse gjith\u00e7ka tjet\u00ebr u realizua automatikisht. Ne gjithashtu caktuam q\u00eb \u00e7do tem\u00eb burimi do t\u00eb p\u00ebrb\u00ebhej nga disa parti, n\u00eb m\u00ebnyr\u00eb q\u00eb secil\u00ebs instanc\u00eb t'i jepet nj\u00eb n\u00ebngrup i till\u00eb partitesh.<\/p>\n

Gjithashtu do t\u00eb p\u00ebrmend se k\u00ebtu \u00ebsht\u00eb zakon q\u00eb t\u00eb b\u00ebhet nj\u00eb kopje rezerv\u00eb e magazin\u00ebs s\u00eb gjendjeve, q\u00eb, p\u00ebr shembull, n\u00eb rast rikuperimi pas nj\u00eb d\u00ebshtimi, ajo t\u00eb transferohet n\u00eb nj\u00eb instanc\u00eb tjet\u00ebr. P\u00ebr \u00e7do magazin\u00eb gjendjesh n\u00eb Kafka Streams krijohet nj\u00eb tem\u00eb e riprodhueshme me nj\u00eb regjist\u00ebr ndryshimesh (ku ndiqen azhurnimet lokale). K\u00ebshtu, Kafka vazhdimisht mbulon magazin\u00ebn e gjendjeve. Prandaj, n\u00eb rast d\u00ebshtimi t\u00eb ndonj\u00eb instanc\u00eb t\u00eb Kafka Streams, magazina e gjendjeve mund t\u00eb rikuperohet shpejt n\u00eb nj\u00eb instanc\u00eb tjet\u00ebr, ku do t\u00eb transferohen partit\u00eb p\u00ebrkat\u00ebse. Testet tona treguan se kjo b\u00ebhet p\u00ebr pak sekonda, edhe n\u00ebse n\u00eb magazin\u00eb ka miliona regjistrime.<\/p>\n

Duke kaluar nga nj\u00eb mikrosh\u00ebrbim me nj\u00eb gjendje t\u00eb ndar\u00eb n\u00eb nj\u00eb klaster mikrosh\u00ebrbimesh, realizimi i Get State API nuk \u00ebsht\u00eb kaq triviale. N\u00eb situat\u00ebn e re, n\u00eb ruajtjen e gjendjeve t\u00eb \u00e7do mikrosh\u00ebrbimi ndodhet vet\u00ebm nj\u00eb pjes\u00eb e tablos\u00eb p\u00ebrkat\u00ebse (ato objekte t\u00eb cilat \u00e7el\u00ebsat e tyre u shfaq\u00ebn n\u00eb nj\u00eb parti t\u00eb caktuar). Duhej t\u00eb p\u00ebrcaktonim se n\u00eb cilin instanc\u00eb ndodhej gjendja e objektit q\u00eb na nevojitej, dhe k\u00ebt\u00eb e b\u00ebnim n\u00eb baz\u00eb t\u00eb metadata e rrjedhjeve, si\u00e7 tregohet m\u00eb posht\u00eb:<\/p>\n

\"Jo
\n
\nIlustrimi 7: me ndihm\u00ebn e metadata e rrjedhave ne p\u00ebrcaktojm\u00eb nga cil\u00eb instanc\u00eb t\u00eb k\u00ebrkojm\u00eb gjendjen e objektit t\u00eb k\u00ebrkuar; nj\u00eb qasje e till\u00eb u p\u00ebrdor p\u00ebr GET ALL API<\/i><\/p>\n

Konkluzionet kryesore<\/b><\/p>\n

Ruajtjet e gjendjeve n\u00eb Kafka Streams de facto mund t\u00eb sh\u00ebrbejn\u00eb si nj\u00eb baz\u00eb t\u00eb dh\u00ebnash t\u00eb shp\u00ebrndar\u00eb, <\/p>\n