Zaawansowane tematy usługi AUTO CDC

Ta strona zawiera zaawansowane tematy dotyczące pracy z tabelami docelowymi AUTO CDC i AUTO CDC FROM SNAPSHOT, w tym operacjami DML, odczytywaniem strumieni danych zmian i monitorowaniem metryk przetwarzania. Aby zapoznać się z wprowadzeniem do interfejsu API AUTO CDC, zalecamy przeczytanie tekstu Interfejsy API AUTO CDC: upraszczanie przechwytywania zmian danych za pomocą potoków.

Dodawanie, zmienianie lub usuwanie danych w docelowej tabeli przesyłania strumieniowego

Jeśli potok publikuje tabele do Unity Catalog, możesz użyć instrukcji języka manipulowania danymi (DML), takich jak instrukcje wstawiania, aktualizowania, usuwania i scalania, aby modyfikować docelowe tabele przesyłania strumieniowego określone przez instrukcje .

Uwaga / Notatka

• Instrukcje DML modyfikujące schemat tabeli przesyłania strumieniowego nie są obsługiwane. Upewnij się, że instrukcje DML nie próbują rozwijać schematu tabeli.
• Instrukcje DML, które aktualizują tabelę strumieniową, mogą być uruchamiane tylko w klastrze Unity Catalog z udostępnionymi zasobami lub w usłudze SQL Warehouse przy użyciu silnika wykonawczego Databricks Runtime 13.3 LTS lub nowszego.
• Ponieważ przesyłanie strumieniowe wymaga źródeł danych wyłącznie do dołączania, jeśli twoje przetwarzanie wymaga przesyłania strumieniowego z tabeli źródłowej, która zawiera zmiany (na przykład przy użyciu instrukcji DML), ustaw flagę skipChangeCommits podczas odczytu tej tabeli źródłowej. Po ustawieniu skipChangeCommits transakcje, które usuwają lub modyfikują rekordy w tabeli źródłowej, są ignorowane. Jeśli twoje przetwarzanie nie wymaga tabeli streamingowej, możesz użyć widoku materializowanego (który nie ma ograniczenia tylko dodaj) jako tabeli docelowej.

** Ponieważ potoki deklaratywne Lakeflow Spark używają określonej SEQUENCE BY kolumny i propagują odpowiednie wartości sekwencyjne do kolumn __START_AT i __END_AT w tabeli docelowej (dla typu SCD Type 2), należy upewnić się, że instrukcje DML używają prawidłowych wartości dla tych kolumn, aby zachować właściwą kolejność rekordów. Zobacz Jak działa funkcja AUTO CDC.

Aby uzyskać więcej informacji na temat używania instrukcji DML z tabelami przesyłania strumieniowego, zobacz Dodawanie, zmienianie lub usuwanie danych w tabeli przesyłania strumieniowego.

Poniższy przykład wstawia aktywny rekord z sekwencją początkową 5:

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

Wskazówka

Jeśli musisz zmienić nazwę kolumn __START_AT i __END_AT w tabeli docelowej SCD Type 2 (na przykład w celu spełnienia wymagań schematu dalszego przetwarzania), utwórz widok na tabeli docelowej:

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

Odczytywanie zestawienia danych zmian z tabeli docelowej usługi AUTO CDC

W środowisku Databricks Runtime 15.2 lub nowszym można odczytać strumień danych zmian z tabeli przesyłania strumieniowego, która jest celem zapytania AUTO CDC lub AUTO CDC FROM SNAPSHOT w taki sam sposób, jak odczytywanie strumienia danych zmian z innych tabel Delta. Aby odczytać strumień danych zmian z docelowej tabeli przesyłania strumieniowego, wymagane są następujące elementy:

• Docelowa tabela przesyłania strumieniowego musi zostać opublikowana w Unity Catalog. Zobacz Używanie Unity Catalogu z potokami.
• Aby odczytać zestawienie zmian danych z docelowej tabeli przesyłania strumieniowego, musisz użyć środowiska Databricks Runtime 15.2 lub nowszego. Aby odczytać zestawienie danych zmian w innym potoku, potok musi być skonfigurowany do używania środowiska Databricks Runtime 15.2 lub nowszego.

Odczytasz kanał danych o zmianach z docelowej tabeli strumieniowej, która została utworzona w Lakeflow Spark Declarative Pipelines w taki sam sposób, jak odczytywanie kanału danych o zmianach z innych tabel Delta. Aby dowiedzieć się więcej na temat korzystania z funkcjonalności zmiany danych Delta, w tym przykładów w językach Python i SQL, zobacz Używanie funkcji kanału aktualizacji danych Delta Lake w Azure Databricks.

Uwaga / Notatka

Rekord zestawienia danych zmian zawiera metadane identyfikujące typ zdarzenia zmiany. Po zaktualizowaniu rekordu w tabeli metadane dotyczące powiązanych rekordów zmian zazwyczaj obejmują _change_type wartości ustawione na update_preimage oraz zdarzenia update_postimage.

Jednak wartości są różne, _change_type jeśli aktualizacje są wprowadzane do docelowej tabeli przesyłania strumieniowego, która obejmuje zmianę wartości klucza podstawowego. Gdy zmiany obejmują aktualizacje kluczy podstawowych, pola metadanych _change_type są ustawiane jako zdarzenia insert i delete. Zmiany kluczy podstawowych mogą wystąpić, gdy ręczne aktualizacje są wprowadzane do jednego z pól klucza w związku z instrukcją UPDATE lub MERGE albo, w przypadku tabel typu SCD 2, gdy pole __start_at zmienia się, aby odzwierciedlić wcześniejszą wartość sekwencji początkowej.

Zapytanie AUTO CDC określa wartości klucza podstawowego, które różnią się w przypadku przetwarzania typu SCD 1 i SCD typu 2:

Typ SCD	Klucz podstawowy
Typ SCD 1 i interfejs Python dla potoków	Klucz podstawowy jest wartością parametru keys w funkcji create_auto_cdc_flow(). W przypadku interfejsu SQL klucz podstawowy to kolumny zdefiniowane przez klauzulę KEYS w instrukcji AUTO CDC ... INTO .
Typ SCD 2	Kluczem podstawowym jest keys parametr lub KEYS klauzula plus wartość zwracana z coalesce(__START_AT, __END_AT) operacji, gdzie __START_AT i __END_AT są odpowiednimi kolumnami z docelowej tabeli strumieniowej.

Uzyskaj dane dotyczące rekordów przetwarzanych przez zapytanie CDC w potokach

Uwaga / Notatka

Poniższe metryki są przechwytywane przez zapytania AUTO CDC, lecz nie przez zapytania AUTO CDC FROM SNAPSHOT.

Następujące metryki są zbierane przez AUTO CDC zapytania:

• num_upserted_rows: liczba wierszy wyjściowych wstawionych lub zaktualizowanych w zestawie danych podczas aktualizacji.
• num_deleted_rows: liczba istniejących wierszy wyjściowych usuniętych z zestawu danych podczas aktualizacji.

Metryka wynikowa dla przepływów innych niż CDC nie jest przechwytywana dla zapytań num_output_rows.

Jakie obiekty danych są używane do przetwarzania CDC w potoku?

Po zadeklarowaniu tabeli docelowej w magazynie metadanych Hive tworzone są dwie struktury danych:

• Widok używający nazwy przypisanej do tabeli docelowej.
• Wewnętrzna tabela pomocnicza używana przez potok do zarządzania przetwarzaniem CDC. Ta tabela ma nazwę utworzoną przez poprzedzenie elementem __apply_changes_storage_ nazwy tabeli docelowej.

Jeśli na przykład zadeklarujesz tabelę docelową o nazwie dp_cdc_target, będziesz widzieć widok o nazwie dp_cdc_target oraz tabelę o nazwie __apply_changes_storage_dp_cdc_target w magazynie metadanych. Wykonaj zapytanie w widoku, aby uzyskać dostęp do przetworzonych danych. Nie należy bezpośrednio modyfikować tabeli bazowej.

Uwaga / Notatka

Te struktury danych dotyczą tylko AUTO CDC przetwarzania, a nie AUTO CDC FROM SNAPSHOT przetwarzania. Dotyczą one wyłącznie magazynu metadanych Hive, a nie Unity Catalog.