Kafka Log Compaction

05.07.2023 | Dominik Marszałek

Kilka słów wprowadzenia - co to jest Kafka?

Kafka jest to rozproszona platforma strumieniowania zdarzeń. Polega ona na tym, że producenci rozgłaszają zaistnienie pewnego zdarzenia, a konsumenci nasłuchują i pobierają daną wiadomość. Warto tutaj wyjaśnić kilka kluczowych pojęć. Brokerami nazywamy serwery, które zajmują się odbieraniem wiadomości od producentów oraz ich przechowywaniem. Wiadomości są pogrupowane w topiki, a każdy topik składa się z dowolnej liczby partycji. Jest ona fizycznym magazynem danych. Partycje dzielone są na segmenty, czyli mniejsze zestawy eventów, które są przechowywane jako pojedynczy plik na dysku brokera. Każdy nowy segment jest tworzony gdy poprzedni przekroczy zdefiniowany czas lub rozmiar. Każda wiadomość na danej partycji ma swój offset i jest to nic innego jak sekwencyjny identyfikator eventu. Tak uporządkowane struktury danych zwane są logami. Producenci mogą rozgłaszać eventy, które składają się z jakieś wartości lub mogą też wysyłać parę: klucz i wartość.

 

Cleanup policy - delete

Podczas gdy producenci wysyłają nowe eventy, log dla danej partycji topika coraz bardziej się rozrasta. Dlatego niezbędne jest określenie sposobu oczyszczania danych, które są już skonsumowane i niepotrzebne. Domyślnym podejściem jest log.cleanup.policy=delete, które polega na usuwaniu najstarszych eventów po przekroczeniu zdefiniowanego czasu lub zadeklarowanego maksymalnego rozmiaru. Dzięki temu, konsumenci, w razie swojej chwilowej niedostępności, mają na przykład siedem dni na odczyt wiadomości.

 

Źródło: opracowanie własne 

 

Cleanup policy - compaction

Innym podejściem do oczyszczania danych jest kompaktowanie eventów - log.cleanup.policy=compact. W dużym skrócie, polega to na tym, że kasowane są przestarzałe wiadomości dla tego samego klucza. Innymi słowy, log compaction zapewnia, że Kafka będzie przetrzymywać przynajmniej ostatnią znaną wartość dla każdego klucza wiadomości.

 

Zastosowanie

Kompaktowość zmienia niejako zastosowanie takiego topika. Pozwala nie patrzeć na taki log jako strumień danych z pewną retencją, ale jako magazyn trzymający najbardziej aktualne wartości dla danych kluczy.

Dobrym przykładem zastosowania jest ksqlDB. Jest to baza danych, która opiera się o strumieniowanie zdarzeń w czasie rzeczywistym. Jest zbudowana na Apache Kafka oraz Kafka Streams. Tabele w tej bazie działają w oparciu o eventy klucz wartość. Dzięki temu, w przypadku kilku eventów dzielących jeden klucz, wybierany jest najnowszy - z najwyższym offsetem. Pod spodem jest to realizowane poprzez log compaction.

Kolejne zastosowanie kompaktowania można zaobserwować w frameworku Samza. Samza jest to platforma przetwarzania strumieniowego w czasie rzeczywistym. Log compaction wspiera changelog topik Samzy zapewniając, że ten changelog nie będzie niepotrzebnie się rozszerzał. W przypadku nadpisywania tej samej wartości kilka razy, log compaction zachowa jedynie ostatnią, najbardziej aktualną wartość.

Innym przykładem użycia log compation jest zapewnienie tolerancji na błędy w aplikacji korzystającej z persystencji in-memory. Dzięki przechowywaniu w logu kompaktowym kafki snapshotu rekordów z tej aplikacji, jest możliwe dość łatwe przywrócenie stanu aplikacji przy crashu. W przypadku braku błędów, taki log nieznacznie będzie się rozszerzał dzięki systematycznemu oczyszczaniu przestarzałych danych. Taki log może również zwyczajnie pełnić funkcję cold data przy restarcie takiej aplikacji.

 

Jak to działa?

Do aktywnego segmentu trafiają nowe eventy. Po pewnym, zdefiniowanym czasie, tworzony jest nowy aktywny segment a ten poprzedni, uznawany jest jako nieoczyszczony. Proces oczyszczania zaczyna się w momencie gdy zostanie spełniony jeden z dwóch warunków:

• upłynął zadeklarowany maksymalny czas w jakim wiadomość może być w stanie nieoczyszczonym
• zdefiniowany stosunek rozmiaru nieoczyszczonych logów do całego logu topika został przekroczony i dla danej wiadomości został przekroczony również minimalny czas po jakim wiadomość może zostać poddana procesowi oczyszczania. 

 

Źródło: opracowanie własne 

 

Następnie z segmentów, które spełniły powyższe wymagania, wybierane są wszystkie najbardziej aktualne eventy. Decyzja którą wiadomość nie usuwać jest podejmowana za pomocą mapy kluczy i odpowiadającym im offsetom. Dla danego klucza wybierany jest offset z najwyższą wartością.

 

Źródło: opracowanie własne 

 

Kolejnym etapem jest porównanie tej mapy z segmentem już oczyszczonym. Jeśli znajduje się w nim jakiś klucz z niższym offsetem, jest on uznawany za nieaktualny i zostanie skasowany.

 

Źródło: opracowanie własne

 

Finalnie oczyszczone eventy są przekopiowane do nowego logu segmentów z zachowaniem ich offsetów a stare segmenty są kasowane.

 

Źródło: opracowanie własne 

 

Co w przypadku usunięcia rekordu?

Z perspektywy nowego konsumenta, który dopiero zaczyna czytać wiadomości nie ma sensu przetrzymywać starych kluczy z wartościami nullowymi. Dlatego w dłuższej perspektywie warto również zadbać o redukcje takich eventów. W tym przypadku wykorzystuje się parametr log.cleaner.delete.retention.ms, który oznacza czas zatrzymania znaczników o usunięciu rekordów. Znaczniki te zwane są nagrobkami. Są one dość istotne z perspektywy konsumentów, którzy mogą mieć chwilową niedostępność, ale nie zaczynają czytać logu od początku tylko od swojego offsetu. Dzięki temu, tacy konsumenci mają szansę na aktualizację wiadomości po swojej stronie. Ten parametr domyślnie przyjmuje wartość 24 godzin. Po tym czasie, eventy z nullowymi wartościami są usuwane.

 

Co gwarantuje log compaction?

• Każdy konsument, który zatrzyma się na head logu, czyli jest na bieżąco z eventami, zobaczy każdy nowy event.
• Każdy konsument, który dopiero został przypisany do grupy konsumentów i zaczyna przetwarzać eventy od początku, zobaczy przynajmniej końcowy stan dla wszystkich rekordów w kolejności w jakim były wpisane.
• Compaction process nigdy nie zmienia kolejności wiadomości, tylko część usunie. Offset dla wiadomości jako trwały identyfikator pozycji w logu nigdy się nie zmienia.

 

Podsumowanie

Log compaction to niewątpliwie bardzo ciekawa i użyteczna funkcjonalność Kafki, która pozwala na nieco inne zastosowanie takiego topika. Jest świetnym rozwiązaniem gdy nie chcemy utracić najbardziej aktualnych wiadomości danego klucza.