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.

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 .

Schema cronologia

L'output dell'operazione history include le colonne seguenti.

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

Informazioni sui partitionBy parametri dell'operazione

Il partitionBy campo nella cronologia tabelle è significativo solo per le operazioni CREATE e OVERWRITE che definiscono o modificano lo schema di partizione di una tabella.

Per le operazioni di accodamento alle tabelle esistenti (APPEND, INSERT, , UPDATEDELETE, MERGE), questo campo potrebbe mostrare una matrice [] vuota o colonne di partizione a seconda del metodo di scrittura usato (.save() vs .saveAsTable()).

Questa incoerenza è un comportamento previsto e non influisce sul modo in cui i dati vengono scritti nelle partizioni. Non è consigliabile usarlo per convalidare le operazioni di accodamento.

Example

Si consideri una tabella partizionata dalla date colonna . Quando si crea la tabella, partitionBy viene popolata:

df.write.format("delta") \
  .partitionBy("date") \
  .saveAsTable("sales_data")

L'operazione CREATE nella cronologia mostra:

operationParameters: {
  "mode": "ErrorIfExists",
  "partitionBy": "[\"date\"]"
}

Quando si aggiungono dati a questa tabella, partitionBy viene visualizzata una matrice vuota:

new_df.write.format("delta") \
  .mode("append") \
  .saveAsTable("sales_data")

L'operazione APPEND mostra:

operationParameters: {
  "mode": "Append",
  "partitionBy": "[]"
}

È previsto il valore vuoto partitionBy . I dati vengono comunque scritti nelle partizioni corrette in base allo schema di partizione esistente della tabella. Si noti che .save() a un percorso potrebbe mostrare colonne di partizione in questo campo, ma questa differenza è un dettaglio di implementazione e non influisce sul comportamento di scrittura.

Metriche operative

L'operazione history restituisce una raccolta di metriche operative nella mappa delle operationMetrics colonne.

Le tabelle seguenti elencano le definizioni delle chiavi della mappa in base all'operazione.

WRITE, CREATE TABLE AS SELECT, REPLACE TABLE AS SELECT, COPY INTO

Per queste operazioni sono disponibili le metriche seguenti:

STREAMING UPDATE

Per questa operazione sono disponibili le metriche seguenti:

DELETE

Per questa operazione sono disponibili le metriche seguenti:

TRUNCATE

Per questa operazione sono disponibili le metriche seguenti:

MERGE

Per questa operazione sono disponibili le metriche seguenti:

UPDATE

Per questa operazione sono disponibili le metriche seguenti:

FSCK

Per questa operazione sono disponibili le metriche seguenti:

CONVERT

Per questa operazione sono disponibili le metriche seguenti:

OPTIMIZE

Per questa operazione sono disponibili le metriche seguenti:

CLONE

Per questa operazione sono disponibili le metriche seguenti:

RESTORE

Per questa operazione sono disponibili le metriche seguenti:

VACUUM

Per questa operazione sono disponibili le metriche seguenti:

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.