Pozyskiwanie danych za pomocą kanału informacyjnego CDC

Gdy systemy źródłowe generują codziennie miliony transakcji, ponowne ładowanie całych tabel staje się niepraktyczne. Potrzebny jest sposób przechwytywania tylko zmian — wstawiania, aktualizacji i usuwania — i efektywnego stosowania ich do tabel docelowych. Ten wzorzec, znany jako przechwytywanie zmian danych (CDC), stanowi podstawę nowoczesnych potoków danych przyrostowych.

Zrozumienie wzorców przetwarzania danych CDC

Pozyskiwanie danych zmian obejmuje odczytywanie strumienia rekordów zmian z systemu źródłowego i stosowanie tych zmian do tabeli docelowej. W przeciwieństwie do ponownego ładowania pełnej tabeli usługa CDC przetwarza tylko to, co zmieniło się od ostatniej synchronizacji, co zmniejsza zarówno czas przetwarzania, jak i zużycie zasobów.

Typowy kanał informacyjny CDC zawiera rekordy o następującej strukturze:

• Kolumny danych: rzeczywiste wartości dla każdego pola w tabeli źródłowej
• Typ operacji: wskazuje, czy rekord to INSERT, UPDATE, czy DELETE
• Kolumna sekwencji: znacznik czasu lub numer sekwencji określający kolejność zmian

Rozważ tabelę pracowników, która śledzi nazwę i lokalizację. Podczas aktualizowania miasta pracownika źródło danych CDC nie wysyła całej tabeli — wysyła jeden rekord z nową wartością miasta, typem operacji UPDATE i numerem sekwencji. Potok przetwarzania odczytuje ten rekord i wprowadza zmianę w odpowiednim wierszu tabeli docelowej.

Takie podejście oferuje kilka korzyści dla inżynierów danych:

• Mniejsze opóźnienie: Zmienia przepływ do miejsca docelowego w ciągu kilku minut, a nie godzin
• Niższe koszty: Przetwarzanie mniejszej liczby rekordów oznacza mniej czasu obliczeniowego i operacji wejścia/wyjścia pamięci masowej
• Minimalny wpływ na źródło: odczytywanie dzienników zmian powoduje mniejsze obciążenie produkcyjnych baz danych niż pełne skanowanie tabel

Przetwarzanie CDC z użyciem AUTO CDC API

Azure Databricks udostępnia interfejs API AUTO CDC w potokach Lakeflow, aby uprościć przetwarzanie danych zmian. Ten interfejs API automatycznie obsługuje złożoność rekordów w niewłaściwej kolejności, deduplikację i zastosowanie zmian.

Aby zaimplementować przetwarzanie CDC, należy najpierw utworzyć tabelę przesyłania strumieniowego jako docelową. Następnie zdefiniuj przepływ, który odczytuje rekordy zmian ze źródła i stosuje je, używając odpowiedniego typu wymiaru wolnozmiennego (SCD).

Typ SCD 1 przechowuje tylko bieżącą wersję każdego rekordu:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.temporary_view
def employees_cdf():
    return spark.readStream.table("bronze.employee_changes")

dp.create_streaming_table("employees")

dp.create_auto_cdc_flow(
    target="employees",
    source="employees_cdf",
    keys=["employee_id"],
    sequence_by=col("sequence_num"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "sequence_num"],
    stored_as_scd_type=1
)

W przypadku typu SCD 1, gdy pracownik przenosi się z Seattle do Portland, tabela docelowa aktualizuje kolumnę miasta bezpośrednio. Rekord Seattle znika — pozostaje tylko Portland.

Typ SCD 2 zachowuje pełną historię zmian:

dp.create_streaming_table("employees_history")

dp.create_auto_cdc_flow(
    target="employees_history",
    source="employees_cdf",
    keys=["employee_id"],
    sequence_by=col("sequence_num"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "sequence_num"],
    stored_as_scd_type=2
)

Typ SCD 2 dodaje kolumny __START_AT i __END_AT do śledzenia, kiedy każda wersja była aktywna. Rekord Seattle pozostaje z datą zakończenia, podczas gdy nowy rekord Portland jest wyświetlany bez daty zakończenia. Analitycy mogą wysyłać zapytania do tabeli, aby zobaczyć dokładnie, kiedy i jak zmieniają się dane pracowników w czasie.

Niektóre systemy źródłowe emitują częściowe rekordy aktualizacji, które zawierają tylko zmienione kolumny, a nie pełne kopie wierszy. W takich przypadkach zastosowanie aktualizacji bezpośrednio spowoduje zastąpienie niezmienionych kolumn wartościami null. Parametr ignore_null_updates (Python) lub klauzula IGNORE NULL UPDATES (SQL) zapobiega temu: gdy zdarzenie CDC pasuje do istniejącego wiersza, wszystkie wartości null w rekordzie przychodzącym pozostawiają istniejące wartości kolumn bez zmiany, zamiast zastępowania ich.

Konfigurowanie przepływów usługi CDC w usłudze SQL

Te same wzorce CDC działają w języku SQL przy użyciu składni deklaratywnej. Przepływy SQL integrują się naturalnie z potokami architektury medalonu, w których definiuje się warstwy brązowe, srebrne i złote.

CREATE OR REFRESH STREAMING TABLE employees;

CREATE FLOW employees_cdc_flow
AS AUTO CDC INTO employees
FROM stream(bronze.employee_changes)
KEYS (employee_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 1;

Składnia SQL odzwierciedla interfejs API języka Python. Należy określić tabelę docelową, strumień źródłowy, kolumny kluczy, warunek usuwania, kolumnę sekwencjonowania i typ SCD. Kolumny, których nie potrzebujesz w miejscu docelowym — na przykład kolumny operacji i sekwencji — są wykluczone z klauzuli EXCEPT .

W przypadku śledzenia historycznego z typem SCD 2 można również ograniczyć, które kolumny wyzwalają nowe rekordy historii:

CREATE FLOW employees_history_flow
AS AUTO CDC INTO employees_history
FROM stream(bronze.employee_changes)
KEYS (employee_id)
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 2
TRACK HISTORY ON * EXCEPT (last_login);

W przypadku tej konfiguracji zmiany w kolumnie last_login aktualizują istniejący rekord na miejscu. Tylko zmiany w innych kolumnach— takich jak nazwa lub dział — tworzą nowe wersje historyczne.

Obsługa zdarzeń niezgodnych z kolejnością

Systemy rozproszone często dostarczają zdarzenia poza kolejnością. Opóźnienie sieci może spowodować nadejście starszej aktualizacji po nowszym. Interfejs API AUTO CDC obsługuje ten scenariusz automatycznie przy użyciu określonej kolumny sekwencji.

Po nadejściu dwóch aktualizacji dla tego samego klucza interfejs API porównuje ich wartości sekwencji. Jeśli nowsza aktualizacja (wyższy numer sekwencji) została już zastosowana, a starsza aktualizacja pojawi się późno, interfejs API ignoruje rekord opóźnionego przybycia. Dzięki temu tabela docelowa zawsze odzwierciedla prawidłowy stan końcowy.

Bez tej wbudowanej obsługi należy napisać złożoną logikę scalania, która sprawdza znaczniki czasu, porównuje wersje i warunkowo stosuje aktualizacje. Interfejs API AUTO CDC hermetyzuje całą tę logikę, umożliwiając skoncentrowanie się na wymaganiach biznesowych, a nie na obsłudze przypadków brzegowych.