Verwenden des Delta Lake-Tabellenverlaufs

Jeder Vorgang, der eine Delta Lake-Tabelle ändert, hat die Erstellung einer neuen Tabellenversion zur Folge. Sie können Verlaufsinformationen nutzen, um Vorgänge zu überwachen, ein Rollback für eine Tabelle durchzuführen oder eine Tabelle zu einem bestimmten Zeitpunkt mittels Zeitreise abzufragen.

Abrufen des Delta-Tabellenverlaufs

Sie können Informationen zu den Vorgängen, zum Benutzer und zum Zeitstempel für jeden Schreibvorgang in eine Delta-Tabelle abrufen, indem Sie den history-Befehl ausführen. Die Vorgänge werden in umgekehrter chronologischer Reihenfolge zurückgegeben.

Die Beibehaltung des Tabellenverlaufs wird durch die Tabelleneinstellung delta.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.

Der Katalog-Explorer bietet eine visuelle Darstellung dieser detaillierten Tabelleninformationen und des Verlaufs für Deltatabellen. Sie können auf die Registerkarte Verlauf klicken, um zusätzlich zu Tabellenschema und Beispieldaten den mit DESCRIBE HISTORY angezeigten Tabellenverlauf zu sehen.

Verlaufsschema

Die Ausgabe des Vorgangs history enthält die folgenden Spalten.

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

Vorgangsmetrikschlüssel

Der Vorgang history gibt eine Auflistung von Vorgangsmetriken in der Spaltenzuordnung operationMetrics zurück.

In den folgenden Tabellen sind die Zuordnungsschlüsseldefinitionen nach Vorgang aufgeführt.

Was ist die Delta Lake-Zeitreise?

Delta Lake unterstützt das Abfragen früherer Tabellenversionen basierend auf dem Zeitstempel oder der Tabellenversion (gemäß Aufzeichnung im Transaktionsprotokoll). Sie können Zeitreisen für Anwendungen wie die folgenden verwenden:

• Erneutes Erstellen von Analysen, Berichten oder Ausgaben (z. B. der Ausgabe eines Machine Learning-Modells). Dies könnte für das Debuggen oder die Überwachung nützlich sein, insbesondere in regulierten Branchen.
• Schreiben komplexer temporaler Abfragen.
• Beheben von Fehlern in Ihren Daten.
• Bereitstellen der Momentaufnahmenisolation für eine Reihe von Abfragen für sich schnell verändernde Tabellen.

Syntax für die Delta-Zeitreise

Sie fragen eine Delta-Tabelle mit Zeitreise ab, indem Sie eine Klausel nach der Spezifikation des Tabellennamens hinzufügen.

• timestamp_expression kann einen der folgenden Werte annehmen:
  • '2018-10-18T22:15:12.013Z', d. h. eine Zeichenfolge, die in einen Zeitstempel umgewandelt 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 LONG-Wert, der aus der Ausgabe von DESCRIBE HISTORY table_spec abgerufen werden kann.

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 nach @ angeben, indem Sie der Version v voranstellen. Eine Beispielsyntax finden Sie im folgenden Code:

SQL

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

Python

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

Was sind Transaktionsprotokoll-Prüfpunkte?

Delta Lake erfasst Tabellenversionen als JSON-Dateien im Verzeichnis _delta_log, das parallel zu Tabellendaten gespeichert wird. Zur Optimierung von Prüfpunktabfragen aggregiert Delta Lake Tabellenversionen in Parquet-Prüfpunktdateien, damit nicht alle JSON-Versionen des Tabellenverlaufs gelesen werden müssen. Azure Databricks optimiert die Prüfpunkthäufigkeit für Datengröße und Workload. Benutzer sollten nicht direkt mit Prüfpunkten interagieren müssen. Die Prüfpunkthäufigkeit kann sich ohne Ankündigung ändern.

Konfigurieren der Datenaufbewahrung für Zeitreiseabfragen

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

Datendateien werden gelöscht, wenn VACUUM gegen eine Tabelle ausgeführt wird. Delta Lake verwaltet das automatische Entfernen von Protokolldateien nach dem Erstellen eines Prüfpunkts der Tabellenversionen.

Da gegen meisten Delta-Tabellen regelmäßig VACUUM ausgeführt wird, sollten Point-in-Time-Abfragen den Aufbewahrungsschwellenwert für VACUUMberücksichtigen, der standardmäßig 7 Tage beträgt.

Um den Schwellenwert für die Datenaufbewahrung für Delta-Tabellen zu erhöhen, müssen Sie die folgenden Tabelleneigenschaften konfigurieren:

• delta.logRetentionDuration = "interval <interval>": Steuert, wie lange der Verlauf einer Tabelle aufbewahrt wird. Der Standardwert lautet interval 30 days.
• delta.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.

Sie können Delta-Eigenschaften während der Tabellenerstellung angeben oder mit einer ALTER TABLE-Anweisung festlegen. Weitere Informationen finden Sie unter Referenz zu Delta-Tabelleneigenschaften.

Wiederherstellen eines früheren Zustands einer Delta-Tabelle

Mit dem Befehl RESTORE können Sie den früheren Zustand einer Delta-Tabelle wiederherstellen. Eine Delta-Tabelle verwaltet intern ihre Verlaufsversionen, mit denen sie auf einen früheren Zustand wiederhergestellt werden kann. Eine Version, die dem früheren Zustand entspricht, oder ein Zeitstempel von der Erstellung des früheren Zustands werden durch den Befehl RESTORE als Optionen unterstützt.

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

Details zur Syntax finden Sie unter RESTORE.

Wiederherstellen von Metriken

RESTORE meldet die folgenden Metriken als einzeiligen Datenrahmen (DataFrame), sobald der Vorgang abgeschlossen ist:

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

  

Beispiele für die Verwendung von Delta Lake-Zeitreisen

• Korrigieren versehentlicher 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
  

• Korrigieren versehentlicher falscher 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 *
  

• Abfragen der Anzahl von Neukunden, die in der letzten Woche gewonnen wurden.

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

Wie finde ich die Version des letzten Commits in der Spark-Sitzung?

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.delta.lastCommitVersionInSession ab.

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.