Apache Kafka z Avro i Schema Repo - gdzie w wiadomości pojawia się schemat Id?

Question

Chcę użyć Avro do serializowania danych dla moich wiadomości Kafka i chciałbym użyć go z repozytorium schematów Avro, więc nie muszę dołączać schematu do każdej wiadomości.

Używanie Avro z Kafką wydaje się być popularną rzeczą do zrobienia, a wiele blogów/pytań przepełnienia stosu/grup użytkowników, itp., Wysyłając identyfikator schematu z wiadomością, ale nie mogę znaleźć właściwego przykładu, gdzie powinien się udać.

Myślę, że powinien on być umieszczony w nagłówku wiadomości Kafki gdzieś, ale nie mogę znaleźć oczywistego miejsca. Jeśli było to w komunikacie Avro, musielibyśmy go rozszyfrować względem schematu, aby uzyskać treść wiadomości i ujawnić schemat, który należy rozszyfrować, co ma oczywiste problemy.

Używam klienta C#, ale przykład w dowolnym języku będzie świetny. Klasa wiadomości ma następujące pola:

public MessageMetadata Meta { get; set; } 
public byte MagicNumber { get; set; } 
public byte Attribute { get; set; } 
public byte[] Key { get; set; } 
public byte[] Value { get; set; } 

ale nie wydają się poprawne. MessageMetaData ma tylko Offset i PartitionId.

A gdzie powinien iść schemat Avro Schema?

Answer 1

Identyfikator schematu jest faktycznie zakodowany w samym komunikacie avro. Spójrz na this, aby zobaczyć, w jaki sposób zaimplementowane są kodery/dekodery.

W ogóle, co się dzieje, kiedy wysłać wiadomość do Avro Kafki:

1. Koder dostaje schematu z obiektu do zakodowania.
2. Koder prosi rejestr schematu o identyfikator dla tego schematu. Jeśli schemat jest już zarejestrowany, otrzymasz istniejący identyfikator, jeśli nie - rejestr zarejestruje schemat i zwróci nowy identyfikator.
3. Obiekt zostanie zakodowany w następujący sposób: [bajt magiczny] [identyfikator schematu] [aktualny komunikat] gdzie bajt magiczny to tylko bajt 0x0 używany do rozróżniania tego rodzaju wiadomości, identyfikator schematu to 4-bajtowa wartość całkowita, pozostałe jest faktyczną zakodowaną wiadomością.

Podczas dekodowania komunikatu powrotem oto co się dzieje:

1. Dekoder odczytuje pierwszy bajt i dba o to 0x0.
2. Dekoder odczytuje następne 4 bajty i konwertuje je na liczbę całkowitą. W ten sposób dekodowany jest identyfikator schematu.
3. Teraz, gdy dekoder ma identyfikator schematu, może poprosić rejestr schematu o rzeczywisty schemat dla tego identyfikatora. Voila!

Jeśli twój klucz jest kodowany Avro, to twój klucz będzie miał format opisany powyżej. To samo dotyczy wartości. W ten sposób klucz i wartość mogą być wartościami Avro i korzystać z różnych schematów.

Edit odpowiedzieć na pytanie w komentarzu:

Rzeczywista schematu są przechowywane w repozytorium schematu (czyli cały punkt repozytorium schematów faktycznie - do przechowywania schematów :)). Format plików kontenera obiektów Avro nie ma nic wspólnego z formatem opisanym powyżej.KafkaAvroEncoder/Decoder używają nieco innego formatu wiadomości (ale faktyczne wiadomości są zakodowane dokładnie w taki sam sposób).

Główna różnica między tymi formatami polega na tym, że pliki kontenerów obiektów mają rzeczywisty schemat i mogą zawierać wiele komunikatów odpowiadających temu schematowi, natomiast format opisany powyżej zawiera tylko identyfikator schematu i dokładnie jedną wiadomość odpowiadającą temu schematowi.

Przekazywanie zakodowanych wiadomości typu kontener-obiekt nie byłoby oczywiste, ponieważ jedna wiadomość Kafki zawierałaby wiele komunikatów Avro. Możesz też upewnić się, że jedna wiadomość Kafki zawiera tylko jedną wiadomość Avro, ale to spowoduje przeniesienie schematu z każdą wiadomością.

Schematy Avro mogą być dość duże (widziałem schematy takie jak 600 KB i więcej), a przenoszenie schematu z każdą wiadomością byłoby naprawdę kosztowne i nieekonomiczne, więc tam jest repozytorium schematów - schemat jest pobierany tylko raz i zostaje zapisany w pamięci podręcznej lokalnie, a wszystkie inne wyszukiwania są po prostu szybkimi mapami.