Stwórz i uzyskaj dostęp do tabeli aktualizacji w Kstreams - apache-kafka, apache-kafka-stream

Chcę uzyskać dostęp do strumienia danych "A", sprawdzić niektóre warunki w stosunku do tabeli "B" i na koniec zaktualizować stan tabeli "B".

(A,B) ====> check conditions in B ====> finally update B

Jak mogę wdrożyć tę samą logikę? strumienie kafek? Z dokumentacji mojej obserwacji wynika, że ​​dladane strumieniowe "A" możemy wykorzystać Kstreams i do aktualizacji tabeli "B" możemy użyć Ktable. Ale tylko operacja dozwolona między kstreams i idable jest złączona i nie aktualizuje możliwego (zamiast tego używa jej tylko jako odnośnika). Tworzy także nowy kstream (którego nie chcę). Krótko mówiąc, chcę uzyskać dostęp do Ktable jako tabeli aktualizującej w odniesieniu do strumienia przychodzącego, więc jak możemy to samo wdrożyć w strumieniach kafka?

Odpowiedzi:

Treść a KTable jest utrzymywany przez API Streams i nie możesz tego zrobić zaktualizuj a KTable bezpośrednio.

Zakładam, że czytasz KTable z tematu; w tym celu zaktualizujesz tabelę, pisząc do tematu, a strumienie Kafki zaktualizują tabelę, czytając z tego tematu. ZA KTable przeczytane z tematu zostało zaprojektowane w celu zawarcia treści tematu.

Oczywiście pisanie do tematu spowoduje aktualizację KTable nie natychmiast, ale z pewnym opóźnieniem (jest to aktualizacja asynchroniczna). Jeśli potrzebujesz aktualizacji synchronizacji, nie możesz jej użyć KTable ale możesz użyć .transform() z dołączonym stanem.