Práce s historií tabulek Delta Lake

Každá operace, která upravuje tabulku Delta Lake, vytvoří novou verzi tabulky. Informace o historii můžete použít k auditování operací, vrácení tabulky zpět nebo dotazování na tabulku v určitém časovém okamžiku pomocí časového cestování.

Načtení historie tabulky Delta

Spuštěním příkazu můžete načíst informace, včetně operací, uživatelů a časového razítka každého zápisu history do tabulky Delta. Operace se vrátí v obráceném chronologickém pořadí.

Uchovávání historie tabulek je určeno nastavením delta.logRetentionDurationtabulky, což je ve výchozím nastavení 30 dnů.

DESCRIBE HISTORY table_name       -- get the full history of the table

DESCRIBE HISTORY table_name LIMIT 1  -- get the last operation only

Podrobnosti o syntaxi Spark SQL najdete v tématu POPIS HISTORIE.

Podrobnosti o syntaxi jazyka Scala/Java/Python najdete v dokumentaci k rozhraní Delta Lake API.

Průzkumník katalogu poskytuje vizuální zobrazení těchto podrobných informací o tabulce a historii tabulek Delta. Kromě schématu tabulky a ukázkových dat můžete kliknutím na kartu Historie zobrazit historii tabulek, která se zobrazí pomocí DESCRIBE HISTORY.

Schéma historie

Výstup operace history má následující sloupce.

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|   isolationLevel|isBlindAppend|    operationMetrics|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|      5|2019-07-29 14:07:47|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          4|WriteSerializable|        false|[numTotalRows -> ...|
|      4|2019-07-29 14:07:41|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          3|WriteSerializable|        false|[numTotalRows -> ...|
|      3|2019-07-29 14:07:29|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          2|WriteSerializable|        false|[numTotalRows -> ...|
|      2|2019-07-29 14:06:56|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          1|WriteSerializable|        false|[numTotalRows -> ...|
|      1|2019-07-29 14:04:31|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          0|WriteSerializable|        false|[numTotalRows -> ...|
|      0|2019-07-29 14:01:40|   ###|     ###|    WRITE|[mode -> ErrorIfE...|null|     ###|      ###|       null|WriteSerializable|         true|[numFiles -> 2, n...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

Klíče metrik operací

Operace history vrátí kolekci metrik operací v mapě operationMetrics sloupců.

Následující tabulky uvádějí definice klíčů mapování podle operace.

Co je časová cesta Delta Lake?

Časová cesta Delta Lake podporuje dotazování předchozích verzí tabulek na základě časového razítka nebo verze tabulky (jak je zaznamenáno v transakčním protokolu). Pro aplikace, jako jsou například tyto, můžete použít časová cesta:

• Opětovné vytváření analýz, sestav nebo výstupů (například výstup modelu strojového učení). To může být užitečné pro ladění nebo auditování, zejména v regulovaných odvětvích.
• Psaní složitých dočasných dotazů
• Oprava chyb v datech
• Poskytuje izolaci snímků pro sadu dotazů pro rychlé změny tabulek.

Syntaxe pohybu v čase Delta

Dotazujete tabulku Delta s časovým cestováním přidáním klauzule za specifikaci názvu tabulky.

• timestamp_expression může být libovolná z těchto možností:
  • '2018-10-18T22:15:12.013Z'to znamená řetězec, který lze přetypovat na časové razítko.
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18'to znamená řetězec kalendářního data.
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Jakýkoli jiný výraz, který je nebo lze přetypovat na časové razítko
• version je dlouhá hodnota, kterou lze získat z výstupu DESCRIBE HISTORY table_spec.

version Ani timestamp_expression nemůže být poddotaz.

Akceptují se pouze řetězce data nebo časového razítka. Příklad: "2019-01-01" a "2019-01-01T00:00:00.000Z". Příklad syntaxe najdete v následujícím kódu:

SQL

SELECT * FROM people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z';
SELECT * FROM people10m VERSION AS OF 123;

Python

df1 = spark.read.option("timestampAsOf", "2019-01-01").table("people10m")
df2 = spark.read.option("versionAsOf", 123).table("people10m")

Syntaxi můžete také použít @ k určení časového razítka nebo verze jako součásti názvu tabulky. Časové razítko musí být ve yyyyMMddHHmmssSSS formátu. Verzi @ můžete zadat tak, že ji předejdete v . Příklad syntaxe najdete v následujícím kódu:

SQL

SELECT * FROM people10m@20190101000000000
SELECT * FROM people10m@v123

Python

spark.read.table("people10m@20190101000000000")
spark.read.table("people10m@v123")

Co jsou kontrolní body transakčního protokolu?

Delta Lake zaznamenává verze tabulek jako soubory JSON v adresáři _delta_log , které se ukládají společně s daty tabulek. Pokud chcete optimalizovat dotazování kontrolních bodů, Delta Lake agreguje verze tabulek do souborů kontrolních bodů Parquet, což brání nutnosti číst všechny verze tabulek JSON historie. Azure Databricks optimalizuje frekvenci vytváření kontrolních bodů pro velikost dat a úlohu. Uživatelé by neměli pracovat s kontrolními body přímo. Četnost kontrolních bodů se může bez předchozího upozornění změnit.

Konfigurace uchovávání dat pro dotazy na časové cesty

Pokud chcete dotazovat předchozí verzi tabulky, musíte zachovat protokol i datové soubory pro danou verzi.

Datové soubory se odstraní při VACUUM spuštění v tabulce. Delta Lake spravuje odebrání souboru protokolu automaticky po verzích tabulek kontrolních bodů.

Vzhledem k tomu, že většina tabulek Delta se VACUUM s nimi spouští pravidelně, měly by dotazy k určitému bodu v čase respektovat prahovou hodnotu VACUUMuchovávání informací, což je ve výchozím nastavení 7 dnů.

Pokud chcete zvýšit prahovou hodnotu uchovávání dat pro tabulky Delta, musíte nakonfigurovat následující vlastnosti tabulky:

• delta.logRetentionDuration = "interval <interval>": Určuje, jak dlouho se historie tabulky uchovává. Výchozí hodnota je interval 30 days.
• delta.deletedFileRetentionDuration = "interval <interval>": určuje prahovou hodnotu VACUUM , která se používá k odebrání datových souborů, na které se už v aktuální verzi tabulky neodkazuje. Výchozí hodnota je interval 7 days.

Během vytváření tabulky můžete zadat vlastnosti Delta nebo je nastavit pomocí ALTER TABLE příkazu. Viz odkaz na vlastnosti tabulky Delta.

Obnovení tabulky Delta do dřívějšího stavu

Tabulku Delta můžete obnovit do dřívějšího RESTORE stavu pomocí příkazu. Tabulka Delta interně udržuje historické verze tabulky, které umožňují obnovení do dřívějšího stavu. Verze odpovídající dřívějšímu stavu nebo časovému razítku toho, když došlo k vytvoření dřívějšího stavu, se příkazem RESTORE podporují jako možnosti.

RESTORE TABLE target_table TO VERSION AS OF <version>;
RESTORE TABLE target_table TO TIMESTAMP AS OF <timestamp>;

Podrobnosti o syntaxi najdete v tématu OBNOVENÍ.

Obnovení metrik

RESTORE Po dokončení operace sestaví následující metriky jako datový rámec s jedním řádkem:

• table_size_after_restore: Velikost tabulky po obnovení.

• num_of_files_after_restore: Počet souborů v tabulce po obnovení.

• num_removed_files: Počet odebraných souborů (logicky odstraněných) z tabulky.

• num_restored_files: Počet obnovených souborů kvůli vrácení zpět

• removed_files_size: Celková velikost v bajtech souborů, které jsou z tabulky odebrány.

• restored_files_size: Celková velikost v bajtech obnovených souborů.

  

Příklady použití pohybu v čase Delta Lake

• Oprava náhodného odstranění tabulky pro uživatele 111:

  INSERT INTO my_table
    SELECT * FROM my_table TIMESTAMP AS OF date_sub(current_date(), 1)
    WHERE userId = 111
  

• Oprava náhodných nesprávných aktualizací tabulky:

  MERGE INTO my_table target
    USING my_table TIMESTAMP AS OF date_sub(current_date(), 1) source
    ON source.userId = target.userId
    WHEN MATCHED THEN UPDATE SET *
  

• Zadejte dotaz na počet nových zákazníků přidaných za poslední týden.

  SELECT count(distinct userId)
  FROM my_table  - (
    SELECT count(distinct userId)
    FROM my_table TIMESTAMP AS OF date_sub(current_date(), 7))
  

Návody najít verzi posledního potvrzení v relaci Sparku?

Pokud chcete získat číslo verze posledního potvrzení napsaného aktuální SparkSession ve všech vláknech a všech tabulkách, zadejte dotaz na konfiguraci spark.databricks.delta.lastCommitVersionInSessionSQL .

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Scala

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Pokud nebyly provedeny SparkSessionžádné potvrzení , dotazování na klíč vrátí prázdnou hodnotu.