Utiliser des tables Delta dans Apache Spark

Avant de travailler avec des données dans Fabric, créez un espace de travail avec la version d’essai Fabric activée.

1. Sur la page d’accueil de Microsoft Fabric , sélectionnez Synapse Data Engineering .
2. Dans la barre de menu de gauche, sélectionnez Espaces de travail (l’icône ressemble à 🗇).
3. Créez un nouvel espace de travail avec le nom de votre choix, en sélectionnant un mode de licence qui inclut la capacité Fabric ( Trial , Premium ou Fabric ).

4. Lorsque votre nouvel espace de travail s’ouvre, il devrait être vide.

   

Maintenant que vous disposez d’un espace de travail, il est temps de créer un lac de données pour les données que vous allez analyser.

1. Dans la page d’accueil de Synapse Data Engineering , créez un nouveau Lakehouse avec le nom de votre choix.

   Après environ une minute, une nouvelle maison au bord du lac vide. Vous devez ingérer certaines données dans le data Lakehouse pour analyse. Il existe plusieurs façons de procéder, mais dans cet exercice, vous allez simplement télécharger un fichier texte sur votre ordinateur local (ou sur la machine virtuelle de votre laboratoire le cas échéant), puis le télécharger sur votre Lakehouse.

2. Téléchargez le fichier de données pour cet exercice à partir de https://github.com/MicrosoftLearning/dp-data/raw/main/products.csv, en l’enregistrant sous le nom products.csv sur votre ordinateur local (ou sur la machine virtuelle de l’atelier, le cas échéant).

3. Revenez à l’onglet du navigateur Web contenant votre Lakehouse et dans le menu … du dossier Fichiers du volet Explorateur , sélectionnez Nouveau sous-dossier et créez un dossier nommé products .

4. Dans le menu … du dossier products , sélectionnez Upload et Upload files , puis téléchargez le fichier products.csv depuis votre ordinateur local (ou la machine virtuelle de laboratoire le cas échéant) vers Lakehouse.

5. Une fois le fichier téléchargé, sélectionnez le dossier des produits ; et vérifiez que le fichier products.csv a été téléchargé, comme indiqué ici :

   

1. Sur la page d’accueil , lors de l’affichage du contenu du dossier products dans votre datalake, dans le menu Ouvrir le notebook , sélectionnez New notebook .

   Après quelques secondes, un nouveau bloc-notes contenant une seule cellule s’ouvrira. Les blocs-notes sont constitués d’une ou plusieurs cellules pouvant contenir du code ou du markdown (texte formaté).

2. Sélectionnez la cellule existante dans le bloc-notes, qui contient un code simple, puis utilisez son icône 🗑 ( Supprimer ) en haut à droite pour la supprimer – vous n’aurez pas besoin de ce code.

3. Dans le volet de l’explorateur Lakehouse sur la gauche, développez Fichiers et sélectionnez des produits pour afficher un nouveau volet affichant le fichier products.csv que vous avez téléchargé précédemment :

   

4. Dans le menu … de products.csv , sélectionnez Charger les données > Spark . Une nouvelle cellule de code contenant le code suivant doit être ajoutée au notebook :

   df = spark.read.format("csv").option("header","true").load("Files/products/products.csv")
   # df now is a Spark DataFrame containing CSV data from "Files/products/products.csv".
   display(df)
   

   Astuce : Vous pouvez masquer le volet contenant les fichiers à gauche en utilisant son icône « . Cela vous aidera à vous concentrer sur le bloc-notes.

5. Utilisez le bouton ▷ ( Exécuter la cellule ) à gauche de la cellule pour l’exécuter.

   Remarque : Comme c’est la première fois que vous exécutez du code Spark dans ce notebook, une session Spark doit être démarrée. Cela signifie que la première exécution peut prendre environ une minute. Les courses suivantes seront plus rapides.

6. Une fois la commande de cellule terminée, examinez le résultat sous la cellule, qui devrait ressembler à ceci :

   Indice	ID produit	Nom du produit	Catégorie	Liste des prix
   1	771	Montagne-100 Argent, 38	Vélo de montagne	3399.9900
   2	772	Montagne-100 Argent, 42	Vélo de montagne	3399.9900
   3	773	Montagne-100 Argent, 44	Vélo de montagne	3399.9900
   …	…	…	…	…

Vous pouvez enregistrer le dataframe en tant que table delta en utilisant la méthode saveAsTable. Delta Lake prend en charge la création de tables gérées et externes .

Créer une table gérée

Les tables gérées sont des tables pour lesquelles les métadonnées du schéma et les fichiers de données sont gérés par Fabric. Les fichiers de données de la table sont créés dans le dossier Tables .

1. Sous les résultats renvoyés par la première cellule de code, utilisez le bouton + Code pour ajouter une nouvelle cellule de code si elle n’existe pas déjà. Entrez ensuite le code suivant dans la nouvelle cellule et exécutez-le :

   df.write.format("delta").saveAsTable("managed_products")
   

2. Dans le volet de l’explorateur Lakehouse , dans le menu … du dossier Tables , sélectionnez Actualiser . Développez ensuite le nœud Tables et vérifiez que la table Managed_products a été créée.

Créer une table externe

Vous pouvez également créer des tables externes pour lesquelles les métadonnées du schéma sont définies dans le métastore du Lakehouse, mais les fichiers de données sont stockés dans un emplacement externe.

1. Ajoutez une autre nouvelle cellule de code et ajoutez-y le code suivant :

   df.write.format("delta").saveAsTable("external_products", path="abfs_path/external_products")
   

2. Dans le volet de l’explorateur Lakehouse , dans le menu … du dossier Fichiers , sélectionnez Copier le chemin ABFS .Le chemin ABFS est le chemin complet vers le dossier Fichiers dans le stockage OneLake de votre Lakehouse – similaire à ceci :abfss://workspace@tenant-onelake.dfs.fabric.microsoft.com/lakehousename.Lakehouse/Files
3. Dans le code que vous avez entré dans la cellule de code, remplacez abfs_path par le chemin que vous avez copié dans le presse-papiers afin que le code enregistre le dataframe en tant que table externe avec des fichiers de données dans un dossier nommé external_products à l’emplacement de votre dossier Fichiers . Le chemin complet devrait ressembler à ceci :abfss://workspace@tenant-onelake.dfs.fabric.microsoft.com/lakehousename.Lakehouse/Files/external_products
4. Dans le volet de l’explorateur Lakehouse , dans le menu … du dossier Tables , sélectionnez Actualiser . Développez ensuite le nœud Tables et vérifiez que la table external_products a été créée.
5. Dans le volet de l’explorateur Lakehouse , dans le menu … du dossier Fichiers , sélectionnez Actualiser . Développez ensuite le nœud Fichiers et vérifiez que le dossier external_products a été créé pour les fichiers de données de la table.

Comparez les tables gérées et externes

Explorons les différences entre les tables gérées et externes.

1. Ajoutez une autre cellule de code et exécutez le code suivant :

   %%sql
   
   DESCRIBE FORMATTED managed_products;
   

   Dans les résultats, affichez la propriété Location de la table, qui doit être un chemin d’accès au stockage OneLake pour le Lakehouse se terminant par /Tables/managed_products (vous devrez peut-être élargir la colonne Type de données pour voir le chemin complet).

2. Modifiez la DESCRIBEcommande pour afficher les détails de la table external_products comme indiqué ici :

   %%sql
   
   DESCRIBE FORMATTED external_products;
   

   Dans les résultats, affichez la propriété Location de la table, qui doit être un chemin d’accès au stockage OneLake pour le Lakehouse se terminant par /Files/external_products (vous devrez peut-être élargir la colonne Type de données pour voir le chemin complet).

   Les fichiers de la table gérée sont stockés dans le dossier Tables du stockage OneLake du Lakehouse. Dans ce cas, un dossier nommé Managed_products a été créé pour stocker les fichiers Parquet et le dossier delta_log pour la table que vous avez créée.

3. Ajoutez une autre cellule de code et exécutez le code suivant :

   %%sql
   
   DROP TABLE managed_products;
   DROP TABLE external_products;
   

4. Dans le volet de l’explorateur Lakehouse , dans le menu … du dossier Tables , sélectionnez Actualiser . Développez ensuite le nœud Tables et vérifiez qu’aucune table n’est répertoriée.
5. Dans le volet de l’explorateur Lakehouse , développez le dossier Fichiers et vérifiez que les produits_externes n’ont pas été supprimés. Sélectionnez ce dossier pour afficher les fichiers de données Parquet et le dossier _delta_log pour les données qui se trouvaient auparavant dans la table external_products . Les métadonnées de la table externe ont été supprimées, mais les fichiers n’ont pas été affectés.

Utiliser SQL pour créer une table

1. Ajoutez une autre cellule de code et exécutez le code suivant :

   %%sql
   
   CREATE TABLE products
   USING DELTA
   LOCATION 'Files/external_products';
   

2. Dans le volet de l’explorateur Lakehouse , dans le menu … du dossier Tables , sélectionnez Actualiser . Développez ensuite le nœud Tables et vérifiez qu’une nouvelle table nommée products est répertoriée. Développez ensuite la table pour vérifier que son schéma correspond à la trame de données d’origine enregistrée dans le dossier external_products .
3. Ajoutez une autre cellule de code et exécutez le code suivant :

   %%sql
   
   SELECT * FROM products;
   

L’historique des transactions pour les tables delta est stocké dans des fichiers JSON dans le dossier delta_log . Vous pouvez utiliser ce journal des transactions pour gérer la gestion des versions des données.

1. Ajoutez une nouvelle cellule de code au notebook et exécutez le code suivant :

   %%sql
   
   UPDATE products
   SET ListPrice = ListPrice * 0.9
   WHERE Category = 'Mountain Bikes';
   

   Ce code met en place une réduction de 10% sur le prix des VTT.

2. Ajoutez une autre cellule de code et exécutez le code suivant :

   %%sql
   
   DESCRIBE HISTORY products;
   

   Les résultats montrent l’historique des transactions enregistrées pour la table.

3. Ajoutez une autre cellule de code et exécutez le code suivant :

   delta_table_path = 'Files/external_products'
   
   # Get the current data
   current_data = spark.read.format("delta").load(delta_table_path)
   display(current_data)
   
   # Get the version 0 data
   original_data = spark.read.format("delta").option("versionAsOf", 0).load(delta_table_path)
   display(original_data)
   

   Les résultats affichent deux dataframe: l’une contenant les données après la réduction de prix et l’autre montrant la version originale des données.

Delta Lake prend en charge le streaming de données. Les tables Delta peuvent être un récepteur ou une source pour les flux de données créés à l’aide de l’API Spark Structured Streaming. Dans cet exemple, vous utiliserez une table delta comme récepteur pour certaines données de streaming dans un scénario d’Internet des objets (IoT) simulé.

1. Ajoutez une nouvelle cellule de code dans le bloc-notes. Ensuite, dans la nouvelle cellule, ajoutez le code suivant et exécutez-le :
   
   
2. Assurez-vous que le message Flux source créé… est imprimé. Le code que vous venez d’exécuter a créé une source de données en streaming basée sur un dossier dans lequel certaines données ont été enregistrées, représentant les lectures d’appareils IoT hypothétiques.
3. Dans une nouvelle cellule de code, ajoutez et exécutez le code suivant :

   # Write the stream to a delta table
   delta_stream_table_path = 'Tables/iotdevicedata'
   checkpointpath = 'Files/delta/checkpoint'
   deltastream = iotstream.writeStream.format("delta").option("checkpointLocation", checkpointpath).start(delta_stream_table_path)
   print("Streaming to delta sink...")
   

   Ce code écrit les données du périphérique de streaming au format delta dans un dossier nommé iotdevicedata . En raison du chemin d’accès à l’emplacement du dossier dans le dossier Tables , une table sera automatiquement créée pour celui-ci.

4. Dans une nouvelle cellule de code, ajoutez et exécutez le code suivant :

   %%sql
   
   SELECT * FROM IotDeviceData;
   

   Ce code interroge la table IotDeviceData , qui contient les données de l’appareil provenant de la source de streaming.

5. Dans une nouvelle cellule de code, ajoutez et exécutez le code suivant :

   # Add more data to the source stream
   more_data = '''{"device":"Dev1","status":"ok"}
   {"device":"Dev1","status":"ok"}
   {"device":"Dev1","status":"ok"}
   {"device":"Dev1","status":"ok"}
   {"device":"Dev1","status":"error"}
   {"device":"Dev2","status":"error"}
   {"device":"Dev1","status":"ok"}'''
   
   mssparkutils.fs.put(inputPath + "more-data.txt", more_data, True)
   

   Ce code écrit davantage de données hypothétiques sur l’appareil dans la source de streaming.

6. Réexécutez la cellule contenant le code suivant :

   %%sql
   
   SELECT * FROM IotDeviceData;
   

   Ce code interroge à nouveau la table IotDeviceData , qui doit désormais inclure les données supplémentaires ajoutées à la source de streaming.

7. Dans une nouvelle cellule de code, ajoutez et exécutez le code suivant :

   deltastream.stop()
   

   Ce code arrête le flux.

Dans cet exercice, vous avez appris à utiliser des tables delta dans Microsoft Fabric.

Si vous avez fini d’explorer votre Lakehouse, vous pouvez supprimer l’espace de travail que vous avez créé pour cet exercice.

1. Dans la barre de gauche, sélectionnez l’icône de votre espace de travail pour afficher tous les éléments qu’il contient.
2. Dans le menu … de la barre d’outils, sélectionnez Paramètres de l’espace de travail .
3. Dans la section Autre , sélectionnez Supprimer cet espace de travail .