🥇Delta Lake Dive : application et évolution du système

Salut Habr ! Je présente à votre attention la traduction de l'article "Plonger dans Delta Lake : application et évolution des schémas" auteurs Burak Yavuz, Brenner Heintz et Denny Lee, préparé en prévision du début du cours Ingénieur de données de l'OTUS.

Les données, comme notre expérience, s’accumulent et évoluent constamment. Pour suivre le rythme, nos modèles mentaux du monde doivent s’adapter à de nouvelles données, dont certaines contiennent de nouvelles dimensions – de nouvelles façons d’observer des choses dont nous n’avions aucune idée auparavant. Ces modèles mentaux ne sont pas très différents des schémas de tableaux qui déterminent la manière dont nous catégorisons et traitons les nouvelles informations.

Cela nous amène à la question de la gestion des schémas. À mesure que les défis et les exigences de votre entreprise évoluent au fil du temps, la structure de vos données évolue également. Delta Lake facilite l'introduction de nouvelles mesures à mesure que les données changent. Les utilisateurs ont accès à une sémantique simple pour gérer leurs schémas de tables. Ces outils incluent Schema Enforcement, qui protège les utilisateurs contre la pollution involontaire de leurs tables avec des erreurs ou des données inutiles, et Schema Evolution, qui permet d'ajouter automatiquement de nouvelles colonnes de données précieuses aux emplacements appropriés. Dans cet article, nous approfondirons l’utilisation de ces outils.

Comprendre les schémas de table

Chaque DataFrame dans Apache Spark contient un schéma qui définit la forme des données, comme les types de données, les colonnes et les métadonnées. Avec Delta Lake, le schéma de table est stocké au format JSON dans le journal des transactions.

Qu’est-ce que l’application du régime ?

Schema Enforcement, également connu sous le nom de Schema Validation, est un mécanisme de sécurité dans Delta Lake qui garantit la qualité des données en rejetant les enregistrements qui ne correspondent pas au schéma de la table. Telle l'hôtesse de la réception d'un restaurant populaire sur réservation, elle vérifie si chaque colonne de données saisie dans le tableau se trouve dans la liste correspondante des colonnes attendues (en d'autres termes, s'il existe une « réservation » pour chacune d'entre elles). ), et rejette tous les enregistrements comportant des colonnes qui ne figurent pas dans la liste.

Comment fonctionne l’application du schéma ?

Delta Lake utilise la vérification du schéma à l'écriture, ce qui signifie que la compatibilité de toutes les nouvelles écritures dans la table est vérifiée avec le schéma de la table cible au moment de l'écriture. Si le schéma est incohérent, Delta Lake abandonne complètement la transaction (aucune donnée n'est écrite) et déclenche une exception pour informer l'utilisateur de l'incohérence.
Delta Lake utilise les règles suivantes pour déterminer si un enregistrement est compatible avec une table. DataFrame inscriptible :

ne peut pas contenir de colonnes supplémentaires qui ne figurent pas dans le schéma de la table cible. A l'inverse, tout va bien si les données entrantes ne contiennent pas absolument toutes les colonnes de la table - ces colonnes se verront simplement attribuer des valeurs nulles.
ne peut pas avoir des types de données de colonne différents des types de données des colonnes de la table cible. Si la colonne de la table cible contient des données StringType, mais que la colonne correspondante dans le DataFrame contient des données IntegerType, l'application du schéma lèvera une exception et empêchera l'opération d'écriture d'avoir lieu.
ne peut pas contenir de noms de colonnes qui diffèrent uniquement par la casse. Cela signifie que vous ne pouvez pas définir de colonnes nommées « Foo » et « foo » dans la même table. Alors que Spark peut être utilisé en mode sensible à la casse ou insensible à la casse (par défaut), Delta Lake préserve la casse mais n'est pas sensible à la casse dans le stockage du schéma. Parquet est sensible à la casse lors du stockage et du renvoi des informations sur les colonnes. Pour éviter d'éventuelles erreurs, corruption ou perte de données (quelque chose que nous avons personnellement vécu chez Databricks), nous avons décidé d'ajouter cette limitation.

Pour illustrer cela, examinons ce qui se passe dans le code ci-dessous lorsque nous essayons d'ajouter des colonnes nouvellement générées à une table Delta Lake qui n'est pas encore configurée pour les accepter.

# Сгенерируем DataFrame ссуд, который мы добавим в нашу таблицу Delta Lake
loans = sql("""
            SELECT addr_state, CAST(rand(10)*count as bigint) AS count,
            CAST(rand(10) * 10000 * count AS double) AS amount
            FROM loan_by_state_delta
            """)

# Вывести исходную схему DataFrame
original_loans.printSchema()

root
  |-- addr_state: string (nullable = true)
  |-- count: integer (nullable = true)
 
# Вывести новую схему DataFrame
loans.printSchema()
 
root
  |-- addr_state: string (nullable = true)
  |-- count: integer (nullable = true)
  |-- amount: double (nullable = true) # new column
 
# Попытка добавить новый DataFrame (с новым столбцом) в существующую таблицу
loans.write.format("delta") 
           .mode("append") 
           .save(DELTALAKE_PATH)

Returns:

A schema mismatch detected when writing to the Delta table.
 
To enable schema migration, please set:
'.option("mergeSchema", "true")'
 
Table schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
 
Data schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
-- amount: double (nullable = true)
 
If Table ACLs are enabled, these options will be ignored. Please use the ALTER TABLE command for changing the schema.

Au lieu d'ajouter automatiquement de nouvelles colonnes, Delta Lake impose un schéma et arrête l'écriture. Pour aider à déterminer quelle colonne (ou ensemble de colonnes) est à l'origine de l'écart, Spark génère les deux schémas à partir de la trace de pile à des fins de comparaison.

Quel est l’avantage d’appliquer un schéma ?

L’application du schéma étant un contrôle assez strict, il s’agit d’un excellent outil à utiliser comme contrôleur d’accès à un ensemble de données propres et entièrement transformés, prêts à être produits ou consommés. Généralement appliqué aux tables qui alimentent directement les données :

Algorithmes d'apprentissage automatique
Tableaux de bord BI
Outils d'analyse et de visualisation de données
Tout système de production nécessitant des schémas sémantiques hautement structurés et fortement typés.

Pour préparer leurs données à ce dernier obstacle, de nombreux utilisateurs utilisent une architecture simple « multi-sauts » qui introduit progressivement une structure dans leurs tables. Pour en savoir plus, vous pouvez consulter l'article Apprentissage automatique de qualité production avec Delta Lake.

Bien sûr, l'application du schéma peut être utilisée n'importe où dans votre pipeline, mais n'oubliez pas que le streaming vers une table dans ce cas peut être frustrant car, par exemple, vous avez oublié que vous avez ajouté une autre colonne aux données entrantes.

Prévenir la dilution des données

À présent, vous vous demandez peut-être : de quoi s’agit-il ? Après tout, il arrive parfois qu'une erreur inattendue de « incompatibilité de schéma » puisse vous perturber dans votre flux de travail, surtout si vous êtes nouveau sur Delta Lake. Pourquoi ne pas simplement laisser le schéma changer si nécessaire afin que je puisse écrire mon DataFrame quoi qu'il arrive ?

Comme le dit le vieil adage, « mieux vaut prévenir que guérir ». À un moment donné, si vous ne prenez pas soin d'appliquer votre schéma, des problèmes de compatibilité des types de données apparaîtront : des sources de données brutes apparemment homogènes peuvent contenir des cas extrêmes, des colonnes corrompues, des mappages mal formés ou d'autres choses effrayantes auxquelles rêver. cauchemars. La meilleure approche consiste à arrêter ces ennemis à la porte - avec l'application du schéma - et à les traiter en pleine lumière, plutôt que plus tard lorsqu'ils commencent à se cacher dans les profondeurs sombres de votre code de production.

L'application d'un schéma vous donne l'assurance que le schéma de votre table ne changera pas à moins que vous n'approuviez la modification. Cela évite la dilution des données, qui peut se produire lorsque de nouvelles colonnes sont ajoutées si fréquemment que des tableaux compressés auparavant précieux perdent leur sens et leur utilité en raison de l'inondation de données. En vous encourageant à être intentionnel, à fixer des normes élevées et à exiger une qualité élevée, l'application des schémas fait exactement ce pour quoi elle a été conçue : vous aider à rester consciencieux et vos feuilles de calcul propres.

Si, après mûre réflexion, vous décidez que vous êtes vraiment nécessaire ajoutez une nouvelle colonne - pas de problème, vous trouverez ci-dessous un correctif sur une ligne. La solution est l'évolution du circuit !

Qu’est-ce que l’évolution du schéma ?

L'évolution du schéma est une fonctionnalité qui permet aux utilisateurs de modifier facilement le schéma de table actuel en fonction des données qui changent au fil du temps. Il est le plus souvent utilisé lors de l'exécution d'une opération d'ajout ou de réécriture pour adapter automatiquement le schéma afin d'inclure une ou plusieurs nouvelles colonnes.

Comment fonctionne l’évolution du schéma ?

En suivant l'exemple de la section précédente, les développeurs peuvent facilement utiliser l'évolution du schéma pour ajouter de nouvelles colonnes qui avaient été précédemment rejetées en raison d'une incohérence de schéma. L'évolution du circuit est activée en ajoutant .option('mergeSchema', 'true') à votre équipe Spark .write или .writeStream.

# Добавьте параметр mergeSchema
loans.write.format("delta") 
           .option("mergeSchema", "true") 
           .mode("append") 
           .save(DELTALAKE_SILVER_PATH)

Pour afficher le graphique, exécutez la requête Spark SQL suivante

# Создайте график с новым столбцом, чтобы подтвердить, что запись прошла успешно
%sql
SELECT addr_state, sum(`amount`) AS amount
FROM loan_by_state_delta
GROUP BY addr_state
ORDER BY sum(`amount`)
DESC LIMIT 10

Vous pouvez également définir cette option pour toute la session Spark en ajoutant spark.databricks.delta.schema.autoMerge = True à la configuration Spark. Mais utilisez-le avec prudence, car l’application du schéma ne vous alertera plus en cas d’incohérences involontaires de schéma.

En incluant le paramètre dans la requête mergeSchema, toutes les colonnes présentes dans le DataFrame mais pas dans la table cible sont automatiquement ajoutées à la fin du schéma dans le cadre d'une transaction d'écriture. Des champs imbriqués peuvent également être ajoutés et ceux-ci seront également ajoutés à la fin des colonnes de structure correspondantes.

Les ingénieurs de données et les scientifiques des données peuvent utiliser cette option pour ajouter de nouvelles colonnes (peut-être une mesure récemment suivie ou la colonne des performances des ventes de ce mois-ci) à leurs tables de production d'apprentissage automatique existantes sans casser les modèles existants basés sur d'anciennes colonnes.

Les types de modifications de schéma suivants sont autorisés dans le cadre de l'évolution du schéma lors d'un ajout ou d'une réécriture de table :

Ajout de nouvelles colonnes (c'est le scénario le plus courant)
Modification des types de données depuis NullType -> tout autre type ou promotion depuis ByteType -> ShortType -> IntegerType

D'autres modifications non autorisées dans l'évolution du schéma nécessitent que le schéma et les données soient réécrits en ajoutant .option("overwriteSchema", "true"). Par exemple, dans le cas où la colonne "Foo" était à l'origine un entier et que le nouveau schéma était un type de données chaîne, alors tous les fichiers Parquet(data) devraient être réécrits. Ces changements incluent :

supprimer une colonne
changer le type de données d'une colonne existante (sur place)
renommer les colonnes qui diffèrent uniquement par la casse (par exemple, "Foo" et "foo")

Enfin, avec la prochaine version de Spark 3.0, le DDL explicite sera entièrement pris en charge (en utilisant ALTER TABLE), permettant aux utilisateurs d'effectuer les actions suivantes sur les schémas de table :

ajout de colonnes
changer les commentaires de la colonne
définition des propriétés de la table qui contrôlent le comportement de la table, telles que la définition de la durée de stockage d'un journal des transactions.

Quel est l’avantage de l’évolution du circuit ?

L'évolution du schéma peut être utilisée chaque fois que vous avoir l'intention modifiez le schéma de votre table (contrairement au cas où vous avez accidentellement ajouté des colonnes à votre DataFrame qui ne devraient pas être là). Il s'agit du moyen le plus simple de migrer votre schéma, car il ajoute automatiquement les noms de colonnes et les types de données corrects sans avoir à les déclarer explicitement.

Conclusion

L'application du schéma rejette toute nouvelle colonne ou toute autre modification de schéma incompatible avec votre table. En définissant et en maintenant ces normes élevées, les analystes et les ingénieurs peuvent être sûrs que leurs données présentent le plus haut niveau d'intégrité, les communiquant clairement et clairement, leur permettant ainsi de prendre de meilleures décisions commerciales.

D’un autre côté, l’évolution des schémas complète l’application en simplifiant supposé changements de schéma automatiques. Après tout, il ne devrait pas être difficile d’ajouter une colonne.

L’application forcée du schéma est yang, là où l’évolution du schéma est yin. Lorsqu'elles sont utilisées ensemble, ces fonctionnalités rendent la suppression du bruit et le réglage du signal plus faciles que jamais.

Nous tenons également à remercier Mukul Murthy et Pranav Anand pour leurs contributions à cet article.

Autres articles de cette série :

Plongez dans Delta Lake : déballage du journal des transactions