Cześć, Habrzytelnicy! Ta książka jest odpowiednia dla każdego dewelopera, który chce zrozumieć przetwarzanie strumieniowe. Zrozumienie programowania rozproszonego pomoże lepiej poznać Kafka i Kafka Streams. Byłoby miło znać same ramy Kafka, ale to nie jest konieczne: opowiem Wam wszystko, co potrzebne. Doświadczeni programiści Kafka, jak i nowicjusze, dzięki tej książce opanują tworzenie interesujących aplikacji do przetwarzania strumieniowego za pomocą biblioteki Kafka Streams. Programiści Java średniego i wyższego poziomu, już przyzwyczajeni do takich pojęć jak serializacja, nauczą się stosować swoje umiejętności do tworzenia aplikacji Kafka Streams. Kod źródłowy książki napisany jest w Java 8 i intensywnie wykorzystuje składnię wyrażeń lambda Java 8, więc umiejętność pracy z funkcjami lambda (nawet w innym języku programowania) przyda się.
Fragment. 5.3. Agregacja i operacje okienne
W tej sekcji przejdziemy do badania najbardziej obiecujących części Kafka Streams. Jak dotąd przyjrzeliśmy się następującym aspektom Kafka Streams:
- tworzenie topologii przetwarzania;
- używanie stanu w aplikacjach strumieniowych;
- wykonywanie połączeń strumieni danych;
- różnice między strumieniami wydarzeń (KStream) a strumieniami aktualizacji (KTable).
W następnych przykładach połączymy wszystkie te elementy w całość. Ponadto zapoznacie się z operacjami okiennymi — jeszcze jedną wspaniałą możliwością aplikacji strumieniowych. Naszym pierwszym przykładem będzie prosta agregacja.
5.3.1. Agregacja wolumenu sprzedaży akcji według branż przemysłu
Agregacja i grupowanie to niezbędne narzędzia w pracy z danymi strumieniowymi. Badanie pojedynczych wpisów w miarę ich pojawiania się często okazuje się niewystarczające. Aby wydobyć dodatkowe informacje z danych, konieczne jest ich grupowanie i łączenie.
W tym przykładzie przyjmiesz rolę intraday tradera, który musi śledzić wolumeny sprzedaży akcji firm w kilku branżach przemysłu. W szczególności interesuje cię pięć firm z największym wolumenem sprzedaży akcji w każdej z branż przemysłu.
Aby przeprowadzić taką agregację, wymagane będą następujące kroki przetwarzania danych w odpowiedni sposób (mówiąc ogólnie).
- Utwórz źródło na podstawie tematu, publikujące surowe informacje o handlu akcjami. Musimy przekształcić obiekt typu StockTransaction w obiekt typu ShareVolume. Problem polega na tym, że obiekt StockTransaction zawiera metadane sprzedaży, a potrzebne są nam tylko dane o liczbie sprzedawanych akcji.
- Grupuj dane ShareVolume według symboli akcji. Po grupowaniu według symboli można zredukować te dane do pośrednich sum wolumenów sprzedaży akcji. Warto zauważyć, że metoda KStream.groupBy zwraca instancję typu KGroupedStream. Można uzyskać instancję KTable, wywołując następnie metodę KGroupedStream.reduce.
Czym jest interfejs KGroupedStream
Metody KStream.groupBy i KStream.groupByKey zwracają instancję KGroupedStream. KGroupedStream jest pośrednią reprezentacją strumienia zdarzeń po grupowaniu według kluczy. Nie jest przeznaczony do bezpośredniej pracy z nim. Zamiast tego KGroupedStream służy do operacji agregacji, których wynikiem zawsze jest KTable. Ponieważ wynikiem operacji agregacji jest KTable i wykorzystuje się w nich pamięć stanu, możliwe, że nie wszystkie aktualizacje w wyniku są przesyłane dalej w potoku.
Metoda KTable.groupBy zwraca analogiczny KGroupedTable — pośrednią reprezentację strumienia aktualizacji, przegrupowanych według klucza.
Zróbmy małą przerwę i spójrzmy na rys. 5.9, na którym pokazano, co osiągnęliśmy. Ta topologia powinna być Wam już dobrze znana.
Przyjrzyjmy się teraz kodowi dla tej topologii (można go znaleźć w pliku src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (listing 5.2).
Podany kod różni się zwięzłością i dużym zakresem działań wykonywanych w kilku linijkach. W pierwszym parametrze metody builder.stream możecie zauważyć coś nowego: wartość enumeracji AutoOffsetReset.EARLIEST (istnieje również LATEST), określona za pomocą metody Consumed.withOffsetResetPolicy. Dzięki tej enumeracji można wskazać strategię resetowania offsetów dla każdego KStream lub KTable, ma ona pierwszeństwo przed parametrem resetu offsetów z konfiguracji.
GroupByKey i GroupBy
W interfejsie KStream znajdują się dwie metody do grupowania rekordów: GroupByKey i GroupBy. Obie zwracają KGroupedTable, więc może pojawić się naturalne pytanie: jaka jest różnica między nimi i kiedy używać której z nich?
Metoda GroupByKey jest stosowana, gdy klucze w KStream są już niepuste. Co więcej, flaga „wymaga ponownego sekcjonowania” nigdy nie była ustawiana.
Metoda GroupBy zakłada, że zmieniłeś klucze do grupowania, więc flaga ponownego sekcjonowania jest ustawiona na true. Wykonanie po metodzie GroupBy operacji łączenia, agregacji itp. spowoduje automatyczne ponowne sekcjonowanie.
Podsumowanie: zawsze należy, gdy to możliwe, używać GroupByKey, a nie GroupBy.
To, co robią metody mapValues i groupBy, jest jasne, więc przyjrzyjmy się metodzie sum() (można ją znaleźć w pliku src/main/java/bbejeck/model/ShareVolume.java) (listing 5.3).
Metoda ShareVolume.sum zwraca pośrednią sumę wolumenu sprzedaży akcji, a wynik całego ciągu obliczeń stanowi obiekt KTable. Teraz rozumiesz, jaką rolę pełni KTable. Po przyjęciu obiektów ShareVolume w odpowiednim obiekcie KTable przechowywana jest ostatnia aktualizacja. Ważne jest, aby pamiętać, że wszystkie aktualizacje są odzwierciedlane w poprzednim shareVolumeKTable, ale nie wszystkie są wysyłane dalej.
Następnie za pomocą tego KTable wykonujemy agregację (według liczby sprzedanych akcji), aby uzyskać pięć firm z najwyższymi wolumenami sprzedaży akcji w każdej branży. Nasze działania będą podobne do działań przy pierwszej agregacji.
- Wykonaj kolejną operację groupBy w celu grupowania poszczególnych obiektów ShareVolume według branż.
- Przystąp do sumowania obiektów ShareVolume. Tym razem obiekt agregujący stanowi kolejkę o ustalonej wielkości. W takiej kolejce o ustalonej wielkości przechowywane są tylko pięć firm z najwyższymi liczbami sprzedanych akcji.
- Przekształć kolejki z poprzedniego punktu na wartość łańcuchową i zwróć pięć najlepiej sprzedających się akcji według branż.
- Zapisz wyniki w formie tekstowej w temacie.
Na rys. 5.10 przedstawiona jest mapa topologii ruchu danych. Jak widać, drugi cykl przetwarzania jest dość prosty.
Teraz, mając jasno określoną strukturę tego drugiego cyklu przetwarzania, można przejść do jego kodu źródłowego (znajdziesz go w pliku src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (listing 5.4).
W tym inicjatorze znajduje się zmienna fixedQueue. To obiekt użytkownika — adapter dla java.util.TreeSet, który jest używany do śledzenia N największych wyników w kolejności malejącej liczby sprzedanych akcji.
Już spotkałeś się z wywołaniami groupBy i mapValues, więc nie będziemy się na nich zatrzymywać (wywołujemy metodę KTable.toStream, ponieważ metoda KTable.print jest uznawana za przestarzałą). Ale jeszcze nie widziałeś KTable-wersji metody aggregate(), więc poświęcimy chwilę na jej omówienie.
Jak pamiętasz, KTable różni się tym, że rekordy z tymi samymi kluczami są traktowane jako aktualizacje. KTable zastępuje stary rekord nowym. Agregacja odbywa się w podobny sposób: agregowane są najnowsze rekordy z tym samym kluczem. Po otrzymaniu rekordu jest on dodawany do instancji klasy FixedSizePriorityQueue za pomocą sumatora (drugi parametr w wywołaniu metody aggregate), ale jeśli już istnieje inny rekord z tym samym kluczem, stary rekord jest usuwany za pomocą reduktora (trzeci parametr w wywołaniu metody aggregate).
To wszystko oznacza, że nasz agregator, FixedSizePriorityQueue, wcale nie agreguje wszystkich wartości z tym samym kluczem, tylko przechowuje przesuwającą się sumę ilości N najlepiej sprzedających się rodzajów akcji. Każdy przychodzący rekord zawiera łączną liczbę sprzedanych dotychczas akcji. KTable dostarczy ci informacji o tym, jakie akcje firm sprzedają się obecnie najlepiej, a przesuwająca agregacja każdego z aktualizacji nie jest konieczna.
Nauczyliśmy się robić dwie ważne rzeczy:
- grupować wartości w KTable według wspólnego klucza;
- wykonywać na tych zgromadzonych wartościach przydatne operacje, takie jak redukcja i agregacja.
Umiejętność wykonywania tych operacji jest ważna dla zrozumienia sensu danych przepływających przez aplikację Kafka Streams i ustalenia, jakie informacje niosą ze sobą.
Połączyliśmy w jedną całość niektóre z kluczowych pojęć, omawianych wcześniej w tej książce. W rozdziale 4 opowiedzieliśmy, jak ważne jest lokalne, odporne na awarie stanie dla aplikacji strumieniowych. Pierwszy przykład z tego rozdziału pokazał, dlaczego lokalne stanie jest tak istotne — umożliwia śledzenie, jakie informacje już zostały zobaczone. Lokalne podejście pozwala uniknąć opóźnień sieciowych, co czyni aplikację bardziej wydajną i odporną na błędy.
Podczas wykonywania jakiejkolwiek operacji agregacji lub redukcji należy podać nazwę magazynu stanu. Operacje redukcji i agregacji zwracają instancję KTable, a KTable korzysta z magazynu stanu do zastępowania starych wyników nowymi. Jak zauważyłeś, nie wszystkie aktualizacje są wysyłane dalej w łańcuchu, co jest ważne, ponieważ operacje agregacji są przeznaczone do uzyskania informacji końcowych. Jeśli nie będziemy stosować lokalnego stanu, KTable będzie przesyłać wszystkie wyniki agregacji i redukcji.
Teraz przyjrzymy się wykonaniu operacji, takich jak agregacja, w określonym przedziale czasowym — tak zwanych operacjach okna (windowing operations).
5.3.2. Operacje okna
W poprzedniej sekcji zapoznaliśmy się z „przesuwającą się” redukcją i agregacją. Aplikacja przeprowadzała ciągłą redukcję wolumenu sprzedaży akcji, a następnie agregację pięciu najbardziej sprzedawanych akcji na giełdzie.
Czasami takie ciągłe agregacje i redukcje wyników są niezbędne. A czasami trzeba wykonać operacje tylko w obrębie określonego przedziału czasowego. Na przykład, obliczyć, ile transakcji giełdowych z akcjami konkretnej firmy miało miejsce w ciągu ostatnich 10 minut. Lub ile użytkowników kliknęło w nowy baner reklamowy w ciągu ostatnich 15 minut. Aplikacja może wykonywać takie operacje wielokrotnie, ale z wynikami odnoszącymi się tylko do określonych przedziałów czasowych (okien czasowych).
Liczenie transakcji giełdowych według klienta
W następnym przykładzie zajmiemy się śledzeniem transakcji giełdowych dla kilku traderów — zarówno dużych organizacji, jak i bystrych inwestorów indywidualnych.
Istnieją dwie możliwe przyczyny takiego śledzenia. Jedną z nich jest potrzeba wiedzy o tym, co kupują/sprzedają liderzy rynku. Jeśli ci duzi gracze i doświadczeni inwestorzy dostrzegają otwierające się przed nimi możliwości, sensowne jest podążanie za ich strategią. Drugą przyczyną jest chęć dostrzegania jakichkolwiek możliwych oznak nielegalnych transakcji z wykorzystaniem informacji poufnych. W tym celu będziesz musiał przeanalizować korelację dużych wzrostów sprzedaży z ważnymi komunikatami prasowymi.
Takie śledzenie składa się z takich etapów jak:
- tworzenie strumienia do odczytu z tematu stock-transactions;
- grupowanie przychodzących wpisów według identyfikatora nabywcy i symbolu giełdowego akcji. Wywołanie metody groupBy zwraca instancję klasy KGroupedStream;
- zwracanie metodą KGroupedStream.windowedBy strumienia danych ograniczonego oknem czasowym, co pozwala na wykonywanie agregacji okien. W zależności od typu okna zwracane jest albo TimeWindowedKStream, albo SessionWindowedKStream;
- zliczanie transakcji dla operacji agregacji. Okienny strumień danych określa, czy konkretny wpis jest brany pod uwagę w tym zliczeniu;
- zapisywanie wyników w temacie lub wyprowadzanie ich na konsolę podczas rozwoju.
Topologia tej aplikacji jest prosta, ale wizualizacja jej obrazu nie zaszkodzi. Przyjrzyjmy się rys. 5.11.
Następnie omówimy funkcjonalność operacji okiennych oraz odpowiedni kod.
Typy okien
W Kafka Streams istnieją trzy typy okien:
- sesyjne;
- w „turlających się” (tumbling);
- w „przesuwających się/„skaczących” (sliding/hopping).
Wybór zależy od wymagań biznesowych. „Turlające się” i „skaczące” okna są ograniczone czasowo, podczas gdy ograniczenia okien sesyjnych związane są z działaniami użytkowników — długość sesji(-y) określa się wyłącznie na podstawie aktywności użytkownika. Najważniejsze jest, aby pamiętać, że wszystkie typy okien opierają się na znacznikach daty/czasu wpisów, a nie na czasie systemowym.
Następnie zaimplementujemy naszą topologię z każdym z typów okien. Pełny kod zostanie podany tylko w pierwszym przykładzie, dla innych typów okien nic się nie zmieni, poza typem operacji okiennej.
Okna sesyjne
Okna sesyjne znacząco różnią się od wszystkich innych typów okien. Ograniczają się nie tyle czasowo, co aktywnością użytkownika (lub aktywnością jednostki, którą chciałbyś śledzić). Okna sesyjne są wyznaczane okresem bezczynności.
Rysunek 5.12 ilustruje pojęcie okien sesyjnych. Mniejsza sesja łączy się z sesją po lewej stronie. Sesja po prawej stronie będzie osobna, ponieważ następuje po długim okresie bezczynności. Okna sesyjne opierają się na działaniach użytkowników, ale stosują znaczniki daty/czasu z zapisów, aby określić, do której sesji odnosi się dany zapis.
Wykorzystanie okien sesyjnych do śledzenia transakcji giełdowych.
Wykorzystamy okna sesyjne do uchwycenia informacji o transakcjach giełdowych. Implementacja okien sesyjnych jest przedstawiona w listing 5.5 (który można znaleźć w pliku src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).
Większość operacji w tej topologii już znasz, więc nie ma potrzeby ponownego ich omawiania. Jednak są też tutaj pewne nowe elementy, które teraz omówimy.
Przy każdej operacji groupBy zazwyczaj wykonywana jest jakaś operacja agregacji (agregacja, redukcja lub zliczanie). Można wykonać zarówno agregację skumulowaną z narastającym wynikiem, jak i agregację okienną, w której uwzględnia się zapisy w określonym czasie.
Kod z listingu 5.5 wykonuje zliczanie transakcji w obrębie okien sesyjnych. Na rys. 5.13 te działania są analizowane krok po kroku.
Za pomocą wywołania windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) tworzymy okno sesyjne z okresem bezczynności 20 sekund oraz okresem przechowywania 15 minut. Okres bezczynności 20 sekund oznacza, że aplikacja uwzględni każdy zapis, który wpłynie w ciągu 20 sekund od zakończenia lub rozpoczęcia bieżącej sesji w obecnej (aktywnej) sesji.
Następnie określamy, jaką operację agregacji należy wykonać w oknie sesji — w tym przypadku count. Jeśli przychodzący zapis wykracza poza granice interwału bezczynności (z dowolnej strony od znacznika daty/czasu), aplikacja tworzy nową sesję. Interwał przechowywania oznacza utrzymywanie sesji przez określony czas i dopuszcza spóźnione dane, które wychodzą poza okres bezczynności sesji, ale wciąż mogą być dołączane. Ponadto początek i koniec nowej sesji, która powstała w wyniku połączenia, odpowiadają najwcześniejszemu i najpóźniejszemu znacznikowi daty/czasu.
Przyjrzyjmy się kilku zapisom z metody count, aby zobaczyć, jak działają sesje (tab. 5.1).
Gdy przychodzą zapisy, szukamy już istniejących sesji z tym samym kluczem, których czas zakończenia jest wcześniejszy niż aktualny znacznik daty/czasu — interwał bezczynności oraz czas rozpoczęcia większy niż aktualny znacznik daty/czasu + interwał bezczynności. Uwzględniając to, cztery zapisy z tab. 5.1 łączą się w jedną sesję w następujący sposób.
1. Pierwszy przychodzi zapis 1, więc czas rozpoczęcia jest równy czasowi zakończenia i wynosi 00:00:00.
2. Następnie przychodzi zapis 2, i szukamy sesji kończących się nie wcześniej niż 23:59:55 i rozpoczynających się nie później niż 00:00:35. Znajdujemy zapis 1 i łączymy sesje 1 i 2. Bierzemy czas rozpoczęcia sesji 1 (wcześniejszy) oraz czas zakończenia sesji 2 (późniejszy), więc nasza nowa sesja zaczyna się o 00:00:00 i kończy o 00:00:15.
3. Przychodzi zapis 3, szukamy sesji między 00:00:30 a 00:01:10 i nie znajdujemy żadnej. Dodajemy drugą sesję dla klucza 123-345-654,FFBE, rozpoczynającą się i kończącą o 00:00:50.
4. Przychodzi zapis 4, i szukamy sesji między 23:59:45 a 00:00:25. Tym razem znajdujemy obie sesje — 1 i 2. Wszystkie trzy sesje łączą się w jedną, z czasem rozpoczęcia 00:00:00 i czasem zakończenia 00:00:15.
Z tego, co zostało omówione w tej sekcji, warto zapamiętać następujące istotne szczegóły:
- sesje to nie okna o stałej wielkości. Czas trwania sesji jest określany przez aktywność w ramach danego interwału czasowego;
- znaczniki daty/czasu w danych określają, czy zdarzenie mieści się w istniejącej sesji, czy w interwale bezczynności.
Następnie omówimy następną odmianę okien — «przewracające» okna.
«Przewracające» okna
„Przewracające się” okna (tumbling) rejestrują zdarzenia, które wpadają w określony interwał czasowy. Wyobraź sobie, że musisz uchwycić wszystkie transakcje giełdowe jakiejś firmy co 20 sekund, więc zbierasz wszystkie zdarzenia w tym czasie. Po zakończeniu 20-sekundowego interwału okno „przewraca się” i przechodzi do nowego 20-sekundowego okresu obserwacji. Rysunek 5.14 ilustruje tę sytuację.
Jak widać, wszystkie zdarzenia, które miały miejsce w ostatnich 20 sekundach, są zawarte w oknie. Po zakończeniu tego okresu tworzone jest nowe okno.
W listing 5.6 znajduje się kod demonstrujący użycie „przewracających się” okien do uchwycenia transakcji giełdowych co 20 sekund (można go znaleźć w pliku src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).
Dzięki tej niewielkiej zmianie wywołania metody TimeWindows.of można używać „przewracającego się” okna. W tym przykładzie nie ma wywołania metody until(), w związku z czym zostanie użyty domyślny interwał przechowywania, wynoszący 24 godziny.
Wreszcie nadszedł czas, aby przejść do ostatniego z typów okien — „skaczących” (hopping) okien.
Okna skaczące („hopping”)
Okna skaczące („sliding/hopping”) są podobne do „przewracających się”, ale istnieje pewna różnica. Okna skaczące nie czekają na zakończenie interwału czasu przed utworzeniem nowego okna do przetwarzania niedawnych zdarzeń. Rozpoczynają nowe obliczenia po interwale oczekiwania, który jest krótszy niż czas trwania okna.
Aby zobrazować różnice między „przewracającymi się” a „skaczącymi” oknami, wróćmy do przykładu z liczeniem transakcji giełdowych. Naszym celem nadal jest liczenie liczby transakcji, ale nie chcemy czekać całego okresu czasu przed zaktualizowaniem licznika. Zamiast tego będziemy aktualizować licznik w krótszych odstępach czasu. Na przykład, będziemy nadal liczyć liczbę transakcji co 20 sekund, ale będziemy aktualizować licznik co 5 sekund, jak pokazano na rys. 5.15. W rezultacie będziemy mieć trzy okna wynikowe z nakładającymi się danymi.
W listing 5.7 znajduje się kod do zadania skaczących okien (można go znaleźć w pliku src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).
Okno „przewracające się” można przekształcić w „skaczące” przez dodanie wywołania metody advanceBy(). W przedstawionym przykładzie interwał zapisywania wynosi 15 minut.
W tej sekcji zobaczyłeś, jak ograniczać wyniki agregacji przez czasowe okna. Chciałbym, abyś zapamiętał z tej sekcji trzy rzeczy:
- rozmiar okien sesyjnych jest ograniczony nie przez przedział czasu, ale przez aktywność użytkowników;
- „przewracające się” okna dają wgląd w wydarzenia w danym okresie czasu;
- czas trwania „skaczących” okien jest stały, ale są one często aktualizowane i mogą zawierać w każdym oknie nakładające się rekordy.
Dalej dowiemy się, jak przekształcić KTable z powrotem w KStream do łączenia.
5.3.3. Łączenie obiektów KStream i KTable
W rozdziale 4 omawialiśmy łączenie dwóch obiektów KStream. Teraz nauczymy się łączyć KTable i KStream. Może być to potrzebne z następującego prostego powodu. KStream to strumień rekordów, a KTable to strumień aktualizacji rekordów, ale czasami może być potrzebne dodanie dodatkowego kontekstu do strumienia rekordów za pomocą aktualizacji z KTable.
Weźmy dane o liczbie transakcji giełdowych i połączmy je z informacjami o giełdowych wiadomościach dotyczących odpowiednich branż przemysłowych. Oto, co trzeba zrobić, aby to osiągnąć, biorąc pod uwagę już istniejący kod.
- Przekształć obiekt KTable z danymi o liczbie transakcji giełdowych w KStream, następnie zamień klucz na klucz oznaczający branżę przemysłową odpowiadającą temu symbolowi akcji.
- Utwórz obiekt KTable, który będzie odczytywał dane z tematu giełdowych wiadomości. Ten nowy KTable będzie kategoryzowany według branż przemysłowych.
- Połącz aktualizacje wiadomości z informacjami o liczbie transakcji giełdowych według branż przemysłowych.
Teraz przyjrzyjmy się, jak zrealizować ten plan działania.
Przekształcanie KTable w KStream
Aby przekształcić KTable w KStream, należy zrobić następujące kroki.
- Wywołać metodę KTable.toStream().
- Za pomocą wywołania metody KStream.map zamień klucz na nazwę branży, a następnie wyodrębnij z instancji Windowed obiekt TransactionSummary.
Połączymy te operacje w łańcuch w następujący sposób (kod można znaleźć w pliku src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.8).
Ponieważ wykonujemy operację KStream.map, ponowne sekcjonowanie dla zwróconej instancji KStream odbywa się automatycznie podczas jej użycia w połączeniu.
Zakończyliśmy proces przekształcania, teraz musimy stworzyć obiekt KTable do odczytu wiadomości giełdowych.
Tworzenie KTable dla wiadomości giełdowych
Na szczęście, aby stworzyć obiekt KTable, wystarczy jedna linia kodu (ten kod można znaleźć w pliku src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.9).
Należy zauważyć, że nie trzeba podawać żadnych obiektów Serde, ponieważ w ustawieniach używane są stringowe Serde. Dzięki zastosowaniu enumeracji EARLIEST tabela jest wypełniana zapisami na początku.
Teraz możemy przejść do ostatniego kroku — połączenia.
Łączenie aktualizacji wiadomości z danymi o liczbie transakcji
Tworzenie połączenia nie sprawia trudności. Skorzystamy z lewego połączenia na wypadek, gdyby nie było wiadomości giełdowych dla danej branży (potrzebny kod można znaleźć w pliku src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (listing 5.10).
Operator leftJoin jest dość prosty. W przeciwieństwie do połączeń z rozdziału 4, metoda JoinWindow nie jest używana, ponieważ podczas wykonywania połączenia KStream-KTable dla każdego klucza w KTable znajduje się tylko jeden zapis. To połączenie nie jest ograniczone czasowo: zapis jest albo w KTable, albo go nie ma. Główna konkluzja: za pomocą obiektów KTable można wzbogacać KStream danymi referencyjnymi rzadziej aktualizowanymi.
A teraz przyjrzymy się bardziej wydajnemu sposobowi wzbogacania zdarzeń z KStream.
5.3.4. Obiekty GlobalKTable
Jak już rozumiesz, istnieje potrzeba wzbogacania strumieni zdarzeń lub dodawania do nich kontekstu. W rozdziale 4 widziałeś połączenia dwóch obiektów KStream, a w poprzednim rozdziale — połączenie KStream i KTable. We wszystkich tych przypadkach konieczne jest ponowne sekcjonowanie strumienia danych przy mapowaniu kluczy na nowy typ lub wartość. Czasami ponowne sekcjonowanie jest wykonywane jawnie, a czasami Kafka Streams robi to automatycznie. Ponowne sekcjonowanie jest konieczne, ponieważ klucze się zmieniły i zapisy powinny znaleźć się w nowych sekcjach, w przeciwnym razie połączenie stanie się niemożliwe (to było omówione w rozdziale 4, w punkcie 'Ponowne sekcjonowanie danych' sekcji 4.2.4).
Re-sekcjonowanie wiąże się z kosztami
Re-sekcjonowanie wymaga wydatków — dodatkowych zasobów na tworzenie pośrednich tematów, przechowywanie powtarzających się danych w jeszcze jednym temacie; oznacza to również zwiększenie opóźnienia w wyniku zapisu i odczytu z tego tematu. Dodatkowo, w przypadku konieczności wykonania połączenia na więcej niż jednym aspekcie lub wymiarze, należy zorganizować połączenia w łańcuch, zmapować rekordy z nowymi kluczami i ponownie przeprowadzić proces re-sekcjonowania.
Połączenie z zestawami danych mniejszych rozmiarów
W niektórych przypadkach objętość danych referencyjnych, z którymi planowane jest połączenie, jest stosunkowo niewielka, więc ich pełne kopie mogą zmieścić się lokalnie na każdym z węzłów. Do takich sytuacji w Kafka Streams przewidziany jest klasa GlobalKTable.
Instancje GlobalKTable są unikalne, ponieważ aplikacja replikująca wszystkie dane na każdym z węzłów. A ponieważ na każdym z węzłów znajdują się wszystkie dane, nie ma potrzeby sekcjonowania strumienia zdarzeń według klucza danych referencyjnych, aby był dostępny we wszystkich sekcjach. Dzięki obiektom GlobalKTable można również realizować połączenia bezkluczowe. Wróćmy do jednego z poprzednich przykładów, aby zobaczyć tę możliwość.
Połączenie obiektów KStream z obiektami GlobalKTable
W podrozdziale 5.3.2 przeprowadziliśmy agregację okienkową transakcji giełdowych według klientów. Wyniki tej agregacji wyglądały mniej więcej tak:
{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16Chociaż te wyniki spełniły postawiony cel, byłoby wygodniej, gdyby wyświetlane były także imię klienta i pełna nazwa firmy. Aby dodać imię kupującego i nazwę firmy, można wykonać zwykłe połączenia, ale wymagałoby to dwóch mapowań kluczy i ponownego sekcjonowania. Dzięki GlobalKTable można uniknąć kosztów związanych z takimi operacjami.
W tym celu skorzystamy z obiektu countStream z listing 5.11 (odpowiedni kod można znaleźć w pliku src/main/java/bbejeck/chapter_5/GlobalKTableExample.java), łącząc go z dwoma obiektami GlobalKTable.
Już to omawialiśmy wcześniej, więc nie będę się powtarzać. Zwrócę jednak uwagę, że kod w funkcji toStream().map jest zorganizowany w obiekt-funkcję dla lepszej czytelności, zamiast być wbudowaną wyrażeniem lambda.
Następny krok to deklaracja dwóch instancji GlobalKTable (przykładowy kod znajduje się w pliku src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (fragment 5.12).
Zauważ, że nazwy tematów są opisywane za pomocą typów wyliczanych.
Teraz, gdy przygotowaliśmy wszystkie komponenty, pozostaje napisać kod do połączenia (który można znaleźć w pliku src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (fragment 5.13).
Chociaż w tym kodzie znajdują się dwa połączenia, są one zorganizowane w formie łańcucha, ponieważ żaden z ich wyników nie jest używany osobno. Wyniki są wyprowadzane na końcu całej operacji.
Podczas uruchamiania powyższej operacji połączenia otrzymasz wyniki w następującym formacie:
{customer='Barney, Smith' company="Exxon", transactions= 17}Istota się nie zmieniła, ale te wyniki wyglądają bardziej zrozumiale.
Jeśli uwzględnimy rozdział 4, już widziałeś kilka typów połączeń w działaniu. Są one wymienione w tabeli 5.2. Ta tabela odzwierciedla możliwości połączeń aktualne dla wersji 1.0.0 Kafka Streams; w przyszłych wydaniach mogą się zmienić.
Na zakończenie przypomnę najważniejsze: możesz łączyć strumienie zdarzeń (KStream) i strumienie aktualizacji (KTable) przy użyciu lokalnego stanu. Ponadto, jeśli rozmiar danych referencyjnych nie jest zbyt duży, można skorzystać z obiektu GlobalKTable. GlobalKTable replikują wszystkie sekcje na każdym z węzłów aplikacji Kafka Streams, zapewniając dostępność wszystkich danych, niezależnie od tego, do której sekcji należy klucz.
Następnie zobaczymy możliwość Kafka Streams, która pozwala obserwować zmiany stanu bez konsumowania danych z tematu Kafka.
5.3.5. Stan dostępny dla zapytań
Już wykonałeś kilka operacji z wykorzystaniem stanu i zawsze wyświetlaliśmy wyniki w konsoli (do celów rozwojowych) lub zapisywaliśmy je do tematu (dla celów eksploatacyjnych). Podczas zapisywania wyników do tematu konieczne jest użycie konsumenta Kafka do ich przeglądania.
Czytanie danych z tych tematów można uznać za rodzaj materializowanych widoków (materialized views). Dla naszych potrzeb możemy użyć definicji materializowanego widoku z „Wikipedii”: „…fizyczny obiekt bazy danych, który zawiera wyniki wykonania zapytania. Na przykład może to być lokalna kopia danych zdalnych, lub podzbiór wierszy i/lub kolumn tabeli lub wyników połączenia, lub tabela przestawna uzyskana za pomocą agregacji” (https://en.wikipedia.org/wiki/Materialized_view).
Kafka Streams umożliwia również wykonywanie interaktywnych zapytań (interactive queries) do magazynów stanu, co daje możliwość bezpośredniego odczytu tych materializowanych widoków. Ważne jest, aby zauważyć, że zapytanie do magazynu stanu ma charakter operacji „tylko do odczytu”. Dzięki temu możesz nie martwić się przypadkowym wprowadzeniem niezgodności w stanie podczas przetwarzania danych przez aplikację.
Możliwość bezpośrednich zapytań do magazynów stanu ma ogromne znaczenie. Oznacza to, że można tworzyć aplikacje — pulpity nawigacyjne, bez potrzeby najpierw pozyskiwania danych od konsumenta Kafka. Zwiększa ona również wydajność aplikacji, ponieważ nie ma potrzeby ponownego zapisywania danych:
- dzięki lokalności danych można szybko uzyskać do nich dostęp;
- eliminuje się duplikację danych, ponieważ nie są one zapisywane w zewnętrznej pamięci.
Najważniejsze, co chciałbym, abyś zapamiętał: można bezpośrednio wykonywać zapytania do stanu z aplikacji. Nie można przecenić możliwości, które to daje. Zamiast konsumować dane z Kafka i zapisywać rekordy w bazie danych dla aplikacji, można wykonywać zapytania do magazynów stanu z tym samym wynikiem. Bezpośrednie zapytania do magazynów stanu oznaczają mniej kodu (brak konsumenta) i mniej oprogramowania (brak potrzeby posiadania tabeli bazy danych do przechowywania wyników).
W tej rozdziale poruszyliśmy dużą ilość informacji, dlatego na chwilę zaprzestaniemy omawiania interaktywnych zapytań do magazynów stanu. Ale nie martwcie się: w rozdziale 9 stworzymy prostą aplikację — pulpit nawigacyjny z interaktywnymi zapytaniami. W celu demonstracji interaktywnych zapytań i możliwości ich dodawania do aplikacji Kafka Streams skorzystamy z niektórych przykładów z tej i poprzedniej rozdziału.
Podsumowanie
- Obiekty KStream reprezentują strumienie zdarzeń, porównywalne z wstawieniami do bazy danych. Obiekty KTable reprezentują strumienie aktualizacji, bardziej przypominają aktualizacje w bazie danych. Rozmiar obiektu KTable nie rośnie, stare rekordy są zastępowane nowymi.
- Obiekty KTable są niezbędne do operacji agregacyjnych.
- Dzięki operacjom okiennym można podzielić agregowane dane na przedziały czasowe.
- Dzięki obiektom GlobalKTable możemy uzyskać dostęp do danych referencyjnych w dowolnym punkcie aplikacji, niezależnie od podziału na sekcje.
- Możliwe są połączenia między obiektami KStream, KTable i GlobalKTable.
Dotychczas koncentrowaliśmy się na tworzeniu aplikacji Kafka Streams przy użyciu wysokopoziomowego DSL KStream. Chociaż podejście wysokopoziomowe pozwala na pisanie czystych i zwięzłych programów, jego wykorzystanie wiąże się z pewnym kompromisem. Praca z DSL KStream oznacza większą zwięzłość kodu kosztem mniejszej kontroli. W następnym rozdziale przyjrzymy się niskopoziomowemu API węzłów przetwarzających i spróbujemy innych kompromisów. Programy będą dłuższe niż dotychczas, ale zyskamy możliwość tworzenia praktycznie dowolnego węzła przetwarzającego, który może być nam potrzebny.
→ Więcej informacji o książce można znaleźć na
→ Dla użytkowników Habr zniżka 25% przy użyciu kuponu — Kafka Streams
→ Po dokonaniu płatności za papierową wersję książki na e-mail zostanie wysłana wersja elektroniczna.
Źródło: habr.com
