{"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\/ro\/blog\/administrirovanie\/ne-tolko-obrabotka-kak-my-sdelali-iz-kafka-streams-raspredelennuyu-bazu-dannyh-i-chto-iz-etogo-vyshlo","title":{"rendered":"Nu doar procesare: Cum am transformat Kafka Streams \u00eentr-o baz\u0103 de date distribuit\u0103 \u0219i ce a rezultat din asta","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

Salut, Habr!<\/p>\n

V\u0103 reamintim c\u0103, pe l\u00e2ng\u0103 cartea despre Kafka<\/a><\/noindex> am publicat o lucrare la fel de interesant\u0103 despre biblioteca Kafka Streams API<\/a><\/noindex>. <\/p>\n

\"Nu
\n
\n\u00cen timp ce comunitatea abia \u00eencepe s\u0103 exploreze limitele acestui instrument puternic. Recent a fost publicat un articol, cu traducerea c\u0103ruia vrem s\u0103 v\u0103 familiariz\u0103m. Autorul poveste\u0219te din experien\u021ba proprie cum s\u0103 transformi Kafka Streams \u00eentr-un depozit de date distribuit. Lectur\u0103 pl\u0103cut\u0103!
\n<\/a><\/noindex>
\nBiblioteca Apache Kafka Streams<\/a><\/noindex> este utilizat\u0103 la nivel mondial \u00een mediul enterprise pentru procesarea fluxurilor distribuite pe baza Apache Kafka. Unul dintre aspectele subestimate ale acestui cadru este c\u0103 permite p\u0103strarea unui stat local, generat pe baza proces\u0103rii fluxurilor.<\/p>\n

\u00cen acest articol, voi explica cum compania noastr\u0103 a reu\u0219it s\u0103 utilizeze eficient aceast\u0103 oportunitate \u00een dezvoltarea unui produs pentru securitatea aplica\u021biilor cloud. Cu ajutorul Kafka Streams, am creat microservicii cu stare partajat\u0103, fiecare dintre ele servind ca o surs\u0103 de informa\u021bii de \u00eencredere, rezistent\u0103 la erori \u0219i cu disponibilitate ridicat\u0103, despre starea obiectelor din sistem. Pentru noi, acesta a fost un pas \u00eenainte at\u00e2t \u00een ceea ce prive\u0219te fiabilitatea, c\u00e2t \u0219i \u00een confortul \u00eentre\u021binerii.<\/p>\n

Dac\u0103 sunte\u021bi interesat de o abordare alternativ\u0103 care permite utilizarea unei baze de date centrale unice pentru a sus\u021bine starea formal\u0103 a obiectelor dumneavoastr\u0103 \u2013 citi\u021bi, va fi interesant... <\/p>\n

De ce am considerat c\u0103 este timpul s\u0103 ne schimb\u0103m abord\u0103rile \u00een gestionarea st\u0103rii partajate <\/b><\/p>\n

Aveam nevoie s\u0103 men\u021binem starea diferitelor obiecte baz\u00e2ndu-ne pe rapoartele agen\u021bilor (de exemplu: a fost un atac asupra site-ului?). \u00cenainte de a migra la Kafka Streams, ne bazam adesea pentru gestionarea st\u0103rii pe o baz\u0103 de date central\u0103 unic\u0103 (+ API de serviciu). Aceast\u0103 abordare are dezavantajele sale: \u00een situa\u021bii intensive din punct de vedere al datelor<\/a><\/noindex> men\u021binerea consisten\u021bei \u0219i sincroniz\u0103rii devine o adev\u0103rat\u0103 provocare. Baza de date poate deveni un blocaj, sau s-ar putea afla \u00eentr-o stare de competi\u021bie<\/a><\/noindex> \u0219i s\u0103 sufere de imprevizibilitate.<\/p>\n

\"Nu
\n
\nIlustrarea 1: Un scenariu tipic de separare a st\u0103rii, \u00eent\u00e2lnit \u00eenainte de migrarea la
\nKafka \u0219i Kafka Streams: agen\u021bii comunic\u0103 viziuni prin API, starea actualizat\u0103 este calculat\u0103 prin baza de date central\u0103. <\/i><\/p>\n

Cunoa\u0219te\u021bi Kafka Streams \u2013 acum a devenit u\u0219or s\u0103 crea\u021bi microservicii cu stare partajat\u0103. <\/b><\/p>\n

Aproape un an \u00een urm\u0103, am decis s\u0103 reg\u00e2ndim seriile noastre de lucru cu starea partajat\u0103 pentru a rezolva astfel de probleme. Imediat am decis s\u0103 \u00eencerc\u0103m Kafka Streams - este binecunoscut\u0103 scalabilitatea, disponibilitatea ridicat\u0103 \u0219i rezilien\u021ba acesteia, precum \u0219i bogata func\u021bionalitate de streaming (inclusiv transform\u0103ri cu p\u0103strarea st\u0103rii). Exact ceea ce ne trebuia, nemaivorbind de c\u00e2t de matur\u0103 \u0219i fiabil\u0103 este sistemul de mesagerie dezvoltat \u00een Kafka.<\/p>\n

Fiecare dintre microserviciile noastre cu starea partajat\u0103 era construit pe baza unei instan\u021be Kafka Streams cu o topologie destul de simpl\u0103. Aceasta consta \u00een 1) o surs\u0103 2) un procesor cu un magazin permanent de chei \u0219i valori 3) un flux:<\/p>\n

\"Nu
\n
\nIlustrarea 2: topologia implicit\u0103 a instan\u021belor noastre de streaming pentru microserviciile cu starea partajat\u0103. Re\u021bine\u021bi: aici exist\u0103, de asemenea, un magazin care con\u021bine metadate despre programare. <\/i><\/p>\n

\u00cen acest nou mod de abordare, agen\u021bii compun mesajele trimise c\u0103tre topicul surs\u0103, iar consumatorii - s\u0103 zicem, serviciul de notific\u0103ri prin po\u0219t\u0103 - primesc starea partajat\u0103 calculat\u0103 prin flux (topicul de ie\u0219ire).<\/p>\n

\"Nu
\n
\nIlustrarea 3: un nou exemplu de flux de sarcini pentru un scenariu cu microservicii partajate: 1) agentul genereaz\u0103 un mesaj care ajunge \u00een topicul surs\u0103 Kafka; 2) microserviciul cu starea partajat\u0103 (folosind Kafka Streams) \u00eel proceseaz\u0103 \u0219i scrie starea calculat\u0103 \u00een topicul final Kafka; dup\u0103 care 3) consumatorii primesc noua stare.<\/i><\/p>\n

Hei, \u0219i acest magazin \u00eencorporat de chei \u0219i valori este cu adev\u0103rat foarte util!<\/b><\/p>\n

A\u0219a cum s-a men\u021bionat anterior, topologia noastr\u0103 cu starea partajat\u0103 con\u021bine un magazin de chei \u0219i valori. Am descoperit c\u00e2teva moduri de utilizare a acestuia, iar dou\u0103 dintre ele sunt descrise mai jos.<\/p>\n

Op\u021biunea #1: utilizarea magazinului de chei \u0219i valori \u00een cadrul calculilor.<\/b><\/i><\/p>\n

Primul nostru depozit de chei \u0219i valori con\u021binea date auxiliare de care aveam nevoie pentru calcule. De exemplu, \u00een unele cazuri, starea partajat\u0103 era determinat\u0103 pe baza principiului \u201emajorit\u0103\u021bii voturilor\u201d. \u00cen depozit se puteau p\u0103stra toate ultimele rapoarte ale agen\u021bilor despre starea unui anumit obiect. Apoi, primind un nou raport de la un anumit agent, puteam salva raportul, extrage din depozit rapoartele tuturor celorlal\u021bi agen\u021bi despre starea aceluia\u0219i obiect \u0219i repeta calculul.
\nIlustra\u021bia 4 de mai jos arat\u0103 cum am deschis accesul la depozitul de chei \u0219i valori pentru metoda de procesare a procesorului, astfel \u00eenc\u00e2t s\u0103 putem procesa un nou mesaj.<\/p>\n

\"Nu
\n
\nIlustra\u021bia 4: deschiderea accesului la depozitul de chei \u0219i valori pentru metoda de procesare a procesorului (dup\u0103 aceasta, \u00een fiecare scenariu care lucreaz\u0103 cu st\u0103ri partajate, este necesar s\u0103 se implementeze metoda doProcess<\/code>)<\/i><\/p>\n

Op\u021biunea #2: crearea unui API CRUD deasupra Kafka Streams<\/b><\/i><\/p>\n

Dup\u0103 ce ne-am configurat fluxul de sarcini de baz\u0103, am \u00eencercat s\u0103 scriem un API RESTful CRUD pentru microserviciile noastre cu stare partajat\u0103. Am dorit s\u0103 putem extrage starea unora sau a tuturor obiectelor, precum \u0219i s\u0103 stabilim sau s\u0103 \u0219tergem starea unui obiect (acest lucru este util pentru suportul p\u0103r\u021bii server).<\/p>\n

Pentru a sus\u021bine toate API-urile Get State, de fiecare dat\u0103 c\u00e2nd trebuia s\u0103 recalcul\u0103m starea \u00een timpul proces\u0103rii, o stocam pe termen lung \u00een depozitul \u00eencorporat de chei \u0219i valori. \u00cen acest caz, devine suficient de simplu s\u0103 implement\u0103m un astfel de API folosind un singur exemplu de Kafka Streams, a\u0219a cum este ilustrat \u00een listingul de mai jos:<\/p>\n

\"Nu
\n
\nIlustra\u021bia 5: utilizarea depozitului \u00eencorporat de chei \u0219i valori pentru ob\u021binerea st\u0103rii pre-calculate a obiectului <\/i><\/p>\n

Actualizarea st\u0103rii unui obiect prin API nu este de asemenea greu de realizat. \u00cen principiu, pentru aceasta trebuie doar s\u0103 cre\u0103m un producer Kafka, iar cu acesta s\u0103 facem o scriere care con\u021bine noua stare. Aceasta asigur\u0103 c\u0103 toate mesajele generate prin API vor fi procesate exact la fel ca cele care provin de la al\u021bi produc\u0103tori (de exemplu, agen\u021bi).<\/p>\n

\"Nu
\n
\nIlustra\u021bia 6: starea unui obiect poate fi stabilit\u0103 prin intermediul unui producer Kafka <\/i><\/p>\n

O mic\u0103 complica\u021bie: Kafka are multe parti\u021bii<\/b><\/p>\n

A\u0219adar, am dorit s\u0103 distribuiem sarcina de procesare \u0219i s\u0103 \u00eembun\u0103t\u0103\u021bim disponibilitatea, oferind pentru fiecare scenariu un cluster de microservicii cu stare partajat\u0103. Configurarea a fost mai simpl\u0103 dec\u00e2t ne-am fi imaginat: dup\u0103 ce am configurat toate instan\u021bele pentru a lucra cu acela\u0219i ID de aplica\u021bie (\u0219i cu acelea\u0219i servere de boot), aproape totul s-a realizat automat. De asemenea, am specificat c\u0103 fiecare topic de intrare va consta din mai multe parti\u021bii, pentru ca fiec\u0103rei instan\u021be s\u0103-i fie atribuit un subset al acestor parti\u021bii.<\/p>\n

De asemenea, a\u0219 dori s\u0103 men\u021bionez c\u0103 aici este obi\u0219nuit s\u0103 faci un backup al stoc\u0103rii st\u0103rilor, pentru ca, de exemplu, \u00een caz de recuperare dup\u0103 o defec\u021biune, s\u0103 transferi aceast\u0103 copie pe o alt\u0103 instan\u021b\u0103. Fiecare stocare de stare \u00een Kafka Streams are un topic replicabil creat cu un jurnal de modific\u0103ri (\u00een care se urm\u0103resc actualiz\u0103rile locale). Astfel, Kafka asigur\u0103 constant stocarea st\u0103rilor. Prin urmare, \u00een caz de defec\u021biune a uneia sau alteia dintre instan\u021be, stocarea st\u0103rilor Kafka Streams poate fi recuperat\u0103 rapid pe alt\u0103 instan\u021b\u0103, unde vor fi transferate parti\u021biile corespunz\u0103toare. Testele noastre au ar\u0103tat c\u0103 acest lucru se face \u00een c\u00e2teva secunde, chiar \u0219i atunci c\u00e2nd stocarea con\u021bine milioane de \u00eenregistr\u0103ri.<\/p>\n

Trec\u00e2nd de la un singur microserviciu cu stare partajat\u0103 la un cluster de microservicii, devine mai pu\u021bin trivial s\u0103 implement\u0103m Get State API. \u00cen noua situa\u021bie, stocarea st\u0103rilor fiec\u0103rui microserviciu con\u021bine doar o parte din imaginea general\u0103 (acele obiecte ai c\u0103ror chei erau mapate la o anumit\u0103 parti\u021bie). A trebuit s\u0103 determin\u0103m pe ce instan\u021b\u0103 se afla starea obiectului dorit, \u0219i am f\u0103cut acest lucru pe baza metadatelor fluxurilor, a\u0219a cum este prezentat mai jos:<\/p>\n

\"Nu
\n
\nIlustrarea 7: folosind metadatele fluxurilor, determin\u0103m de pe ce instan\u021b\u0103 s\u0103 cerem starea obiectului necesar; o abordare similar\u0103 a fost utilizat\u0103 cu GET ALL API<\/i><\/p>\n

Concluzii principale<\/b><\/p>\n

Stoc\u0103rile st\u0103rilor \u00een Kafka Streams pot de facto servi ca o baz\u0103 de date distribuit\u0103, <\/p>\n