a mai adatfeldolgozó architektúrákban az Apache Kafkát gyakran használják az ingress szakaszban. Általában ez a lépés a bejövő üzenetadatok gazdagítására és szűrésére szolgál; azonban az adatfeldolgozási folyamat későbbi szakaszáig nem lehet interaktív lekérdezéseket készíteni. Ennek oka az, hogy bár a Kafka-téma minden üzenete alapértelmezés szerint megmarad, egyelőre nem áll rendelkezésre olyan mechanizmus, amely lehetővé tenné egy adott üzenet gyors keresését egy témában.

Mindazonáltal az új adatok lekérdezésének lehetősége a folyamat ezen korai szakaszában elkerülné a hagyományos feldolgozási folyamatok késedelmét, amelyek általában hosszú távú kötegelt előfeldolgozási lépéseket tartalmaznak, és a végfelhasználók számára szinte azonnali hozzáférést biztosítanának a bejövő adatokhoz.

az adatfeldolgozó alkalmazások Kafka-val történő felépítéséhez a Kafka Streams könyvtárat, amelyet a Kafka projekt részeként tartanak fenn, általánosan használják az adattranszformációk és elemzések meghatározására. A Kafka Stream egyik fontos jellemzője az állami áruházak, amelyek egy gyors helyi kulcs-érték tároló absztrakcióját kínálják, amely olvasható és írható, amikor az üzeneteket Kafka Stream-ekkel dolgozzák fel. Ezeket a kulcsérték-tárolókat folyamatosan meg lehet tölteni egy Kafka-téma új üzeneteivel egy megfelelő adatfolyam-processzor meghatározásával, így most már lehetséges az üzenetek gyors lekérése az alapul szolgáló témából.

ennek a Kafka Streams funkciónak a tetejére építve létrehozunk egy egységes REST API-t, amely egyetlen lekérdezési végpontot biztosít egy adott Kafka témához.

összefoglalva, a Kafka Streams processzorok állami tárolókkal és HTTP szerverekkel való kombinálása hatékonyan bármilyen Kafka témát gyorsan csak olvasható kulcs-érték tárolóvá változtathat.

A Kafka Streams könyvtárként épül fel, amely beágyazható egy önálló Java vagy Scala alkalmazásba. Lehetővé teszi a fejlesztők számára, hogy olyan adatfolyam-processzorokat határozzanak meg, amelyek adatátalakításokat vagy összesítéseket hajtanak végre a Kafka üzeneteken, biztosítva, hogy minden bemeneti üzenet pontosan egyszer kerüljön feldolgozásra. A Java Stream API által inspirált Kafka Streams DSL használatával a stream processzorok és az állami tárolók rugalmasan láncolhatók.

Továbbá, amikor egy Kafka stream-alapú alkalmazás több példányát indítja el, a folyamatok automatikusan egy terheléselosztó, nagy rendelkezésre állású feldolgozási klasztert alkotnak anélkül, hogy a Kafka-tól eltérő külső rendszerektől függnének.

az állami áruházakat alkalmazó Kafka Streams alkalmazás architektúrájának szemléltetésére képzelje el a következő forgatókönyvet: vasúttársaságként minden alkalommal, amikor egy ügyfél foglal egy utazást a weboldalunkon, egy új üzenet kerül beillesztésre az ügyfélazonosítóból és az időbélyegből a Kafka témájába. Pontosabban, egy vagy több Kafka gyártó beilleszti ezeket az üzeneteket egy kilenc partícióval rendelkező témába. Mivel az ügyfélazonosítót választják kulcsként minden üzenethez,az adott ügyfélhez tartozó adatok mindig a téma ugyanazon partíciójába kerülnek. Implicit módon a Kafka-gyártók a DefaultPartitioner parancsot használják üzenetek hozzárendelésére a partíciókhoz.

most tegyük fel, hogy van egy Kafka Streams alkalmazásunk, amely beolvassa a téma üzeneteit, és egy állami áruházban tartja őket. Mivel az egyes üzenetek kulcsa csak az ügyfélazonosítóból áll, az állami áruház megfelelő értéke mindig az ügyfél legutóbbi foglalásának időbélyege lesz. A párhuzamos feldolgozás maximális fokának elérése érdekében akár kilenc példányt is elindíthatunk a Kafka Streams alkalmazásból. Ebben az esetben minden alkalmazáspéldányhoz pontosan az egyik témakört kell hozzárendelni. Minden bemeneti partícióhoz a Kafka Streams külön állami tárolót hoz létre, amely viszont csak az adott partícióhoz tartozó ügyfelek adatait tárolja.

a kapott alkalmazás architektúrát az alábbi ábra szemlélteti.

4-9 partíciókért felelős Kafka stream processzorok kimaradnak ebben az ábrán. A szaggatott nyilak azt jelzik, hogy a partícióban lévő új üzenetek további adatfolyam-processzorokhoz és azok állami tárolóihoz is továbbítódnak, lehetővé téve a gyors meghibásodást, ha az elsődlegesen hozzárendelt processzor meghibásodik.

lehetőség van a memóriában vagy az állandó állapotban lévő tárolók használatára az alkalmazásban. A memóriában lévő állapotú üzletek műveletei még gyorsabbak a perzisztens változathoz képest, amely belsőleg egy RocksDB áruházat használ. Másrészt a tartós állapotú tárolók gyorsabban visszaállíthatók, ha egy Kafka Streams alkalmazás sikertelen, és újra kell indítani. Ezenkívül a tárolónkénti adatmennyiséget nem korlátozza a fő memória mennyisége, ha állandó állapotú tárolókat használ.

forgatókönyvünkben nincs szükség olyan változásnapló-témára, amely rögzíti az írási műveleteket az állami tárolókba: az állami tároló helyreállításához szükséges összes adat beszerezhető az eredeti bemeneti témából.

REST végpont hozzáadása a stream processzorokhoz

az eddig bemutatott architektúrával kilenc állami áruházunk van, amelyek felhasználhatók az adott bemeneti téma partícióhoz tartozó ügyfelek legfrissebb foglalási dátumának lekérésére.

annak érdekében, hogy ezeket az információkat a Kafka Streams processzorokon kívülről is elérhetővé tegyük, minden egyes stream processor alkalmazáspéldányon fel kell tárnunk egy szolgáltatási végpontot, és válaszolnunk kell a Kafka Streams által kezelt belső állapottárolóból érkező bejövő kérésekre.

további követelményként nem várhatjuk el, hogy a kérelmező alkalmazás tudja, melyik Kafka Streams példány felelős jelenleg egy adott ügyfél adatainak feldolgozásáért. Következésképpen minden szolgáltatási végpont felelős a lekérdezés átirányításáért a megfelelő alkalmazáspéldányra, ha az ügyféladatok nem állnak rendelkezésre helyben.

úgy döntöttünk, hogy a szolgáltatás végpontját REST API-ként valósítjuk meg, amelynek előnye, hogy minden HTTP-t támogató kliensből elérhető, és lehetővé teszi az átlátszó terheléselosztó hozzáadását nagyon könnyen.

a KafkaStreams objektum, amely minden Kafka Streams alkalmazásban elérhető, csak olvasható hozzáférést biztosít az összes helyi állami áruházhoz, és meghatározhatja az adott ügyfélazonosítóért felelős alkalmazáspéldányt is. Amikor ezt az objektumot használjuk REST szolgáltatásunk felépítéséhez, az architektúra a következőképpen néz ki:

összefoglalva, javasolt architektúránk a Kafka témákat használja az üzenetadatok megbízható tárolására nyugalmi állapotban, és fenntartja az adatok második ábrázolását az állami üzletekben a gyors lekérdezések támogatása érdekében.

az alkalmazás skálázhatóságának biztosítása

a skálázható alkalmazás megszerzéséhez biztosítanunk kell, hogy a feldolgozási terhelés egyformán kiegyensúlyozott legyen a Kafka Streams alkalmazás minden példányán. Az egyes adatfolyam-processzorok terhelése a kezelendő adatok és lekérdezések mennyiségétől függ.

pontosabban, fontos, hogy válasszon egy particionálási sémát a Kafka témához úgy, hogy a bejövő üzenetek és lekérdezések terhelése kiegyensúlyozott legyen az összes partíción, következésképpen az összes folyamprocesszoron is.

Ha a memóriában tárolt állapotokat kell használni, a témakörben lévő partíciók számának elég nagynak kell lennie ahhoz, hogy minden adatfolyam-processzor képes legyen megtartani egy partíció adatmennyiségét a fő memóriában. Ne feledje, hogy hiba esetén a stream processzornak akár két partíciót is el kell tartania a fő memóriában.

a folyamprocesszor hibáinak figyelembevételéhez a készenléti replikák számát A num.standby.replicas beállítással lehet konfigurálni a Kafka Stream-ekben, amely biztosítja, hogy további adatfolyam-processzorok is feliratkozzanak az adott processzorpartíció üzeneteire. Hiba esetén ezek a processzorok gyorsan átvehetik a lekérdezések megválaszolását, ahelyett, hogy csak a hiba bekövetkezése után kezdenék rekonstruálni az állami tárolót.

végül a kívánt számú stream processzornak illeszkednie kell a rendelkezésre álló hardverhez. Minden stream processor alkalmazáspéldányhoz legalább egy CPU magot kell fenntartani. Többmagos rendszereken lehetőség van az adatfolyam-szálak számának növelésére alkalmazáspéldányonként, ami csökkenti a CPU-magonként külön Java alkalmazás indításával felmerülő költségeket.