Arbeiten mit Tabellenverlauf

Bei Apache Iceberg- und Delta Lake-Tabellen erstellt jeder Vorgang, der eine Tabelle ändert, eine neue Tabellenversion. Verwenden Sie Verlaufsinformationen zum Überwachen von Vorgängen, zum Zurücksetzen einer Tabelle oder zum Abfragen einer Tabelle zu einem bestimmten Zeitpunkt mithilfe von Zeitreisen.

Tabellenverlauf abrufen

Führen Sie den DESCRIBE HISTORY Befehl aus, um Informationen einschließlich der Vorgänge, des Benutzers und des Zeitstempels für jeden Schreibvorgang in eine Tabelle abzurufen. Die Vorgänge werden in umgekehrter chronologischer Reihenfolge zurückgegeben.

Für die Spalten, die DESCRIBE HISTORY zurückgeben, die Werte in der Spalte operationParameters und die Metriken pro Operation in der Spalte operationMetrics , siehe Tabellengeschichtsschema und Operationsmetriken.

Die Beibehaltung des Tabellenverlaufs wird durch die Tabelleneinstellung logRetentionDurationbestimmt, die standardmäßig auf 30 Tage festgelegt ist.

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

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

Details zur Spark SQL-Syntax finden Sie unter DESCRIBE HISTORY.

Details zur Syntax von Scala, Java und Python finden Sie in der Dokumentation zur Delta Lake-API.

Katalog-Explorer zeigt den Tabellenverlauf visuell auf der Registerkarte "Verlauf " an.

Identifizieren des Vorgangstyps OPTIMIZE

Automatische Komprimierung, Liquid Clustering und Z-Sortierung werden in der Tabellenhistorie alle als OPTIMIZE-Vorgänge angezeigt. Überprüfen Sie die operationParameters-Spalte, um zu ermitteln, welches ausgeführt wurde.

Führen Sie die folgenden Schritte aus, um jeden OPTIMIZE Vorgang im Verlauf einer Tabelle zu klassifizieren:

SELECT
  version,
  timestamp,
  CASE
    WHEN operationParameters.clusterBy IS NOT NULL AND operationParameters.clusterBy <> '[]' THEN 'Liquid clustering'
    WHEN operationParameters.zOrderBy IS NOT NULL AND operationParameters.zOrderBy <> '[]' THEN 'Z-ordering'
    WHEN operationParameters.auto = 'true' THEN 'Auto compaction'
    ELSE 'Manual OPTIMIZE'
  END AS optimize_type,
  operationParameters.auto AS is_auto_compaction,
  operationParameters.clusterBy AS cluster_by,
  operationParameters.zOrderBy AS z_order_by,
  operationMetrics.numRemovedFiles AS files_compacted,
  operationMetrics.numAddedFiles AS files_added,
  operationMetrics.numRemovedBytes AS bytes_removed,
  operationMetrics.numAddedBytes AS bytes_added
FROM (DESCRIBE HISTORY table_name)
WHERE operation = 'OPTIMIZE'
ORDER BY version DESC;

In den folgenden Abschnitten werden die einzelnen operationParameters Werte ausführlich beschrieben. Definitionen der von der vorangehenden Abfrage ausgewählten operationMetricsSchlüssel finden Sie unter Vorgangsmetriken.

Automatische Komprimierung

Die automatische Komprimierung legt den auto Parameter auf true. Azure Databricks löst die automatische Komprimierung automatisch nach einem Schreibvorgang aus. Wenn auto auf false gesetzt ist, hat ein Benutzer oder ein geplanter Auftrag den Befehl OPTIMIZE ausgeführt.

Ein Autokomprimierungsvorgang zeigt z. B. Folgendes:

operationParameters: {
  "auto": "true"
}

Weitere Informationen zur automatischen Komprimierung finden Sie unter AutoKomprimierung.

Flüssigkeitsclusterbildung

Durch flüssiges Clustering wird der clusterBy Parameter mit den Namen der Clusterspalten aufgefüllt. Ein leeres clusterBy Array ([]) gibt nur die Dateikomprimierung an.

Zum Beispiel zeigt ein Vorgang, der Daten nach den Spalten date und region gruppiert hat, Folgendes:

operationParameters: {
  "clusterBy": "[\"date\",\"region\"]"
}

Weitere Informationen zum Flüssigclustering finden Sie unter Verwenden von Flüssigclustering für Tabellen.

Z-Sortierung

Die Z-Sortierung füllt den zOrderBy Parameter mit den Spaltennamen der Z-Reihenfolge auf. Ein leeres zOrderBy Array ([]) gibt an, dass der Vorgang keine Z-Sortierung angewendet hat.

Ein Vorgang, der die Z-Sortierung auf die date Spalte angewendet hat, zeigt beispielsweise Folgendes:

operationParameters: {
  "zOrderBy": "[\"date\"]"
}

Betriebsumfang

Der predicate Parameter gibt an, ob der Vorgang in der vollständigen Tabelle oder nur in einem Teil davon ausgeführt wurde:

• Ein leeres predicate Array ([]) bedeutet, dass der Vorgang auf der gesamten Tabelle ausgeführt wurde.
• Ein aufgefülltes predicate Array bedeutet, dass ein gezielter OPTIMIZE table_name WHERE <partition_predicate> Befehl nur für die Partitionen ausgeführt wird, die mit dem Prädikat übereinstimmen.

Beispielsweise zeigt ein Vorgang für die Partitionen, die mit year = 2024 übereinstimmen, Folgendes:

operationParameters: {
  "predicate": "[\"'year = 2024\"]"
}

Zeitreise

Die Zeitreise unterstützt das Abfragen früherer Tabellenversionen basierend auf der Zeitstempel- oder Tabellenversion (wie im Transaktionsprotokoll aufgezeichnet). Sie können Zeitreisen für Anwendungen wie die folgenden verwenden:

• Erneutes Erstellen von Analysen, Berichten oder Ausgaben, z. B. der Ausgabe eines Modells für maschinelles Lernen. Dies kann für das Debuggen oder die Überwachung nützlich sein, insbesondere in regulierten Branchen.
• Schreiben komplexer temporaler Abfragen.
• Beheben von Fehlern in Ihren Daten.
• Gewährleistung von Schnappschussisolation für mehrere Abfragen bei sich schnell ändernden Tabellen.

Zeitreisesyntax

Sie fragen eine Tabelle mit Zeitreise ab, indem Sie eine Klausel nach der Tabellennamenspezifikation hinzufügen.

• timestamp_expression kann einen der folgenden Werte annehmen:
  • '2018-10-18T22:15:12.013Z', d. h. eine Zeichenfolge, die zu einem Zeitstempel konvertiert werden kann
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', d. h. eine Datumszeichenfolge
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Jeder andere Ausdruck, der ein Zeitstempel ist oder in einen Zeitstempel umgewandelt werden kann
• version ist ein langer Wert, der sich aus der Ausgabe von DESCRIBE HISTORY table_spec.

Weder timestamp_expression noch version können Unterabfragen sein.

Es werden nur Datums- oder Zeitstempelzeichenfolgen akzeptiert. Beispiel: "2019-01-01" und "2019-01-01T00:00:00.000Z". Eine Beispielsyntax finden Sie im folgenden Code:

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

Sie können auch die Syntax @ verwenden, um den Zeitstempel oder die Version als Teil des Tabellennamens anzugeben. Der Zeitstempel muss im Format yyyyMMddHHmmssSSS vorliegen. Sie können eine Version mit @v. Eine Beispielsyntax finden Sie im folgenden Code:

SQL

-- Timestamp version
SELECT * FROM people10m@20190101000000000
-- Version number
SELECT * FROM people10m@v123

Python

# Timestamp version
spark.read.table("people10m@20190101000000000")
# Version number
spark.read.table("people10m@v123")

Konfigurieren der Datenaufbewahrung für Zeitreiseabfragen

Um eine frühere Tabellenversion abzufragen, müssen Sie sowohl das Protokoll als auch die Datendateien für diese Version beibehalten:

• Datendateien werden gelöscht, wenn VACUUM gegen eine Tabelle ausgeführt wird.
• Protokolldateien werden nach dem Überprüfen der Tabellenversionen automatisch entfernt.

Um den Schwellenwert für die Datenaufbewahrung für Tabellen zu erhöhen, müssen Sie die folgenden Tabelleneigenschaften konfigurieren, wobei <format> durch delta oder iceberg ersetzt wird:

• <format>.logRetentionDuration = "interval <interval>": Steuert, wie lange der Verlauf einer Tabelle aufbewahrt wird. Der Standardwert lautet interval 30 days.
  • In Databricks Runtime 18.0 und höher muss logRetentionDuration größer oder gleich deletedFileRetentionDuration sein. Bei verwalteten Tabellen im Unity-Katalog gilt dies für Databricks Runtime 12.2 und höher.
• <format>.deletedFileRetentionDuration = "interval <interval>": bestimmt den Schwellenwert, den VACUUM verwendet, um Datendateien zu entfernen, auf die in der aktuellen Tabellenversion nicht mehr verwiesen wird. Der Standardwert lautet interval 7 days.

Wenn Sie z. B. auf 30 Tage verlaufsgeschichtliche Daten zugreifen möchten, legen Sie diese fest delta.deletedFileRetentionDuration = "interval 30 days", die der Standardeinstellung entspricht delta.logRetentionDuration.

Sie können Tabelleneigenschaften während der Tabellenerstellung angeben oder mit einer ALTER TABLE Anweisung festlegen. Siehe Referenz zu Tabelleneigenschaften.

Beispiele für Zeitreisen

So beheben Sie versehentliche Löschungen in einer Tabelle für den Benutzer 111:

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

So beheben Sie versehentlich falsche Updates für eine Tabelle:

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 *

So fragen Sie die Anzahl der neuen Kunden ab, die in der letzten Woche hinzugefügt wurden:

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

Transaktionsprotokollprüfpunkte

Das Transaktionsprotokoll zeichnet Tabellenversionen als JSON-Dateien im Transaktionsprotokollverzeichnis zusammen mit Tabellendaten auf.

Um das Abfragen von Checkpoints zu optimieren, werden Tabellenversionen in Parquet-Checkpoint-Dateien zusammengefasst, was die Leistung verbessert, da so nicht alle JSON-Versionen der Tabellenhistorie gelesen werden müssen. Benutzer müssen nicht direkt mit Prüfpunkten interagieren.

Azure Databricks optimiert die Prüfpunkthäufigkeit für Datengröße und Workload. Die Prüfpunkthäufigkeit kann sich ohne Ankündigung ändern.

Wiederherstellen einer Tabelle in einem früheren Zustand

Verwenden Sie den RESTORE Befehl, um eine Tabelle auf eine frühere Version oder einen Zeitstempel wiederherzustellen, einschließlich für diese Szenarien:

• Sie können eine Tabelle, die bereits wiederhergestellt wurde, erneut wiederherstellen.
• Sie können eine geklonte Tabelle wiederherstellen.

Berücksichtigen Sie die folgenden Anforderungen:

• Um eine Tabelle wiederherzustellen, müssen Sie über die Berechtigung für die Tabelle verfügen MODIFY .
• Nachdem Datendateien manuell oder durch VACUUM gelöscht wurden, können Sie eine Tabelle nicht auf eine ältere Version zurücksetzen, die auf diese Dateien verweist. Die teilweise Wiederherstellung auf diese Version ist weiterhin möglich, wenn spark.sql.files.ignoreMissingFiles auf true festgelegt wird.
• Verwenden Sie zum Wiederherstellen nach Zeitstempel die Formate yyyy-MM-dd HH:mm:ss oder yyyy-MM-dd.

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

Ausführliche Informationen zur Syntax finden Sie unter RESTORE.

Streamingverhalten

Die Wiederherstellung ist ein datenverändernder Vorgang und kann zu doppelten Daten für nachgeschaltete Workloads führen. Vom Befehl hinzugefügte RESTORE Protokolleinträge enthalten "dataChange", das auf "true" gesetzt ist.

Bei nachgelagerten Workloads, z. B. einem strukturierten Streamingauftrag , der die Aktualisierungen an einer Tabelle verarbeitet, werden die vom Wiederherstellungsvorgang hinzugefügten Datenänderungsprotokolleinträge als neue Datenaktualisierungen betrachtet, und die Verarbeitung kann zu doppelten Daten führen.

Beispiel:

Im vorherigen Beispiel führt der RESTORE Befehl zu Updates, die beim Lesen der Tabellenversion 0 und 1 zuvor angezeigt wurden. Wenn eine Streamingabfrage diese Tabelle erneut liest, werden diese Dateien als neu hinzugefügte Daten betrachtet und erneut verarbeitet.

Wiederherstellen von Metriken

Nach Abschluss meldet RESTORE die folgenden Metriken als einzeiligen DataFrame:

• table_size_after_restore: Die Größe der Tabelle nach der Wiederherstellung.

• num_of_files_after_restore Die Anzahl von Dateien in der Tabelle nach der Wiederherstellung.

• num_removed_files: Die Anzahl von Dateien, die aus der Tabelle entfernt (logisch gelöscht) wurden.

• num_restored_files: Die Anzahl von Dateien, die aufgrund eines Rollbacks wiederhergestellt wurden.

• removed_files_size: Die Gesamtgröße in Bytes der aus der Tabelle entfernten Dateien.

• restored_files_size: Die Gesamtgröße in Bytes der wiederhergestellten Dateien.

  

Letzte Commit-Version suchen

Um die Versionsnummer des letzten Commits zu erhalten, der von der aktuellen SparkSession für alle Threads und alle Tabellen geschrieben wurde, fragen Sie die SQL-Konfiguration spark.databricks.<format>.lastCommitVersionInSession ab. Ersetzen Sie <format> je nach Tabellenformat durch delta oder iceberg.

Beispiel:

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Wenn von der SparkSession keine Commits durchgeführt wurden, wird beim Abfragen des Schlüssels ein leerer Wert zurückgegeben.