Werken met tabelgeschiedenis

Voor Apache Iceberg- en Delta Lake-tabellen maakt elke bewerking waarmee een tabel wordt gewijzigd een nieuwe tabelversie. Gebruik geschiedenisgegevens om bewerkingen te controleren, een tabel terug te draaien of een query uit te voeren op een bepaald tijdstip met behulp van tijdreizen.

Tabelgeschiedenis ophalen

Voer de DESCRIBE HISTORY opdracht uit om informatie op te halen, waaronder de bewerkingen, de gebruiker en de tijdstempel voor elke schrijfbewerking naar een tabel. De bewerkingen worden geretourneerd in omgekeerde chronologische volgorde.

Voor de kolommen die DESCRIBE HISTORY retourneren, de waarden in de kolom operationParameters en de metrische gegevens per bewerking in de kolom operationMetrics, zie Schema voor tabelgeschiedenis en metrische gegevens voor bewerkingen.

Retentie van tabelgeschiedenis wordt bepaald door de tabelinstelling logRetentionDuration, die standaard 30 dagen is.

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

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

Zie DESCRIBE HISTORYvoor details van spark SQL-syntaxis.

Zie de documentatie van de Delta Lake-API voor Scala, Java en Python syntaxis.

Catalogusverkenner toont tabelgeschiedenis visueel op het tabblad Geschiedenis .

Het type OPTIMIZE bewerking identificeren

Automatische compressie, vloeistofclustering en Z-volgorde worden allemaal als bewerkingen in de tabelgeschiedenis OPTIMIZE weergegeven. Inspecteer de kolom operationParameters om te bepalen welke werd uitgevoerd.

Als u elke OPTIMIZE bewerking in de geschiedenis van een tabel wilt classificeren, voert u het volgende uit:

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 de volgende secties wordt elke operationParameters waarde gedetailleerd beschreven. Zie Bewerkingsstatistieken voor definities van de operationMetrics sleutels die door de voorgaande query worden geselecteerd.

Automatische compressie

Met automatische compressie wordt de auto parameter ingesteld op true. Azure Databricks activeert automatische compactie automatisch na een schrijfbewerking. Wanneer auto is false, heeft een gebruiker of geplande taak de OPTIMIZE opdracht uitgevoerd.

Een automatische compressiebewerking toont bijvoorbeeld het volgende:

operationParameters: {
  "auto": "true"
}

Zie Auto comprimeren voor meer informatie over automatische compressie.

Clusteren van vloeistoffen

Met liquide clustering wordt de clusterBy parameter gevuld met de clusterkolomnamen. Een lege clusterBy matrix ([]) geeft alleen bestandscompressie aan.

Bijvoorbeeld, een bewerking die gegevens op basis van de kolommen date en region heeft gegroepeerd, ziet er als volgt uit:

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

Zie Liquid Clustering gebruiken voor tabellen voor meer informatie over liquide clustering.

Z-volgorde

Z-ordering vult de zOrderBy parameter met de kolomnamen van Z-volgorde. Een lege zOrderBy matrix ([]) geeft aan dat de bewerking geen Z-volgorde heeft toegepast.

Een bewerking die Z-volgorde in de date kolom heeft toegepast, toont bijvoorbeeld het volgende:

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

Werkingsbereik

De predicate parameter geeft aan of de bewerking is uitgevoerd in de volledige tabel of slechts een deel ervan:

• Een lege predicate matrix ([]) betekent dat de bewerking in de hele tabel is uitgevoerd.
• Een gevulde predicate matrix betekent dat een doelopdracht OPTIMIZE table_name WHERE <partition_predicate> alleen wordt uitgevoerd op de partities die overeenkomen met het predicaat.

Een bewerking die is gericht op de partities die overeenkomen year = 2024 , toont bijvoorbeeld het volgende:

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

Tijdreizen

Met tijdreizen kan men eerdere tabelversies opvragen op basis van tijdstempel of tabelversie (zoals vastgelegd in het transactielogboek). U kunt tijdreizen gebruiken voor toepassingen zoals:

• Analyses, rapporten of uitvoer opnieuw maken, zoals de uitvoer van een machine learning-model. Dit kan handig zijn voor foutopsporing of controle, met name in gereglementeerde branches.
• Complexe tijdelijke query's schrijven.
• Fouten in uw gegevens corrigeren.
• Het bieden van momentopname-isolatie voor een set query's voor snel wijzigende tabellen.

Syntaxis van tijdreizen

U kunt een query uitvoeren op een tabel met tijdreizen door een component toe te voegen na de tabelnaamspecificatie.

• timestamp_expression kan een van de volgende zijn:
  • '2018-10-18T22:15:12.013Z', dat wil zeggen, een tekenreeks die kan worden omgezet naar een tijdstempel
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', dat wil gezegd, een datumtekenreeks
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Elke andere expressie die wel of niet kan worden omgezet in een tijdstempel
• version is een lange waarde die kan worden verkregen uit de uitvoer van DESCRIBE HISTORY table_spec.

Noch timestamp_expression noch version kan een subquery zijn.

Alleen datum- of tijdstempeltekenreeksen worden geaccepteerd. Bijvoorbeeld, "2019-01-01" en "2019-01-01T00:00:00.000Z". Zie de volgende code voor voorbeeldsyntaxis:

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

U kunt de @ syntaxis ook gebruiken om de tijdstempel of versie op te geven als onderdeel van de tabelnaam. De tijdstempel moet een yyyyMMddHHmmssSSS indeling hebben. U kunt een versie opgeven met @v. Zie de volgende code voor voorbeeldsyntaxis:

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

Gegevensretentie configureren voor tijdgebonden query's

Als u een query wilt uitvoeren op een eerdere tabelversie, moet u zowel het logboek als de gegevensbestanden voor die versie behouden:

• Gegevensbestanden worden verwijderd wanneer VACUUM een tabel uitvoert.
• Logbestanden worden automatisch verwijderd nadat er controlepunten voor tabelversies zijn gemaakt.

Als u de drempelwaarde voor gegevensretentie voor tabellen wilt verhogen, moet u de volgende tabeleigenschappen configureren, waarbij u deze vervangt door <format> een delta of iceberg:

• <format>.logRetentionDuration = "interval <interval>": bepaalt hoe lang de geschiedenis voor een tabel wordt bewaard. De standaardwaarde is interval 30 days.
  • In Databricks Runtime 18.0 en hoger logRetentionDuration moet deze groter zijn dan of gelijk zijn aan deletedFileRetentionDuration. Voor beheerde tabellen van Unity Catalog is dit van toepassing op Databricks Runtime 12.2 en hoger.
• <format>.deletedFileRetentionDuration = "interval <interval>": bepaalt de drempelwaarde VACUUM die wordt gebruikt om gegevensbestanden te verwijderen waarnaar niet meer wordt verwezen in de huidige tabelversie. De standaardwaarde is interval 7 days.

Als u bijvoorbeeld 30 dagen aan historische gegevens wilt openen, stelt u deze in delta.deletedFileRetentionDuration = "interval 30 days", die overeenkomt met de standaardinstelling voor delta.logRetentionDuration.

U kunt tabeleigenschappen opgeven tijdens het maken van een tabel of deze instellen met een ALTER TABLE instructie. Zie naslaginformatie over tabeleigenschappen.

Voorbeelden van tijdreizen

Onopzettelijke verwijderingen herstellen in een tabel voor de gebruiker 111:

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

Per ongeluk onjuiste wijzigingen in een tabel oplossen:

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 *

Een query uitvoeren op het aantal nieuwe klanten dat in de afgelopen week is toegevoegd:

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

Controlepunten voor transactielogboeken

In het transactielogboek worden tabelversies vastgelegd als JSON-bestanden in de map met transactielogboeken naast tabelgegevens.

Om controlepuntquery's te optimaliseren, worden tabelversies samengevoegd met Parquet-controlepuntbestanden, waardoor de prestaties worden verbeterd door te voorkomen dat alle JSON-versies van de tabelgeschiedenis moeten worden gelezen. Gebruikers hoeven niet rechtstreeks met controlepunten te communiceren.

Azure Databricks optimaliseert de frequentie van controlepunten voor gegevensgrootte en workload. De controlepuntfrequentie kan zonder kennisgeving worden gewijzigd.

Een tabel herstellen naar een eerdere status

Gebruik de RESTORE opdracht om een tabel te herstellen naar een eerdere versie of tijdstempel, waaronder voor deze scenario's:

• U kunt een al herstelde tabel herstellen.
• U kunt een gekloonde tabel herstellen.

Houd rekening met de volgende vereisten:

• Als u een tabel wilt herstellen, moet u gemachtigd zijn MODIFY voor de tabel.
• Nadat gegevensbestanden handmatig of per VACUUMzijn verwijderd, kunt u een tabel niet herstellen naar een oudere versie die verwijst naar die bestanden. Herstellen naar deze versie is gedeeltelijk nog steeds mogelijk als spark.sql.files.ignoreMissingFiles is ingesteld op true.
• Als u wilt herstellen op tijdstempel, gebruikt u de indelingen yyyy-MM-dd HH:mm:ss of yyyy-MM-dd.

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

Zie RESTOREvoor syntaxisdetails.

Streaminggedrag

Herstellen is een bewerking voor het wijzigen van gegevens en kan leiden tot dubbele gegevens voor downstreamworkloads. Logboekvermeldingen die door de RESTORE opdracht zijn toegevoegd, bevatten dataChange ingesteld op true.

Voor downstream-workloads, zoals een Structured Streaming-taak die de updates in een tabel verwerkt, worden de logboekvermeldingen van gegevenswijzigingen die door de herstelbewerking zijn toegevoegd, beschouwd als nieuwe gegevensupdates, en kan de verwerking ervan resulteren in dubbele gegevens.

Voorbeeld:

In het voorgaande voorbeeld resulteert de RESTORE opdracht in updates die eerder werden gezien bij het lezen van de tabelversie 0 en 1. Als een streamingquery deze tabel opnieuw leest, worden deze bestanden beschouwd als nieuw toegevoegde gegevens en worden deze opnieuw verwerkt.

Metrische gegevens herstellen

Nadat dit is voltooid, rapporteert RESTORE de volgende metrische gegevens als een DataFrame met één rij:

• table_size_after_restore: De grootte van de tabel na het herstellen.

• num_of_files_after_restore: Het aantal bestanden in de tabel na het herstellen.

• num_removed_files: Het aantal bestanden dat uit de tabel is verwijderd (logisch verwijderd).

• num_restored_files: Aantal bestanden teruggezet door het terugdraaien van wijzigingen.

• removed_files_size: Totale grootte in bytes van de bestanden die uit de tabel worden verwijderd.

• restored_files_size: Totale grootte in bytes van de bestanden die worden hersteld.

  

Zoek de laatste commitversie

Als u het versienummer van de laatste doorvoering wilt ophalen die door de huidige SparkSession is geschreven in alle threads en alle tabellen, voert u een query uit op de SQL-configuratie spark.databricks.<format>.lastCommitVersionInSession. Vervang <format> door delta of iceberg, afhankelijk van de indeling van uw tabel.

Voorbeeld:

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Als er geen commits zijn uitgevoerd door de SparkSession, retourneert een query op de sleutel een lege waarde.