Utiliser l’historique des tables

Pour les tables Apache Iceberg et Delta Lake, chaque opération qui modifie une table crée une nouvelle version de table. Utilisez les informations d’historique pour auditer les opérations, restaurer une table ou interroger une table à un point précis dans le temps en utilisant le voyage dans le temps.

Récupérer l’historique des tables

Exécutez la DESCRIBE HISTORY commande pour récupérer des informations, notamment les opérations, l’utilisateur et l’horodatage pour chaque écriture dans une table. Les opérations sont retournées dans l’ordre chronologique inverse.

Pour les colonnes qui DESCRIBE HISTORY retournent, les valeurs dans la operationParameters colonne, et les métriques par opération dans la operationMetrics colonne, voir Historique de la table, schéma et métriques d’opération.

La rétention de l’historique des tables est déterminée par le paramètre de table logRetentionDuration, qui est de 30 jours par défaut.

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

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

Pour plus d’informations sur la syntaxe Spark SQL, consultez DESCRIBE HISTORY.

Pour plus d’informations sur la syntaxe Scala, Java et Python, consultez la documentation de l’API Delta Lake.

L’Explorateur de catalogue affiche visuellement l’historique des tables sous l’onglet Historique .

Identifier le type d’opération OPTIMIZE

Le compactage automatique, le clustering liquide et l’ordre Z apparaissent tous dans l’historique des tables en tant qu’opérations OPTIMIZE . Pour déterminer lequel a été exécuté, inspectez la colonne operationParameters.

Pour classifier chaque OPTIMIZE opération dans l’historique d’une table, exécutez ce qui suit :

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;

Les sections suivantes décrivent chaque operationParameters valeur en détail. Pour les définitions des operationMetrics clés sélectionnées par la requête précédente, voir Métriques d’opération.

Compactage automatique

Le compactage automatique définit le auto paramètre sur true. Azure Databricks déclenche automatiquement le compactage automatique après une écriture. Lorsque auto est false, un utilisateur ou une tâche planifiée a exécuté la commande OPTIMIZE.

Par exemple, une opération de compactage automatique affiche les éléments suivants :

operationParameters: {
  "auto": "true"
}

Pour plus d’informations sur le compactage automatique, consultez Compactage automatique.

Regroupement de liquide

Le clustering liquid renseigne le paramètre clusterBy avec les noms des colonnes de clustering. Un tableau vide clusterBy ([]) indique uniquement le compactage de fichiers.

Par exemple, une opération qui a regroupé les données par les colonnes date et region affiche ce qui suit :

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

Pour plus d’informations sur le clustering liquide, consultez Utiliser le clustering liquide pour les tables.

Classement Z

Le tri Z remplit le paramètre zOrderBy avec les noms des colonnes du tri Z. Un tableau vide zOrderBy ([]) indique que l’opération n’a pas appliqué l’ordre Z.

Par exemple, une opération qui a appliqué l’ordre Z sur la date colonne affiche les éléments suivants :

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

Étendue opération

Le predicate paramètre indique si l’opération s’est exécutée sur la table complète ou uniquement une partie de celle-ci :

• Un tableau vide predicate ([]) signifie que l’opération s’est exécutée sur l’ensemble de la table.
• Un tableau rempli predicate signifie qu’une commande ciblée OPTIMIZE table_name WHERE <partition_predicate> s’exécute uniquement sur les partitions qui correspondent au prédicat.

Par exemple, une opération ciblée sur les partitions correspondantes year = 2024 affiche les éléments suivants :

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

Voyage dans le temps

Le voyage dans le temps permet d’interroger des versions précédentes d’une table en se basant sur l’horodatage ou sur la version de table (telle qu’enregistrée dans le journal des transactions). Vous pouvez utiliser le voyage dans le temps pour des applications telles que celles qui suivent :

• Recréation d’analyses, de rapports ou de sorties, telles que la sortie d’un modèle Machine Learning. Cela peut être utile pour le débogage ou l’audit, en particulier dans les secteurs réglementés.
• Écriture de requêtes temporelles complexes.
• Correction des erreurs dans vos données.
• Fournir un instantané d’isolation pour un ensemble de requêtes des tables à variation rapide.

Syntaxe de voyage dans le temps

Vous interrogez une table avec le voyage temporel en ajoutant une clause après la spécification du nom de la table.

• timestamp_expression peut être n’importe quel :
  • '2018-10-18T22:15:12.013Z', autrement dit, une chaîne qui peut être convertie en timestamp
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', autrement dit, une chaîne de date
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Toute autre expression qui est ou qui peut être convertie en un timestamp
• version est une valeur de type long qui peut être obtenue à partir de la sortie de DESCRIBE HISTORY table_spec.

Ni timestamp_expression ni version ne peuvent être des sous-requêtes.

Seules les chaînes de date ou timestamp sont acceptées. Par exemple : "2019-01-01" et "2019-01-01T00:00:00.000Z". Pour obtenir un exemple de syntaxe, consultez le code suivant :

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

Vous pouvez également utiliser la syntaxe @ pour spécifier le timestamp ou la version dans le cadre du nom de la table. Le timestamp doit être au format yyyyMMddHHmmssSSS. Vous pouvez spécifier une version avec @v. Pour obtenir un exemple de syntaxe, consultez le code suivant :

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

Configurer la conservation des données pour des requêtes de voyage dans le temps

Pour interroger une version de table précédente, vous devez conserver le journal et les fichiers de données pour cette version :

• Les fichiers de données sont supprimés lorsque VACUUM s’exécute sur une table.
• Les fichiers journaux sont supprimés automatiquement après les points de contrôle des versions de table.

Pour augmenter le seuil de rétention des données des tables, vous devez configurer les propriétés de la table suivantes, en remplaçant <format> par delta ou iceberg :

• <format>.logRetentionDuration = "interval <interval>" : contrôle la durée de conservation de l’historique d’une table. La valeur par défaut est interval 30 days.
  • Dans Databricks Runtime 18.0 et versions ultérieures, logRetentionDuration doit être supérieur ou égal à deletedFileRetentionDuration. Pour les tables managées du catalogue Unity, cela s’applique à Databricks Runtime 12.2 et versions ultérieures.
• <format>.deletedFileRetentionDuration = "interval <interval>" : détermine le seuil que VACUUM utilise pour supprimer des fichiers de données qui ne sont plus référencés dans la version actuelle de la table. La valeur par défaut est interval 7 days.

Par exemple, pour accéder à 30 jours de données historiques, définir delta.deletedFileRetentionDuration = "interval 30 days", qui correspond au paramètre par défaut pour delta.logRetentionDuration.

Vous pouvez spécifier des propriétés de table lors de la création de la table ou les définir avec une ALTER TABLE instruction. Consultez les informations de référence sur les propriétés de la table.

Exemples de voyages temporels

Pour corriger des suppressions accidentelles dans une table pour l’utilisateur 111 :

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

Pour corriger les mises à jour incorrectes accidentelles d’une table :

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 *

Pour interroger le nombre de nouveaux clients ajoutés au cours de la semaine dernière :

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

Points de contrôle du journal des transactions

Le journal des transactions enregistre les versions de table sous forme de fichiers JSON dans le répertoire du journal des transactions, ainsi que les données de table.

Pour optimiser l’interrogation de points de contrôle, les versions de table sont agrégées aux fichiers de point de contrôle Parquet, ce qui améliore les performances en empêchant la lecture de toutes les versions JSON de l’historique des tables. Les utilisateurs n’ont pas besoin d’interagir directement avec les points de contrôle.

Azure Databricks optimise la fréquence des points de contrôle pour la taille des données et la charge de travail. La fréquence du point de contrôle est susceptible d’être modifiée sans préavis.

Restaurer une table à un état antérieur

Utilisez la RESTORE commande pour restaurer une table vers une version ou un horodatage précédent, notamment pour les scénarios suivants :

• Vous pouvez restaurer une table déjà restaurée.
• Vous pouvez restaurer une table clonée.

Tenez compte des exigences suivantes :

• Pour restaurer une table, vous devez disposer droits MODIFY pour la table.
• Une fois les fichiers de données supprimés, manuellement ou par VACUUM, vous ne pouvez pas restaurer une table à une version antérieure qui fait référence à ces fichiers. Une restauration partielle vers cette version reste possible si spark.sql.files.ignoreMissingFiles a la valeur true.
• Pour restaurer par horodatage, utilisez les formats yyyy-MM-dd HH:mm:ss ou yyyy-MM-dd.

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

Pour plus d’informations sur la syntaxe, consultez RESTORE.

Comportement de diffusion en continu

La restauration est une opération de modification des données et peut entraîner des données en double pour les charges de travail en aval. Les entrées de journal ajoutées par la commande RESTORE contiennent dataChange défini sur true.

Pour les charges de travail en aval, telles qu’un travail de streaming structuré qui traite les mises à jour d’une table, les entrées du journal des modifications de données ajoutées par l’opération de restauration sont considérées comme de nouvelles mises à jour de données et le traitement des données peut entraîner des données en double.

Par exemple:

Dans l’exemple précédent, la RESTORE commande génère des mises à jour qui ont été vues précédemment lors de la lecture de la table version 0 et 1. Si une requête de diffusion en continu lit à nouveau cette table, ces fichiers sont considérés comme des données nouvellement ajoutées et sont à nouveau traités.

Restaurer des métriques

Une fois terminé, RESTORE signale les métriques suivantes sous la forme d’un DataFrame de ligne unique :

• table_size_after_restore : taille de la table après restauration.

• num_of_files_after_restore : nombre de fichiers dans la table après restauration.

• num_removed_files : nombre de fichiers supprimés (logiquement) de la table.

• num_restored_files : nombre de fichiers restaurés en raison d'un retour en arrière.

• removed_files_size : taille totale en octets des fichiers supprimés de la table.

• restored_files_size : taille totale en octets des fichiers restaurés.

  

Trouver la version du dernier commit

Pour obtenir le numéro de version de la dernière validation écrite par la SparkSession en cours sur l’ensemble des threads et des tables, interrogez la configuration SQL spark.databricks.<format>.lastCommitVersionInSession. Remplacez <format> par l’un delta ou l’autre, icebergselon le format de votre tableau.

Par exemple:

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Si aucune action de validation n’a été effectuée par SparkSession, l’interrogation de la clé retourne une valeur vide.