Usare la cronologia delle tabelle

Per le tabelle Apache Iceberg e Delta Lake, ogni operazione che modifica una tabella crea una nuova versione della tabella. Usa le informazioni sulla cronologia per verificare le operazioni, eseguire il rollback di una tabella o interrogare una tabella in uno specifico momento usando il time travel.

Recuperare la cronologia delle tabelle

Eseguire il DESCRIBE HISTORY comando per recuperare informazioni, incluse le operazioni, l'utente e il timestamp per ogni scrittura in una tabella. Le operazioni vengono restituite in ordine cronologico inverso.

Per le colonne che DESCRIBE HISTORY restituiscono, i valori nella operationParameters colonna e le metriche per operazione nella operationMetrics colonna, vedi Schema della storia della tabella e metriche di operazione.

La conservazione della cronologia delle tabelle è determinata dall'impostazione della tabella logRetentionDuration, ovvero 30 giorni per impostazione predefinita.

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

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

Per informazioni dettagliate sulla sintassi di Spark SQL, vedere DESCRIBE HISTORY.

Per informazioni dettagliate sulla sintassi di Scala, Java e Python, vedere la documentazione dell'API Delta Lake.

Esplora cataloghi mostra visivamente la cronologia delle tabelle nella scheda Cronologia .

Identificare il tipo di OPTIMIZE operazione

La compattazione automatica, il clustering liquido e l'ordinamento Z vengono visualizzati nella cronologia delle tabelle come OPTIMIZE operazioni. Per determinare quale è stata eseguita, esamina la colonna operationParameters.

Per classificare ogni OPTIMIZE operazione nella cronologia di una tabella, eseguire le operazioni seguenti:

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;

Le sezioni seguenti descrivono in dettaglio ogni operationParameters valore. Per le definizioni delle operationMetrics chiavi selezionate dalla query precedente, vedi Metriche operative.

Compattazione automatica

La compattazione automatica imposta il auto parametro su true. Azure Databricks attiva automaticamente la compattazione automatica dopo una scrittura. Quando auto è false, un utente o un processo pianificato ha eseguito il OPTIMIZE comando .

Ad esempio, un'operazione di compattazione automatica mostra quanto segue:

operationParameters: {
  "auto": "true"
}

Per altre informazioni sulla compattazione automatica, vedere Compattazione automatica.

Raggruppamento liquido

Liquid clustering popola il parametro clusterBy con i nomi delle colonne di clustering. Una matrice vuota clusterBy ([]) indica solo la compattazione dei file.

Ad esempio, un'operazione che raggruppa i dati in base alle date colonne e region mostra quanto segue:

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

Per altre informazioni sul clustering liquido, vedere Usare il clustering liquido per le tabelle.

Ordinamento Z

L'ordinamento Z popola il zOrderBy parametro con i nomi delle colonne dell'ordine Z. Una matrice vuota zOrderBy ([]) indica che l'operazione non ha applicato l'ordinamento Z.

Ad esempio, un'operazione che ha applicato l'ordinamento Z nella date colonna mostra quanto segue:

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

Ambito dell'operazione

Il predicate parametro indica se l'operazione è stata eseguita nella tabella completa o solo in parte:

• Una matrice vuota predicate ([]) indica che l'operazione è stata eseguita sull'intera tabella.
• Un array predicate popolato significa che un comando OPTIMIZE table_name WHERE <partition_predicate> mirato è stato eseguito solo sulle partizioni che corrispondono al predicato.

Ad esempio, un'operazione rivolta alle partizioni corrispondenti a year = 2024 mostra quanto segue:

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

Spostamento cronologico

Il viaggio nel tempo consente l'esecuzione di query sulle versioni precedenti della tabella in base al timestamp o alla versione della tabella, come registrato nel log delle transazioni. È possibile usare il tempo di viaggio per le applicazioni, ad esempio le seguenti:

• Ricreare analisi, report o output, ad esempio l'output di un modello di apprendimento automatico. Ciò può essere utile per il debug o il controllo, in particolare nei settori regolamentati.
• Scrivere query temporali complesse.
• Correggere gli errori nei dati.
• Fornire isolamento dello snapshot per un set di query per tabelle a modifica rapida.

Sintassi di spostamento temporale

Per eseguire una query su una tabella con time travel, aggiungere una clausola dopo la specifica del nome della tabella.

• timestamp_expression può essere uno qualsiasi di:
  • '2018-10-18T22:15:12.013Z', ovvero una stringa che può essere convertita in un timestamp
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', ovvero una stringa di data
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Qualsiasi altra espressione che è o può essere convertita in un timestamp
• version è un valore long che può essere ottenuto dall'output di DESCRIBE HISTORY table_spec.

Né timestamp_expression né version possono essere sottoquery.

Vengono accettate solo stringhe di data o timestamp. Ad esempio, "2019-01-01" e "2019-01-01T00:00:00.000Z". Vedere il codice seguente per una sintassi di esempio:

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

È anche possibile usare la @ sintassi per specificare il timestamp o la versione come parte del nome della tabella. Il timestamp deve essere in yyyyMMddHHmmssSSS formato . È possibile specificare una versione con @v. Vedere il codice seguente per una sintassi di esempio:

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

Configurare la conservazione dei dati per le query di spostamento del tempo

Per eseguire una query su una versione precedente della tabella, è necessario conservare sia il log che i file di dati per tale versione:

• I file di dati vengono eliminati quando VACUUM viene eseguito su una tabella.
• I file di log vengono rimossi automaticamente dopo il checkpoint delle versioni delle tabelle.

Per aumentare la soglia di conservazione dei dati per le tabelle, è necessario configurare le proprietà della tabella seguenti, sostituendo <format> con delta o iceberg:

• <format>.logRetentionDuration = "interval <interval>": controlla per quanto tempo viene mantenuta la cronologia di una tabella. Il valore predefinito è interval 30 days.
  • In Databricks Runtime 18.0 e versioni successive deve logRetentionDuration essere maggiore o uguale a deletedFileRetentionDuration. Per le tabelle gestite di Unity Catalog, questo vale per Databricks Runtime 12.2 e versioni successive.
• <format>.deletedFileRetentionDuration = "interval <interval>": determina l'utilizzo della soglia VACUUM per rimuovere i file di dati a cui non si fa più riferimento nella versione della tabella corrente. Il valore predefinito è interval 7 days.

Ad esempio, per accedere a 30 giorni di dati cronologici, impostare delta.deletedFileRetentionDuration = "interval 30 days", che corrisponde all'impostazione predefinita per delta.logRetentionDuration.

È possibile specificare le proprietà della tabella durante la creazione della tabella o impostarle con un'istruzione ALTER TABLE . Vedere Informazioni di riferimento sulle proprietà della tabella.

Esempi di viaggi in tempo

Per correggere le eliminazioni accidentali in una tabella per l'utente 111:

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

Per correggere gli aggiornamenti accidentali non corretti in una tabella:

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 *

Per eseguire una query sul numero di nuovi clienti aggiunti nell'ultima settimana:

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

Checkpoint del log delle transazioni

Il log delle transazioni registra le versioni della tabella come file JSON all'interno della directory del log delle transazioni insieme ai dati della tabella.

Per ottimizzare l'esecuzione di query di checkpoint, le versioni delle tabelle vengono aggregate ai file di checkpoint Parquet, che migliorano le prestazioni impedendo la necessità di leggere tutte le versioni JSON della cronologia delle tabelle. Gli utenti non devono interagire direttamente con i checkpoint.

Azure Databricks ottimizza la frequenza di checkpoint per le dimensioni dei dati e il carico di lavoro. La frequenza del checkpoint è soggetta a modifiche senza preavviso.

Ripristinare uno stato precedente di una tabella

Usare il RESTORE comando per ripristinare una tabella in una versione o un timestamp precedente, incluso per questi scenari:

• È possibile ripristinare una tabella già ripristinata.
• È possibile ripristinare una tabella clonata .

Considerare i requisiti seguenti:

• Per ripristinare una tabella, è necessario disporre MODIFY dell'autorizzazione per la tabella.
• Dopo l'eliminazione dei file di dati, manualmente o da VACUUM, non è possibile ripristinare una tabella in una versione precedente che fa riferimento a tali file. Il ripristino in questa versione parzialmente è comunque possibile se spark.sql.files.ignoreMissingFiles è impostato su true.
• Per eseguire il ripristino in base al timestamp, usare i formati yyyy-MM-dd HH:mm:ss o yyyy-MM-dd.

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

Per informazioni dettagliate sulla sintassi, vedere RESTORE.

Comportamento di streaming

Il ripristino è un'operazione di modifica dei dati e potrebbe comportare dati duplicati per i carichi di lavoro downstream. Le voci di log aggiunte dal RESTORE comando contengono dataChange impostato su true.

Per i carichi di lavoro downstream, ad esempio un processo di streaming strutturato che elabora gli aggiornamenti a una tabella, le voci del log delle modifiche dei dati aggiunte dall'operazione di ripristino sono considerate nuovi aggiornamenti dei dati e l'elaborazione può comportare dati duplicati.

Per esempio:

Nell'esempio precedente, il RESTORE comando restituisce gli aggiornamenti visualizzati in precedenza durante la lettura della tabella 0 e 1. Se una query di streaming legge nuovamente questa tabella, questi file vengono considerati come dati appena aggiunti e vengono elaborati di nuovo.

Ripristinare le metriche

Al termine dell'operazione, RESTORE riporta le seguenti metriche in un DataFrame a riga singola:

• table_size_after_restore: dimensioni della tabella dopo il ripristino.

• num_of_files_after_restore: numero di file nella tabella dopo il ripristino.

• num_removed_files: numero di file rimossi (eliminati logicamente) dalla tabella.

• num_restored_files: Numero di file ripristinati a seguito di rollback.

• removed_files_size: dimensione totale in byte dei file rimossi dalla tabella.

• restored_files_size: dimensioni totali in byte dei file ripristinati.

  

Trovare l'ultima versione del commit

Per ottenere il numero di versione dell'ultimo commit scritto dall'oggetto corrente SparkSession in tutti i thread e in tutte le tabelle, eseguire una query sulla configurazione SQL spark.databricks.<format>.lastCommitVersionInSession. Sostituire <format> con delta o iceberg, a seconda del formato della tabella.

Per esempio:

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Se non sono stati eseguiti commit da SparkSession, l'esecuzione della query sulla chiave restituisce un valore vuoto.