Meilleures pratiques : Delta Lake

Cet article décrit les meilleures pratiques lors de l’utilisation de Delta Lake.

Databricks recommande d’utiliser l’optimisation prédictive. Consultez Optimisation prédictive pour les tables managées Unity Catalog.

Lors de la suppression et de la recréation d’une table dans le même emplacement, vous devez toujours utiliser une instruction CREATE OR REPLACE TABLE. Voir Supprimer ou remplacer une table Delta.

Supprimer les configurations Delta héritées

Databricks recommande de supprimer la plupart des configurations Delta héritées explicites des configurations Spark et des propriétés de table lors de la mise à niveau vers une nouvelle version de Databricks Runtime. Les configurations héritées peuvent empêcher les nouvelles optimisations et les valeurs par défaut introduites par Databricks d’être appliquées aux charges de travail migrées.

Utiliser le clustering liquide pour ignorer des données optimisées

Databricks recommande d’utiliser le clustering liquide, plutôt que le partitionnement, l’ordre Z ou d’autres stratégies d’organisation de données, pour optimiser la disposition de l’omission des données. Consultez Utilisation des clustering liquides pour les tableaux Delta.

Compacter des fichiers

L’optimisation prédictive exécute automatiquement les commandes OPTIMIZE et VACUUM sur des tables managées par Unity Catalog. Consultez Optimisation prédictive pour les tables managées Unity Catalog.

Databricks recommande fréquemment d’exécuter la commande OPTIMISER pour compacter les petits fichiers.

Remplacer le contenu ou le schéma d’une table

Il se peut que vous souhaitiez remplacer une table Delta. Par exemple :

• Vous découvrez que les données de la table sont incorrectes et que vous souhaitez remplacer le contenu.
• Vous souhaitez réécrire la table entière pour effectuer des modifications de schéma incompatibles (telles que une modification des types de colonne).

Vous pouvez supprimer l’intégralité du répertoire d’une table Delta et créer une table sur le même chemin d’accès, ce qui n’est pas recommandé car :

• La suppression d’un répertoire n’est pas efficace. La suppression d’un répertoire contenant des fichiers très volumineux peut prendre des heures, voire des jours.
• Vous perdez la totalité du contenu des fichiers supprimés. Il sera difficile de faire une récupération si vous supprimez la mauvaise table.
• La suppression du répertoire n’est pas atomique. Lors de la suppression de la table, une requête simultanée qui lit la table peut échouer ou afficher une table partielle.

Si vous n’avez pas besoin de modifier le schéma de la table, vous pouvez supprimer des données d’une table Delta et insérer vos nouvelles données, ou mettre à jour la table pour corriger les valeurs incorrectes.

Si vous souhaitez modifier le schéma de la table, vous pouvez remplacer la table entière de manière atomique. Par exemple :

Python

dataframe.write \
  .mode("overwrite") \
  .option("overwriteSchema", "true") \
  .saveAsTable("<your-table>") # Managed table

dataframe.write \
  .mode("overwrite") \
  .option("overwriteSchema", "true") \
  .option("path", "<your-table-path>") \
  .saveAsTable("<your-table>") # External table

SQL

REPLACE TABLE <your-table> AS SELECT ... -- Managed table
REPLACE TABLE <your-table> LOCATION "<your-table-path>" AS SELECT ... -- External table

Scala

dataframe.write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable("<your-table>") // Managed table

dataframe.write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .option("path", "<your-table-path>")
  .saveAsTable("<your-table>") // External table

Cette approche présente plusieurs avantages :

• Le remplacement d’une table est beaucoup plus rapide, car elle n’a pas besoin de répertorier le répertoire de manière récursive ou de supprimer des fichiers.
• L’ancienne version de la table existe toujours. Si vous supprimez la table incorrecte, vous pouvez facilement récupérer les anciennes données à l’aide du voyage dans le temps. Consultez Utiliser l’historique des tables Delta Lake.
• Il s’agit d’une opération atomique. Les requêtes simultanées peuvent toujours lire la table pendant que vous supprimez la table.
• En raison des garanties de transaction ACID Delta Lake, si le remplacement de la table échoue, la table sera à son état précédent.

En outre, si vous voulez supprimer d’anciens fichiers pour économiser sur les coûts de stockage après avoir écrasé la table, vous pouvez utiliser VACUUM pour les supprimer. Cette fonction est optimisée pour la suppression de fichiers et est généralement plus rapide que supprimer le répertoire entier.

Mise en cache Spark

Databricks vous déconseille d’utiliser la mise en cache Spark pour les raisons suivantes :

• Vous perdez toutes les données ignorées qui peuvent provenir de filtres supplémentaires ajoutés au-dessus du cache DataFrame.
• Les données mises en cache peuvent ne pas être mises à jour, si la table est consultée avec un identificateur différent.

Différences entre Delta Lake et Parquet sur Apache Spark

Delta Lake gère automatiquement les opérations suivantes. Vous ne devez jamais effectuer ces opérations manuellement :

• REFRESH TABLE : les tables Delta retournent toujours les informations les plus récentes. Il n’est donc pas nécessaire d’appeler REFRESH TABLE manuellement après des modifications.
• Ajouter et supprimer des partitions : Delta Lake effectue automatiquement le suivi de l’ensemble des partitions présentes dans une table, et met à jour la liste à mesure que des données sont ajoutées ou supprimées. Par conséquent, il n’est pas nécessaire d’exécuter ALTER TABLE [ADD|DROP] PARTITION ou MSCK.
• Charger une partition unique : la lecture directe des partitions n’est pas nécessaire. Par exemple, vous n’avez pas besoin d’exécuter spark.read.format("parquet").load("/data/date=2017-01-01"). Utilisez plutôt une clause WHERE pour ignorer les données, par exemple spark.read.table("<table-name>").where("date = '2017-01-01'").
• Ne pas modifier manuellement des fichiers de données : Delta Lake utilise le journal des transactions pour valider de manière atomique les modifications apportées à la table. Ne modifiez, n’ajoutez et ne supprimez pas directement des fichiers de données Parquet dans une table Delta, car cela peut entraîner une perte de données ou l’endommagement de la table.

Améliorer les performances de la fusion Delta Lake

Vous pouvez réduire le temps pris par la fusion en adoptant les approches suivantes :

• Réduire l’espace de recherche des correspondances : par défaut, l’opération merge recherche dans la table Delta entière des correspondances dans la table source. Une manière d’accélérer l’opération merge consiste à réduire l’espace de recherche en ajoutant des contraintes connues dans la condition de correspondance. Par exemple, supposons que vous avez une table partitionnée par country et date, et que vous souhaitez utiliser merge pour mettre à jour les informations pour le dernier jour et un pays spécifique. L’ajout de la condition suivante accélère la requête, car elle recherche uniquement des correspondances dans les partitions appropriées :

  events.date = current_date() AND events.country = 'USA'
  

  Cela contribue également à réduire les risques de conflits avec d’autres opérations concurrentes. Pour plus d’informations, consultez Niveaux d’isolation et conflits d’écriture sur Azure Databricks.

• Compacter les fichiers : si les données sont stockées dans de nombreux fichiers de petite taille, la lecture des données pour rechercher des correspondances peut devenir lente. Vous pouvez compacter des fichiers de petite taille dans des fichiers plus volumineux pour améliorer le débit de lecture. Pour plus d’informations, consultez Optimiser la disposition des fichiers de données.

• Contrôler les partitions aléatoires pour les écritures : l’opération merge mélange plusieurs fois les données pour calculer et écrire les données mises à jour. Le nombre de tâches utilisées pour mélanger est contrôlé par la configuration de session Spark spark.sql.shuffle.partitions. La définition de ce paramètre permet non seulement de contrôler le parallélisme, mais aussi de déterminer le nombre de fichiers de sortie. L’augmentation de la valeur augmente le parallélisme, mais génère aussi un plus grand nombre de fichiers de données plus petits.

• Activer les écritures optimisées : pour les tables partitionnées, merge peut produire un beaucoup plus grand nombre de petits fichiers que le nombre de partitions générées de façon aléatoire. Cela est dû au fait que chaque tâche de mélange peut écrire plusieurs fichiers dans plusieurs partitions, et devenir un goulot d’étranglement des performances. Vous pouvez réduire le nombre de fichiers en activant les écritures optimisées. Consultez Écritures optimisées pour Delta Lake sur Azure Databricks.

• Ajuster la taille des fichiers dans le tableau : Azure Databricks peut détecter automatiquement si une table Delta comporte des opérations fréquentes merge de réécriture de fichiers et peut choisir de réduire la taille des fichiers réécrits en prévision de nouvelles réécritures de fichiers à l'avenir. Pour plus d’informations, consultez la section relative au réglage des tailles de fichier.

• Low Shuffle Merge : Low Shuffle Merge fournit une implémentation optimisée de MERGE qui offre de meilleures performances pour les charges de travail les plus courantes. En outre, elle conserve les optimisations de disposition des données existantes, telles que l’ordre de plan sur les données non modifiées.

Gérer les moyennes de données

Au début de chaque requête, les tables Delta se mettent à jour automatiquement vers la version la plus récente de la table. Ce processus peut être observé dans des blocs-notes lorsque l’état de la commande indique : Updating the Delta table's state. Toutefois, lors de l’exécution d’une analyse d’historique sur une table, vous n’avez pas forcément besoin de données jusqu’à la dernière minute, en particulier pour les tables où les données de streaming sont souvent ingérées. Dans ce cas, les requêtes peuvent être exécutées sur des instantanés obsolètes de votre table Delta. Cette approche peut réduire la latence dans l’obtention des résultats des requêtes.

Vous pouvez configurer la tolérance aux données obsolètes en définissant la configuration de session Spark spark.databricks.delta.stalenessLimit avec une valeur de chaîne de temps telle que 1h ou 15m (pendant 1 heure ou 15 minutes, respectivement). Cette configuration est spécifique à la session et n’affecte pas les autres clients qui accèdent à la table. Si l’état de la table a été mis à jour dans la limite d’obsolescence, une requête sur la table retourne les résultats sans attendre la mise à jour de la table la plus récente. Ce paramètre n’empêche jamais votre table de mettre à jour et, lorsque des données obsolètes sont retournées, la mise à jour se déroule en arrière-plan. Si la dernière mise à jour de table est antérieure à la limite d’obsolescence, la requête ne retourne pas de résultats tant que la mise à jour de l’état de la table n’est pas terminée.

Points de contrôle améliorés pour les requêtes à faible latence

Delta Lake écrit des points de contrôle comme état d’agrégation d’une table Delta à une fréquence optimisée. Ces points de contrôle servent de point de départ pour calculer l’état le plus récent de la table. Sans les points de contrôle, Delta Lake doit lire une grande collection de fichiers JSON (fichiers « delta ») représentant les validations dans le journal des transactions pour calculer l’état d’une table. En outre, les statistiques au niveau des colonnes, Delta Lake, utilisées pour exécuter les données ignorées sont stockées dans le point de contrôle.

Les statistiques au niveau des colonnes sont stockées sous forme de struct et de JSON (pour une compatibilité ascendante). Le format de struct rend Delta Lake Read beaucoup plus rapide, car :

• Delta Lake n’effectue pas d’analyse JSON coûteuse pour obtenir des statistiques au niveau des colonnes.
• Les fonctionnalités de nettoyage de colonne parquet réduisent considérablement les E/S nécessaires pour lire les statistiques d’une colonne.

Le format de struct active une collection d’optimisations qui réduisent la surcharge des opérations de lecture Delta Lake de quelques secondes à des dizaines de millisecondes, ce qui réduit considérablement la latence pour les requêtes courtes.

Gérer les statistiques au niveau des colonnes dans les points de contrôle

Vous gérez la façon dont les statistiques sont écrites dans les points de contrôle à l’aide des propriétés de table delta.checkpoint.writeStatsAsJson et delta.checkpoint.writeStatsAsStruct. Si les deux propriétés de table sont false, Delta Lake ne peut pas effectuer le saut de données.

• Batch écrit des statistiques d’écriture à la fois dans JSON et dans le format struct. delta.checkpoint.writeStatsAsJson a la valeur true.
• delta.checkpoint.writeStatsAsStruct n’est pas défini par défaut.
• Les lecteurs utilisent la colonne de struct lorsqu’ils sont disponibles et reviennent dans le cas de l’utilisation de la colonne JSON.

Activer les points de contrôle améliorés pour les requêtes Structured Streaming

Si vos charges Structured Streaming n’ont pas de conditions de latence faible (moins d’une minute), vous pouvez activer des points de contrôle améliorés en exécutant la commande SQL suivante :

ALTER TABLE <table-name> SET TBLPROPERTIES
('delta.checkpoint.writeStatsAsStruct' = 'true')

Vous pouvez également améliorer la latence d’écriture des points de contrôle en configurant les propriétés de table suivantes :

ALTER TABLE <table-name> SET TBLPROPERTIES
(
 'delta.checkpoint.writeStatsAsStruct' = 'true',
 'delta.checkpoint.writeStatsAsJson' = 'false'
)

Si les données ignorées ne sont pas utiles dans votre application, vous pouvez définir les deux propriétés sur « faux ». Ensuite, aucune statistique n’est collectée ou écrite. Databricks ne recommande pas cette configuration.