Używanie zestawienia danych zmian usługi Delta Lake w usłudze Azure Databricks

Zestawienie zmian danych umożliwia usłudze Azure Databricks śledzenie zmian na poziomie wiersza między wersjami tabeli delty. Po włączeniu w tabeli delty środowisko uruchomieniowe rejestruje zdarzenia zmiany dla wszystkich danych zapisanych w tabeli. Obejmuje to dane wierszy wraz z metadanymi wskazującymi, czy określony wiersz został wstawiony, usunięty lub zaktualizowany.

Przyrostowe przetwarzanie zmian danych

Usługa Databricks zaleca używanie zestawienia zmian danych w połączeniu ze strukturą przesyłania strumieniowego w celu przyrostowego przetwarzania zmian z tabel różnicowych. Musisz użyć przesyłania strumieniowego ze strukturą dla usługi Azure Databricks, aby automatycznie śledzić wersje zestawienia zmian w tabeli.

Aby odczytać zestawienie danych zmian z tabeli, musisz włączyć zestawienie danych zmian w tej tabeli. Zobacz Włączanie zestawienia danych zmian.

Ustaw opcję readChangeFeed na true wartość podczas konfigurowania strumienia względem tabeli, aby odczytać zestawienie danych zmian, jak pokazano w poniższym przykładzie składni:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("myDeltaTable")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("myDeltaTable")

Domyślnie strumień zwraca najnowszą migawkę tabeli po pierwszym uruchomieniu strumienia w postaci INSERT i przyszłych zmian w miarę zmieniania danych.

Zmiana zatwierdzeń danych w ramach transakcji usługi Delta Lake i staje się dostępna w tym samym czasie nowych zatwierdzeń danych w tabeli.

Opcjonalnie możesz określić wersję początkową. Zobacz Czy należy określić wersję początkową?.

Źródło danych zmian obsługuje również wykonywanie wsadowe, które wymaga określenia wersji początkowej. Zobacz Odczyt zmian w zapytaniach wsadowych.

Opcje, takie jak limity szybkości (maxFilesPerTrigger, maxBytesPerTrigger) i excludeRegex są również obsługiwane podczas odczytywania danych zmiany.

Ograniczanie szybkości może być niepodzielne dla wersji innych niż początkowa wersja migawki. Oznacza to, że cała wersja zatwierdzenia będzie ograniczona lub zostanie zwrócone całe zatwierdzenie.

Czy należy określić wersję początkową?

Opcjonalnie możesz określić wersję początkową, jeśli chcesz zignorować zmiany, które wystąpiły przed określoną wersją. Wersję można określić przy użyciu znacznika czasu lub numeru identyfikatora wersji zarejestrowanego w dzienniku transakcji delty.

Podczas konfigurowania obciążeń przesyłania strumieniowego ze strukturą dotyczących zestawienia zmian ważne jest, aby zrozumieć, jak określenie wersji początkowej ma wpływ na przetwarzanie.

Wiele obciążeń przesyłania strumieniowego, zwłaszcza nowych potoków przetwarzania danych, korzysta z domyślnego zachowania. W przypadku zachowania domyślnego pierwsza partia jest przetwarzana, gdy strumień najpierw rejestruje wszystkie istniejące rekordy w tabeli jako INSERT operacje w zestawieniach danych zmian.

Jeśli tabela docelowa zawiera już wszystkie rekordy z odpowiednimi zmianami do określonego punktu, określ wersję początkową, aby uniknąć przetwarzania stanu tabeli źródłowej jako INSERT zdarzeń.

Następująca przykładowa składnia odzyskiwania po niepowodzeniu przesyłania strumieniowego, w którym punkt kontrolny został uszkodzony. W tym przykładzie przyjęto założenie, że spełnione są następujące warunki:

1. Źródło danych zmian zostało włączone w tabeli źródłowej podczas tworzenia tabeli.
2. Docelowa tabela podrzędna przetworzyła wszystkie zmiany do wersji 75.
3. Historia wersji tabeli źródłowej jest dostępna dla wersji 70 lub nowszych.

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")

W tym przykładzie należy również określić nową lokalizację punktu kontrolnego.

Zobacz Czy mogę użyć zestawienia zmian danych, aby odtworzyć całą historię tabeli?.

Odczytywanie zmian w zapytaniach wsadowych

Składnia zapytań wsadowych umożliwia odczytywanie wszystkich zmian rozpoczynających się od określonej wersji lub odczytywanie zmian w określonym zakresie wersji.

Należy określić wersję jako liczbę całkowitą i znaczniki czasu jako ciąg w formacie yyyy-MM-dd[ HH:mm:ss[.SSS]].

Wersje początkowe i końcowe są uwzględniane w zapytaniach. Aby odczytać zmiany z określonej wersji początkowej do najnowszej wersji tabeli, określ tylko wersję początkową.

Jeśli podasz wersję niższą lub sygnaturę czasową starszą niż ta, która zarejestrowała zdarzenia zmiany — oznacza to, że po włączeniu zestawienia danych zmian zostanie zgłoszony błąd wskazujący, że źródło danych zmian nie zostało włączone.

W poniższych przykładach składni pokazano, jak używać opcji uruchamiania i kończenia wersji z odczytami wsadowymi:

SQL

-- version as ints or longs e.g. changes from version 0 to 10
SELECT * FROM table_changes('tableName', 0, 10)

-- timestamp as string formatted timestamps
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')

-- providing only the startingVersion/timestamp
SELECT * FROM table_changes('tableName', 0)

-- database/schema names inside the string for table name, with backticks for escaping dots and special characters
SELECT * FROM table_changes('dbName.`dotted.tableName`', '2021-04-21 06:45:46' , '2021-05-21 12:00:00')

Python

# version as ints or longs
spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .option("endingVersion", 10) \
  .table("myDeltaTable")

# timestamps as formatted timestamp
spark.read \
  .option("readChangeFeed", "true") \
  .option("startingTimestamp", '2021-04-21 05:45:46') \
  .option("endingTimestamp", '2021-05-21 12:00:00') \
  .table("myDeltaTable")

# providing only the startingVersion/timestamp
spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")

Scala

// version as ints or longs
spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .option("endingVersion", 10)
  .table("myDeltaTable")

// timestamps as formatted timestamp
spark.read
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2021-04-21 05:45:46")
  .option("endingTimestamp", "2021-05-21 12:00:00")
  .table("myDeltaTable")

// providing only the startingVersion/timestamp
spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")

set spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled = true;

Jaki jest schemat zestawienia zmian danych?

Podczas odczytywania ze źródła danych zmian dla tabeli jest używany schemat najnowszej wersji tabeli.

Oprócz kolumn danych ze schematu tabeli delta zestawienie danych zawiera kolumny metadanych identyfikujące typ zdarzenia zmiany:

(1) preimage jest wartością przed aktualizacją, postimage jest wartością po aktualizacji.

Włączanie zestawienia danych zmian

Źródło danych zmian można odczytywać tylko dla tabel z obsługą. Należy jawnie włączyć opcję zestawienia danych zmian przy użyciu jednej z następujących metod:

• Nowa tabela: ustaw właściwość delta.enableChangeDataFeed = true tabeli w poleceniu CREATE TABLE .

  CREATE TABLE student (id INT, name STRING, age INT) TBLPROPERTIES (delta.enableChangeDataFeed = true)
  

• Istniejąca tabela: ustaw właściwość delta.enableChangeDataFeed = true tabeli w poleceniu ALTER TABLE .

  ALTER TABLE myDeltaTable SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
  

• Wszystkie nowe tabele:

  set spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;
  

Zmienianie magazynu danych

Włączenie zestawienia danych zmian powoduje niewielki wzrost kosztów magazynowania dla tabeli. Rekordy danych zmiany są generowane podczas uruchamiania zapytania i są zazwyczaj znacznie mniejsze niż całkowity rozmiar przepisanych plików.

Usługa Azure Databricks rejestruje dane zmiany danych dla UPDATEoperacji , DELETEi MERGE w _change_data folderze w katalogu tabeli. Niektóre operacje, takie jak operacje wstawiania i usuwanie pełnej partycji, nie generują danych w _change_data katalogu, ponieważ usługa Azure Databricks może efektywnie obliczyć zestawienie danych zmian bezpośrednio z dziennika transakcji.

Wszystkie operacje odczytu względem plików danych w folderze _change_data powinny przechodzić przez obsługiwane interfejsy API usługi Delta Lake.

Pliki w folderze _change_data są zgodne z zasadami przechowywania tabeli. Zmiana danych źródła danych jest usuwana po uruchomieniu VACUUM polecenia.

Czy mogę użyć zestawienia zmian danych, aby odtworzyć całą historię tabeli?

Zestawienie danych zmian nie ma służyć jako trwały rekord wszystkich zmian w tabeli. Zmiana źródła danych rejestruje tylko zmiany, które występują po jej włączeniu.

Zestawienie danych zmian i usługa Delta Lake umożliwiają zawsze odtworzenie pełnej migawki tabeli źródłowej, co oznacza, że możesz rozpocząć nowe przesyłanie strumieniowe odczytane względem tabeli z włączonym zestawieniem zmian i przechwycić bieżącą wersję tej tabeli oraz wszystkie zmiany, które występują po.

Rekordy w kanale danych zmian należy traktować jako przejściowe i dostępne tylko dla określonego okna przechowywania. Dziennik transakcji delty usuwa wersje tabel i odpowiadające im wersje zestawienia danych zmian w regularnych odstępach czasu. Po usunięciu wersji z dziennika transakcji nie można już odczytać zestawienia danych zmian dla tej wersji.

Jeśli przypadek użycia wymaga zachowania trwałej historii wszystkich zmian w tabeli, należy użyć logiki przyrostowej do zapisywania rekordów ze źródła danych zmian do nowej tabeli. W poniższym przykładzie kodu pokazano użycie metody trigger.AvailableNow, która wykorzystuje przyrostowe przetwarzanie przesyłania strumieniowego ze strukturą, ale przetwarza dostępne dane jako obciążenie wsadowe. To obciążenie można zaplanować asynchronicznie przy użyciu głównych potoków przetwarzania, aby utworzyć kopię zapasową zestawienia zmian danych na potrzeby inspekcji lub pełną powtarzalność.

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("target_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.AvailableNow)
  .toTable("target_table")

Zmienianie ograniczeń źródła danych dla tabel z włączonym mapowaniem kolumn

Po włączeniu mapowania kolumn w tabeli delty można usuwać lub zmieniać nazwy kolumn w tabeli bez ponownego zapisywania plików danych dla istniejących danych. Po włączeniu mapowania kolumn zmiana źródła danych ma ograniczenia po wprowadzeniu zmian schematu nie addytywnego, takich jak zmiana nazwy lub usunięcie kolumny, zmiana typu danych lub zmiany wartości null.

W środowisku Databricks Runtime 12.2 LTS lub nowszym można wykonywać operacje odczytu wsadowego dla zestawienia danych zmian dla tabel z włączonym mapowaniem kolumn, które doświadczyły zmian schematu nie addytywnego. Zamiast używać schematu najnowszej wersji tabeli, operacje odczytu używają schematu końcowej wersji tabeli określonej w zapytaniu. Zapytania nadal kończą się niepowodzeniem, jeśli określony zakres wersji obejmuje zmianę schematu nie addytywnego.