Transformowanie danych przed zapisem do celu.
GIFRÖST umożliwia transformowanie danych w kilku miejscach potoku CDC: podczas odczytu snapshotu inicjalnego, w konektorze źródłowym przed zapisem rekordu do Kafki oraz w konektorze docelowym przed zapisem do systemu docelowego. Kafka pozostaje rdzeniem całego przepływu, a transformacje na konektorach pozwalają dopasować dane do reguł biznesowych, bezpieczeństwa i schematu docelowego.
Punkty transformacji
Transformacje można wykonywać w różnych momentach, zależnie od celu. Transformacje na etapie snapshotu są przydatne, gdy ładowanie inicjalne ma używać konkretnego zapytania do bazy źródłowej. Single Message Transformations (SMT) są przydatne wtedy, gdy każdy rekord Kafka Connect ma zostać przefiltrowany, wzbogacony, przemianowany, zamaskowany, przekierowany albo dopasowany przed przejściem do kolejnego etapu.
Kafka jest rdzeniem ścieżki dostarczania danych w GIFRÖST. Konektory źródłowe mogą transformować rekordy zanim Kafka je zapisze, a konektory docelowe mogą wykonać kolejne transformacje po odczycie z Kafki i przed zapisem do systemu docelowego.
Debezium może używać customowych zapytań snapshotowych dla wybranych tabel. Pozwala to
ograniczyć albo ukształtować dane już po stronie bazy źródłowej, na przykład przez warunki
WHERE, sortowanie albo obsługiwane przez bazę wyrażenia w zapytaniu użytym do
zasilenia.
Konektor źródłowy może wykonać SMT po utworzeniu rekordu i przed zapisaniem go w Kafce. To miejsce na filtrowanie eventów, maskowanie danych wrażliwych, dodawanie metadanych, rename pól, routing topiców albo spłaszczanie struktury zdarzeń Debezium.
Konektor docelowy może wykonać kolejny łańcuch SMT po pobraniu rekordu z Kafki i przed zapisem do systemu docelowego. Umożliwia to formatowanie pod konkretną bazę, budowanie klucza, konwersję typów, dopasowanie schematu albo końcową anonimizację.
Konfiguracja w interfejsie
W kreatorze konektora transformacje są konfigurowane jako łańcuch. Każda transformacja ma unikalną nazwę, typ oraz zestaw wymaganych parametrów. Użytkownik może wybierać zapisane transformacje, edytować ich konfigurację i przypisywać je do definicji konektorów źródłowych lub docelowych.
Kreator konektora zawiera osobny krok Transforms, w którym użytkownik buduje łańcuch SMT dodawany do konfiguracji konektora.
Zapisane transformacje mogą być ponownie używane w wielu konfiguracjach konektorów. Przykład pokazuje transformację filtrującą z językiem wyrażeń i warunkiem filtrowania.
Przykładowe transformacje
Platforma może korzystać z wbudowanych transformacji Kafka Connect, transformacji specyficznych dla Debezium oraz customowych transformacji dostarczonych jako pluginy Kafka Connect. Poniższa lista jest przykładowa; dokładny zestaw dostępny we wdrożeniu zależy od zainstalowanych konektorów i pluginów.
| Transformacja | Typowe zastosowanie | Przykład użycia |
|---|---|---|
Filter | Przepuszczanie albo odrzucanie rekordów na podstawie wyrażenia lub predykatu. | Przepuszczenie tylko rekordów, dla których status = ACTIVE, amount > 1000 albo operacja CDC to UPDATE. |
ContentBasedRouter | Routing rekordów do różnych topiców na podstawie zawartości rekordu. | Wysyłanie klientów z country = PL do jednego topicu, a klientów z country = DE do innego. |
ValueToKey + ExtractField | Budowanie albo uproszczenie klucza rekordu Kafka na podstawie pól z wartości rekordu. | Utworzenie klucza docelowego z customer_id, order_id albo innego identyfikatora biznesowego. |
MaskField | Maskowanie wartości pól wrażliwych. | Anonimizacja danych osobowych zanim rekord opuści konektor. |
ReplaceField | Zmiana nazw pól, wybór pól albo wykluczanie pól. | Dopasowanie nazw kolumn źródłowych do konwencji systemu docelowego. |
InsertField / InsertHeader | Dodawanie pól metadanych albo nagłówków. | Dodanie czasu przetwarzania, nazwy źródła, środowiska albo statycznych metadanych. |
HeaderFrom / HeaderToValue | Przenoszenie wartości między payloadem i nagłówkami rekordu. | Udostępnienie metadanych źródła, identyfikatora tenanta albo trace ID systemom downstream. |
RegexRouter / TimestampRouter | Routing rekordów przez zmianę nazw topiców. | Normalizacja nazw topiców albo podział topiców docelowych według czasu. |
TimestampConverter / Cast | Konwersja typów pól i formatów czasu. | Dostosowanie wartości źródłowych do formatu oczekiwanego przez konektor docelowy. |
ExtractNewRecordState | Spłaszczenie zdarzeń Debezium do prostszej struktury rekordu. | Udostępnienie tylko nowego stanu wiersza konsumentom, którzy nie potrzebują pełnej koperty CDC. |
Flatten / HoistField | Przekształcanie zagnieżdżonych rekordów do struktury płaskiej albo opakowanej. | Przygotowanie payloadu dla sinków, które oczekują prostego obiektu albo konkretnego pola opakowującego. |
TombstoneHandler | Kontrola sposobu obsługi rekordów tombstone. | Ignorowanie, logowanie, zatrzymanie albo routing rekordów tombstone zależnie od zachowania systemu docelowego. |
TimezoneConverter | Obsługa metadanych Debezium i pól czasu. | Normalizacja stref czasowych w polach eventu przed dostarczeniem danych do bazy docelowej albo file sinka. |
Wartość biznesowa: przepuszczanie tylko wierszy, w których
status ma wartość ACTIVE, priority ma wartość
HIGH albo amount przekracza uzgodniony próg.
Operacja CDC: przepuszczanie tylko eventów INSERT,
UPDATE albo DELETE, zależnie od tego, co powinien otrzymać system
docelowy.
Tenant albo region: routing lub filtrowanie rekordów według wartości takich
jak tenant_id, country, region albo
source_system.
Jakość danych: odrzucanie rekordów, w których wymagane pola są puste, brakujące albo poza akceptowanym zakresem, zanim trafią do konektora docelowego.
Konwersja typów przez custom converters
Część konwersji typów powinna zostać wykonana jeszcze zanim rekord zostanie wyemitowany przez Debezium, czyli na poziomie mapowania typów w konektorze. GIFRÖST wspiera ten model przez możliwość dodania konfiguracji Debezium custom converters do konfiguracji konektora. Jest to szczególnie przydatne, gdy baza źródłowa używa typów legacy, typów specyficznych dla vendora albo typów niejednoznacznych, które w Kafce i w systemie docelowym powinny być widoczne jako inny typ logiczny.
Konwersja na poziomie konektora: custom converters są konfigurowane na konektorze źródłowym. Debezium stosuje je podczas budowania schematu i wartości emitowanego eventu, zanim rekord trafi do dalszych SMT albo konektorów docelowych.
Selektywne stosowanie: konwertery zwykle można ograniczyć przez wyrażenie
regularne selector, dzięki czemu konwersja obejmuje wybrane tabele lub kolumny,
a nie cały konektor.
Wsparcie zależne od bazy: poniższe przykłady dla Oracle są reprezentatywne. Ten sam model konfiguracji dotyczy także innych konektorów Debezium, jeśli udostępniają one własne custom converters albo równoważne opcje mapowania typów na poziomie konektora.
| Przykład Oracle | Rozwiązywany problem | Efekt w strumieniu CDC |
|---|---|---|
NumberOneToBooleanConverter | Starsze schematy Oracle często modelują wartości boolean jako NUMBER(1) z wartościami 0 i 1. | Kolumny mogą być emitowane jako logiczny BOOL zamiast liczbowego INT8. |
NumberToZeroScaleConverter | Oracle może używać kolumn NUMBER z ujemną skalą, co bywa problematyczne dla formatów takich jak Avro albo dla konsumentów downstream. | Wartości numeryczne są emitowane ze skalą zero zgodnie z wybranym trybem obsługi decimal. |
RawToStringConverter | Systemy legacy mogą przechowywać dane znakowe w kolumnach RAW, które Debezium domyślnie emituje jako bajty. | Wybrane kolumny RAW mogą zostać zdekodowane i wyemitowane jako logiczne wartości STRING. |
Przykładowa konfiguracja konektora dla konwersji Oracle NUMBER(1) do boolean:
converters=number-to-boolean
number-to-boolean.type=io.debezium.connector.oracle.converters.NumberOneToBooleanConverter
number-to-boolean.selector=.*.MY_TABLE.DATA
Przykładowa konfiguracja konektora dla konwersji Oracle RAW do string:
converters=raw-to-string
raw-to-string.type=io.debezium.connector.oracle.converters.RawToStringConverter
raw-to-string.selector=.*.MY_TABLE.DATA
raw-to-string.charset=UTF-8
Customowe transformacje
Customowy kod Java: Kafka Connect obsługuje własne pluginy SMT
implementujące org.apache.kafka.connect.transforms.Transformation. Po wgraniu do
ścieżki pluginów workera mogą być używane przez konektory tak jak transformacje wbudowane.
Konfiguracja per konektor: transformacje są konfigurowane we właściwościach konektora, więc użytkownik może przypisać inne łańcuchy transformacji do konektorów źródłowych i inne do konektorów docelowych.
Granica operacyjna: SMT to lekkie operacje wykonywane rekord po rekordzie. Bardzo dobrze nadają się do filtrowania, maskowania, routingu, zmian pól i prostego wzbogacania. Przetwarzanie stanowe albo joiny powinny być realizowane przez komponent stream-processingowy.
Kafka Connect opisuje SMT w Single Message Transformations Reference oraz punkt rozszerzeń Java w Apache Kafka Transformation interface. Debezium opisuje swoje transformacje w Debezium Transformations oraz custom converters konektora Oracle w Debezium Oracle custom converters. Customizacja zapytań snapshotowych jest opisana w Debezium connector snapshot properties.