Arbeta med Delta Lake-tabellhistorik

Varje åtgärd som ändrar en Delta Lake-tabell skapar en ny tabellversion. Du kan använda historikinformation för att granska åtgärder, återställa en tabell eller köra frågor mot en tabell vid en viss tidpunkt med hjälp av tidsresor.

Hämta historik för deltatabell

Du kan hämta information, inklusive åtgärder, användare och tidsstämpel för varje skrivning till en Delta-tabell genom att köra history kommandot . Åtgärderna returneras i omvänd kronologisk ordning.

Kvarhållning av tabellhistorik bestäms av tabellinställningen delta.logRetentionDuration, som är 30 dagar som standard.

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

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

Information om Spark SQL-syntax finns i BESKRIVA HISTORIK.

Se Delta Lake API-dokumentationen för Scala/Java/Python-syntaxinformation.

Katalogutforskaren innehåller en visuell vy över den här detaljerade tabellinformationen och historiken för Delta-tabeller. Förutom tabellschemat och exempeldata kan du klicka på fliken Historik för att se tabellhistoriken som visas med DESCRIBE HISTORY.

Historikschema

Utdata för åtgärden history har följande kolumner.

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|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...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

Åtgärdsmåttnycklar

Åtgärden history returnerar en samling åtgärdsmått i kolumnkartan operationMetrics .

I följande tabeller visas mappningsnyckeldefinitionerna efter åtgärd.

Vad är Delta Lake-tidsresor?

Delta Lake-tidsresor stöder frågor mot tidigare tabellversioner baserat på tidsstämpel eller tabellversion (som registrerats i transaktionsloggen). Du kan använda tidsresor för program, till exempel följande:

• Återskapa analyser, rapporter eller utdata (till exempel utdata från en maskininlärningsmodell). Detta kan vara användbart för felsökning eller granskning, särskilt i reglerade branscher.
• Skriva komplexa temporala frågor.
• Åtgärda misstag i dina data.
• Tillhandahålla ögonblicksbildisolering för en uppsättning frågor för snabbväxande tabeller.

Syntax för Delta-tidsresor

Du kör frågor mot en Delta-tabell med tidsåtgång genom att lägga till en sats efter tabellnamnspecifikationen.

• timestamp_expression kan vara något av följande:
  • '2018-10-18T22:15:12.013Z', det vill: en sträng som kan gjutas till en tidsstämpel
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', det vill: en datumsträng
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Andra uttryck som är eller kan gjutas till en tidsstämpel
• version är ett långt värde som kan hämtas från utdata DESCRIBE HISTORY table_specfrån .

Varken timestamp_expression eller version kan vara underfrågor.

Endast datum- eller tidsstämpelsträngar accepteras. Till exempel "2019-01-01" och "2019-01-01T00:00:00.000Z". Se följande kod för exempelsyntax:

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")

Du kan också använda syntaxen @ för att ange tidsstämpeln eller versionen som en del av tabellnamnet. Tidsstämpeln måste vara i yyyyMMddHHmmssSSS format. Du kan ange en version efter @ genom att lägga till en v till versionen. Se följande kod för exempelsyntax:

SQL

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

Python

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

Vad är kontrollpunkter för transaktionsloggar?

Delta Lake registrerar tabellversioner som JSON-filer i _delta_log katalogen, som lagras tillsammans med tabelldata. För att optimera frågor om kontrollpunkter aggregerar Delta Lake tabellversioner till Parquet-kontrollpunktsfiler, vilket förhindrar att alla JSON-versioner av tabellhistoriken behöver läsas. Azure Databricks optimerar kontrollpunktsfrekvensen för datastorlek och arbetsbelastning. Användare bör inte behöva interagera direkt med kontrollpunkter. Kontrollpunktsfrekvensen kan komma att ändras utan föregående meddelande.

Konfigurera datakvarhållning för frågor om tidsresor

Om du vill köra frågor mot en tidigare tabellversion måste du behålla både loggen och datafilerna för den versionen.

Datafiler tas bort när VACUUM de körs mot en tabell. Delta Lake hanterar borttagning av loggfiler automatiskt efter kontrollpunktstabellversioner.

Eftersom de flesta Delta-tabeller har VACUUM körts mot dem regelbundet bör punkt-i-tid-frågor respektera kvarhållningströskeln för VACUUM, vilket är 7 dagar som standard.

För att öka tröskelvärdet för datakvarhållning för Delta-tabeller måste du konfigurera följande tabellegenskaper:

• delta.logRetentionDuration = "interval <interval>": styr hur länge historiken för en tabell sparas. Standardvärdet är interval 30 days.
• delta.deletedFileRetentionDuration = "interval <interval>": anger tröskelvärdet VACUUM som används för att ta bort datafiler som inte längre refereras till i den aktuella tabellversionen. Standardvärdet är interval 7 days.

Du kan ange Delta-egenskaper när tabellen skapas eller ange dem med en ALTER TABLE instruktion. Se Referens för egenskaper för Delta-tabell.

Återställa en Delta-tabell till ett tidigare tillstånd

Du kan återställa en Delta-tabell till dess tidigare tillstånd med hjälp RESTORE av kommandot . En Delta-tabell underhåller internt historiska versioner av tabellen som gör att den kan återställas till ett tidigare tillstånd. En version som motsvarar det tidigare tillståndet eller en tidsstämpel för när det tidigare tillståndet skapades stöds som alternativ med kommandot RESTORE.

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

Syntaxinformation finns i ÅTERSTÄLL.

Återställa mått

RESTORE rapporterar följande mått som en dataram på en rad när åtgärden är klar:

• table_size_after_restore: Tabellens storlek efter återställningen.

• num_of_files_after_restore: Antalet filer i tabellen efter återställningen.

• num_removed_files: Antal filer som tagits bort (logiskt borttagna) från tabellen.

• num_restored_files: Antal filer som återställts på grund av återställning.

• removed_files_size: Total storlek i byte för de filer som tas bort från tabellen.

• restored_files_size: Total storlek i byte för de filer som återställs.

  

Exempel på användning av Delta Lake-tidsresor

• Åtgärda oavsiktliga borttagningar till en tabell för användaren 111:

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

• Åtgärda oavsiktliga felaktiga uppdateringar av en tabell:

  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 *
  

• Fråga antalet nya kunder som lagts till under den senaste veckan.

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

Hur hittar jag den senaste incheckningens version i Spark-sessionen?

Om du vill hämta versionsnumret för den senaste incheckningen som skrivits av den aktuella SparkSession i alla trådar och alla tabeller frågar du SQL-konfigurationen spark.databricks.delta.lastCommitVersionInSession.

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Om inga incheckningar har gjorts av SparkSessionreturnerar frågan till nyckeln ett tomt värde.