Organisez votre lakehouse Fabric à l’aide de la conception d’architecture en médaillon

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.

   

5. Accédez aux paramètres de l’espace de travail et activez la fonctionnalité d’aperçu de la modification du modèle de données . Cela vous permettra de créer des relations entre les tables de votre Lakehouse à l’aide d’un ensemble de données Power BI.

   

   Remarque : Vous devrez peut-être actualiser l’onglet du navigateur après avoir activé la fonction d’aperçu.

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 nommé Sales .

   Après environ une minute, une nouvelle maison au bord du lac vide sera créée. 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 de cet exercice à partir de https://github.com/MicrosoftLearning/dp-data/blob/main/orders.zip. Extrayez les fichiers et enregistrez-les sous leurs noms d’origine sur votre ordinateur local (ou sur la machine virtuelle de laboratoire le cas échéant). Il doit y avoir 3 fichiers contenant des données de ventes sur 3 ans : 2019.csv, 2020.csv et 2021.csv.

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é bronze .

4. Dans le menu … du dossier bronze , sélectionnez Télécharger et Télécharger des fichiers , puis téléchargez les 3 fichiers (2019.csv, 2020.csv et 2021.csv) depuis votre ordinateur local (ou la machine virtuelle de laboratoire le cas échéant) vers Lakehouse. Utilisez la touche Maj pour télécharger les 3 fichiers en même temps.

5. Une fois les fichiers téléchargés, sélectionnez le dossier bronze ; et vérifiez que les fichiers ont été téléchargés, comme indiqué ici :

   

Maintenant que vous disposez de données dans la couche bronze de votre Lakehouse, vous pouvez utiliser un bloc-notes pour transformer les données et les charger dans une table delta dans la couche Argent.

1. Sur la page d’accueil , lors de l’affichage du contenu du dossier bronze dans votre lac de données, dans le menu Ouvrir le notebook , sélectionnez Nouveau 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. Lorsque le bloc-notes s’ouvre, renommez-le Transform data for Silver en sélectionnant le texte Notebook xxxx en haut à gauche du bloc-notes et en saisissant le nouveau nom.
3. Sélectionnez la cellule existante dans le bloc-notes, qui contient du code simple commenté. Mettez en surbrillance et supprimez ces deux lignes – vous n’aurez pas besoin de ce code.
   

   Remarque : les notebooks vous permettent d’exécuter du code dans divers langages, notamment Python, Scala et SQL. Dans cet exercice, vous utiliserez PySpark et SQL. Vous pouvez également ajouter des cellules de démarque pour fournir du texte et des images formatés pour documenter votre code.

4. Collez le code suivant dans la cellule
   

    

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

   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	Numéro de commande	Numéro de ligne de commande de vente	Date de commande	Nom du client	E-mail	Article	Quantité	Prix ​​unitaire	Impôt
   1	SO49172	1	2021-01-01	Brian Howard	brian23@adventure-works.com	Route-250 Rouge, 52	1	2443.35	195.468
   2	SO49173	1	2021-01-01	Linda Álvarez	linda19@adventure-works.com	Montagne-200 Argent, 38	1	2071.4197	165.7136
   …	…	…	…	…	…	…	…	…	…

   Le code que vous avez exécuté a chargé les données des fichiers CSV du dossier bronze dans un dataframe Spark, puis a affiché les premières lignes du dataframe.

   Remarque : Vous pouvez effacer, masquer et redimensionner automatiquement le contenu de la sortie de cellule en sélectionnant le menu … en haut à gauche du volet de sortie.

7. Vous allez maintenant ajouter des colonnes pour la validation et le nettoyage des données , en utilisant un dataframe PySpark pour ajouter des colonnes et mettre à jour les valeurs de certaines des colonnes existantes. Utilisez le bouton + pour ajouter un nouveau bloc de code et ajoutez le code suivant à la cellule :

    from pyspark.sql.functions import when, lit, col, current_timestamp, input_file_name
       
    # Add columns IsFlagged, CreatedTS and ModifiedTS
    df = df.withColumn("FileName", input_file_name()) \
        .withColumn("IsFlagged", when(col("OrderDate") < '2019-08-01',True).otherwise(False)) \
        .withColumn("CreatedTS", current_timestamp()).withColumn("ModifiedTS", current_timestamp())
       
    # Update CustomerName to "Unknown" if CustomerName null or empty
    df = df.withColumn("CustomerName", when((col("CustomerName").isNull() | (col("CustomerName")=="")),lit("Unknown")).otherwise(col("CustomerName")))
   

   La première ligne du code importe les fonctions nécessaires depuis PySpark. Vous ajoutez ensuite de nouvelles colonnes au cadre de données afin de pouvoir suivre le nom du fichier source, si la commande a été signalée comme étant antérieure à l’exercice qui vous intéresse et quand la ligne a été créée et modifiée.

   Enfin, vous mettez à jour la colonne CustomerName sur « Inconnu » si elle est nulle ou vide.

8. Exécutez la cellule pour exécuter le code à l’aide du bouton **▷ ( Exécuter la cellule )**.
9. Vous allez ensuite définir le schéma de la table sales_silver dans la base de données des ventes à l’aide du format Delta Lake. Créez un nouveau bloc de code et ajoutez le code suivant à la cellule :

    # Define the schema for the sales_silver table
       
    from pyspark.sql.types import *
    from delta.tables import *
       
    DeltaTable.createIfNotExists(spark) \
        .tableName("sales.sales_silver") \
        .addColumn("SalesOrderNumber", StringType()) \
        .addColumn("SalesOrderLineNumber", IntegerType()) \
        .addColumn("OrderDate", DateType()) \
        .addColumn("CustomerName", StringType()) \
        .addColumn("Email", StringType()) \
        .addColumn("Item", StringType()) \
        .addColumn("Quantity", IntegerType()) \
        .addColumn("UnitPrice", FloatType()) \
        .addColumn("Tax", FloatType()) \
        .addColumn("FileName", StringType()) \
        .addColumn("IsFlagged", BooleanType()) \
        .addColumn("CreatedTS", DateType()) \
        .addColumn("ModifiedTS", DateType()) \
        .execute()
   

10. Exécutez la cellule pour exécuter le code à l’aide du bouton **▷ ( Exécuter la cellule )**.
11. Sélectionnez le … dans la section Tables du volet de l’explorateur Lakehouse et sélectionnez Actualiser . Vous devriez maintenant voir la nouvelle table sales_silver répertoriée. Le ▲ (icône triangulaire) indique qu’il s’agit d’une table Delta.
    

    Remarque : Si vous ne voyez pas le nouveau tableau, attendez quelques secondes, puis sélectionnez à nouveau Actualiser ou actualisez l’intégralité de l’onglet du navigateur.

12. Vous allez maintenant effectuer une opération d’insertion sur une table Delta, en mettant à jour les enregistrements existants en fonction de conditions spécifiques et en insérant de nouveaux enregistrements lorsqu’aucune correspondance n’est trouvée. Ajoutez un nouveau bloc de code et collez le code suivant :

    # Update existing records and insert new ones based on a condition defined by the columns SalesOrderNumber, OrderDate, CustomerName, and Item.
    
    from delta.tables import *
        
    deltaTable = DeltaTable.forPath(spark, 'Tables/sales_silver')
        
    dfUpdates = df
        
    deltaTable.alias('silver') \
      .merge(
        dfUpdates.alias('updates'),
        'silver.SalesOrderNumber = updates.SalesOrderNumber and silver.OrderDate = updates.OrderDate and silver.CustomerName = updates.CustomerName and silver.Item = updates.Item'
      ) \
       .whenMatchedUpdate(set =
        {
              
        }
      ) \
     .whenNotMatchedInsert(values =
        {
          "SalesOrderNumber": "updates.SalesOrderNumber",
          "SalesOrderLineNumber": "updates.SalesOrderLineNumber",
          "OrderDate": "updates.OrderDate",
          "CustomerName": "updates.CustomerName",
          "Email": "updates.Email",
          "Item": "updates.Item",
          "Quantity": "updates.Quantity",
          "UnitPrice": "updates.UnitPrice",
          "Tax": "updates.Tax",
          "FileName": "updates.FileName",
          "IsFlagged": "updates.IsFlagged",
          "CreatedTS": "updates.CreatedTS",
          "ModifiedTS": "updates.ModifiedTS"
        }
      ) \
      .execute()
    

    Cette opération est importante car elle vous permet de mettre à jour les enregistrements existants dans la table en fonction des valeurs de colonnes spécifiques et d’insérer de nouveaux enregistrements lorsqu’aucune correspondance n’est trouvée. Il s’agit d’une exigence courante lorsque vous chargez des données à partir d’un système source pouvant contenir des mises à jour d’enregistrements existants et nouveaux.

Vous disposez désormais de données dans votre table Silver Delta qui sont prêtes à être transformées et modélisées davantage.

Maintenant que vous disposez de données dans votre couche Silver, vous pouvez utiliser le point de terminaison SQL pour explorer les données et effectuer une analyse de base. C’est une option intéressante pour vous si vous êtes familier avec SQL et que vous souhaitez effectuer une exploration de base de vos données. Dans cet exercice, nous utilisons la vue du point de terminaison SQL dans Fabric, mais notez que vous pouvez également utiliser d’autres outils tels que SQL Server Management Studio (SSMS) et Azure Data Explorer.

1. Revenez à votre espace de travail et notez que vous avez maintenant répertorié quelques actifs. Sélectionnez le point de terminaison SQL pour ouvrir votre Lakehouse dans la vue du point de terminaison SQL.
2. Sélectionnez Nouvelle requête SQL dans le ruban, ce qui ouvrira un éditeur de requête SQL. Notez que vous pouvez renommer votre requête à l’aide de l’ élément de menu … à côté du nom de la requête existante dans le volet de l’explorateur de Lakehouse.Nous allons exécuter deux requêtes SQL pour explorer nos données.
3. Collez la requête suivante dans l’éditeur de requêtes et sélectionnez Exécuter :

    SELECT YEAR(OrderDate) AS Year
        , CAST (SUM(Quantity * (UnitPrice + Tax)) AS DECIMAL(12, 2)) AS TotalSales
    FROM sales_silver
    GROUP BY YEAR(OrderDate) 
    ORDER BY YEAR(OrderDate)
   

   Cette requête calcule les ventes totales pour chaque année dans la table sales_silver. Vos résultats devraient ressembler à ceci :

   

4. Voyons maintenant quels clients achètent le plus (en termes de quantité). Collez la requête suivante dans l’éditeur de requêtes et sélectionnez Exécuter :

    SELECT TOP 10 CustomerName, SUM(Quantity) AS TotalQuantity
    FROM sales_silver
    GROUP BY CustomerName
    ORDER BY TotalQuantity DESC
   

   Cette requête calcule la quantité totale d’articles achetés par chaque client dans la table sales_silver, puis renvoie les 10 principaux clients en termes de quantité.

L’exploration des données au niveau de la couche d’argent est utile pour l’analyse de base, mais vous devrez transformer davantage les données et les modéliser dans un schéma en étoile pour permettre une analyse et un reporting plus avancés. Vous le ferez dans la section suivante.

Vous avez réussi à extraire les données de votre couche de bronze, à les transformer et à les charger dans une table Delta d’argent. Vous allez maintenant utiliser un nouveau notebook pour transformer davantage les données, les modéliser dans un schéma en étoile et les charger dans des tables Gold Delta.

Notez que vous auriez pu faire tout cela dans un seul bloc-notes, mais pour les besoins de cet exercice, vous utilisez des blocs-notes distincts pour démontrer le processus de transformation des données du bronze en argent, puis de l’argent en or. Cela peut faciliter le débogage, le dépannage et la réutilisation.

1. Revenez à la page d’accueil de Data Engineering et créez un nouveau bloc-notes appelé Transform data for Gold .
2. Dans le volet de l’explorateur Lakehouse, ajoutez votre Lakehouse Sales en sélectionnant Ajouter , puis en sélectionnant le Lakehouse Sales que vous avez créé précédemment. Vous devriez voir la table sales_silver répertoriée dans la section Tables du volet de l’explorateur.
3. Dans le bloc de code existant, supprimez le texte passe-partout et ajoutez le code suivant pour charger les données dans votre dataframe et commencer à créer votre schéma en étoile, puis exécutez-le :

    # Load data to the dataframe as a starting point to create the gold layer
    df = spark.read.table("Sales.sales_silver")
   

4. Ajoutez un nouveau bloc de code et collez le code suivant pour créer votre table de dimensions de date et l’exécuter :

    from pyspark.sql.types import *
    from delta.tables import*
       
    # Define the schema for the dimdate_gold table
    DeltaTable.createIfNotExists(spark) \
        .tableName("sales.dimdate_gold") \
        .addColumn("OrderDate", DateType()) \
        .addColumn("Day", IntegerType()) \
        .addColumn("Month", IntegerType()) \
        .addColumn("Year", IntegerType()) \
        .addColumn("mmmyyyy", StringType()) \
        .addColumn("yyyymm", StringType()) \
        .execute()
   

   Remarque : Vous pouvez exécuter la display(df)commande à tout moment pour vérifier la progression de votre travail. Dans ce cas, vous exécuteriez ‘display(dfdimDate_gold)’ pour voir le contenu de la trame de données dimDate_gold.

5. Dans un nouveau bloc de code, ajoutez et exécutez le code suivant pour créer un dataframe pour votre dimension de date, dimdate_gold :

    from pyspark.sql.functions import col, dayofmonth, month, year, date_format
       
    # Create dataframe for dimDate_gold
       
    dfdimDate_gold = df.dropDuplicates(["OrderDate"]).select(col("OrderDate"), \
            dayofmonth("OrderDate").alias("Day"), \
            month("OrderDate").alias("Month"), \
            year("OrderDate").alias("Year"), \
            date_format(col("OrderDate"), "MMM-yyyy").alias("mmmyyyy"), \
            date_format(col("OrderDate"), "yyyyMM").alias("yyyymm"), \
        ).orderBy("OrderDate")
   
    # Display the first 10 rows of the dataframe to preview your data
   
    display(dfdimDate_gold.head(10))
   

6. Vous divisez le code en nouveaux blocs de code afin de pouvoir comprendre et observer ce qui se passe dans le bloc-notes à mesure que vous transformez les données. Dans un autre nouveau bloc de code, ajoutez et exécutez le code suivant pour mettre à jour la dimension de date à mesure que de nouvelles données arrivent :

    from delta.tables import *
       
    deltaTable = DeltaTable.forPath(spark, 'Tables/dimdate_gold')
       
    dfUpdates = dfdimDate_gold
       
    deltaTable.alias('silver') \
      .merge(
        dfUpdates.alias('updates'),
        'silver.OrderDate = updates.OrderDate'
      ) \
       .whenMatchedUpdate(set =
        {
             
        }
      ) \
     .whenNotMatchedInsert(values =
        {
          "OrderDate": "updates.OrderDate",
          "Day": "updates.Day",
          "Month": "updates.Month",
          "Year": "updates.Year",
          "mmmyyyy": "updates.mmmyyyy",
          "yyyymm": "yyyymm"
        }
      ) \
      .execute()
   

   Bravo! Votre dimension de date est entièrement configurée. Vous allez maintenant créer votre dimension client.

7. Pour créer la table de dimension client, ajoutez un nouveau bloc de code , collez et exécutez le code suivant :

    from pyspark.sql.types import *
    from delta.tables import *
       
    # Create customer_gold dimension delta table
    DeltaTable.createIfNotExists(spark) \
        .tableName("sales.dimcustomer_gold") \
        .addColumn("CustomerName", StringType()) \
        .addColumn("Email",  StringType()) \
        .addColumn("First", StringType()) \
        .addColumn("Last", StringType()) \
        .addColumn("CustomerID", LongType()) \
        .execute()
   

8. Dans un nouveau bloc de code, ajoutez et exécutez le code suivant pour supprimer les clients en double, sélectionnez des colonnes spécifiques et divisez la colonne « CustomerName » pour créer des colonnes de nom « First » et « Last » :

    from pyspark.sql.functions import col, split
       
    # Create customer_silver dataframe
       
    dfdimCustomer_silver = df.dropDuplicates(["CustomerName","Email"]).select(col("CustomerName"),col("Email")) \
        .withColumn("First",split(col("CustomerName"), " ").getItem(0)) \
        .withColumn("Last",split(col("CustomerName"), " ").getItem(1)) 
       
    # Display the first 10 rows of the dataframe to preview your data
   
    display(dfdimCustomer_silver.head(10))
   

   Ici, vous avez créé un nouveau DataFrame dfdimCustomer_silver en effectuant diverses transformations telles que la suppression des doublons, la sélection de colonnes spécifiques et la division de la colonne « CustomerName » pour créer des colonnes de nom « First » et « Last ». Le résultat est un DataFrame avec des données client nettoyées et structurées, y compris des colonnes de nom « Prénom » et « Nom » distinctes extraites de la colonne « NomClient ».

9. Nous allons ensuite créer la colonne ID pour nos clients . Dans un nouveau bloc de code, collez et exécutez ce qui suit :

    from pyspark.sql.functions import monotonically_increasing_id, col, when, coalesce, max, lit
       
    dfdimCustomer_temp = spark.read.table("Sales.dimCustomer_gold")
       
    MAXCustomerID = dfdimCustomer_temp.select(coalesce(max(col("CustomerID")),lit(0)).alias("MAXCustomerID")).first()[0]
       
    dfdimCustomer_gold = dfdimCustomer_silver.join(dfdimCustomer_temp,(dfdimCustomer_silver.CustomerName == dfdimCustomer_temp.CustomerName) & (dfdimCustomer_silver.Email == dfdimCustomer_temp.Email), "left_anti")
       
    dfdimCustomer_gold = dfdimCustomer_gold.withColumn("CustomerID",monotonically_increasing_id() + MAXCustomerID + 1)
   
    # Display the first 10 rows of the dataframe to preview your data
   
    display(dfdimCustomer_gold.head(10))
   

   Ici, vous nettoyez et transformez les données client (dfdimCustomer_silver) en effectuant une anti-jointure gauche pour exclure les doublons qui existent déjà dans la table dimCustomer_gold, puis en générant des valeurs CustomerID uniques à l’aide de la fonction monotonically_increasing_id().

10. Vous allez désormais vous assurer que votre table client reste à jour à mesure que de nouvelles données arrivent. Dans un nouveau bloc de code , collez et exécutez ce qui suit :

    from delta.tables import *
    
    deltaTable = DeltaTable.forPath(spark, 'Tables/dimcustomer_gold')
        
    dfUpdates = dfdimCustomer_gold
        
    deltaTable.alias('silver') \
      .merge(
        dfUpdates.alias('updates'),
        'silver.CustomerName = updates.CustomerName AND silver.Email = updates.Email'
      ) \
       .whenMatchedUpdate(set =
        {
              
        }
      ) \
     .whenNotMatchedInsert(values =
        {
          "CustomerName": "updates.CustomerName",
          "Email": "updates.Email",
          "First": "updates.First",
          "Last": "updates.Last",
          "CustomerID": "updates.CustomerID"
        }
      ) \
      .execute()
    

11. Vous allez maintenant répéter ces étapes pour créer votre dimension de produit . Dans un nouveau bloc de code, collez et exécutez ce qui suit :

    from pyspark.sql.types import *
    from delta.tables import *
        
    DeltaTable.createIfNotExists(spark) \
        .tableName("sales.dimproduct_gold") \
        .addColumn("ItemName", StringType()) \
        .addColumn("ItemID", LongType()) \
        .addColumn("ItemInfo", StringType()) \
        .execute()
    

12. Ajoutez un autre bloc de code pour créer le dataframe product_silver .

    from pyspark.sql.functions import col, split, lit
        
    # Create product_silver dataframe
        
    dfdimProduct_silver = df.dropDuplicates(["Item"]).select(col("Item")) \
        .withColumn("ItemName",split(col("Item"), ", ").getItem(0)) \
        .withColumn("ItemInfo",when((split(col("Item"), ", ").getItem(1).isNull() | (split(col("Item"), ", ").getItem(1)=="")),lit("")).otherwise(split(col("Item"), ", ").getItem(1))) 
        
    # Display the first 10 rows of the dataframe to preview your data
    
    display(dfdimProduct_silver.head(10))
    

13. Vous allez maintenant créer des identifiants pour votre table dimProduct_gold . Ajoutez la syntaxe suivante à un nouveau bloc de code et exécutez-le :

    from pyspark.sql.functions import monotonically_increasing_id, col, lit, max, coalesce
        
    #dfdimProduct_temp = dfdimProduct_silver
    dfdimProduct_temp = spark.read.table("Sales.dimProduct_gold")
        
    MAXProductID = dfdimProduct_temp.select(coalesce(max(col("ItemID")),lit(0)).alias("MAXItemID")).first()[0]
        
    dfdimProduct_gold = dfdimProduct_silver.join(dfdimProduct_temp,(dfdimProduct_silver.ItemName == dfdimProduct_temp.ItemName) & (dfdimProduct_silver.ItemInfo == dfdimProduct_temp.ItemInfo), "left_anti")
        
    dfdimProduct_gold = dfdimProduct_gold.withColumn("ItemID",monotonically_increasing_id() + MAXProductID + 1)
        
    # Display the first 10 rows of the dataframe to preview your data
    
    display(dfdimProduct_gold.head(10))
    

    Cela calcule le prochain ID de produit disponible en fonction des données actuelles du tableau, attribue ces nouveaux ID aux produits, puis affiche les informations de produit mises à jour.

14. Semblable à ce que vous avez fait avec vos autres dimensions, vous devez vous assurer que votre table de produits reste à jour à mesure que de nouvelles données arrivent. Dans un nouveau bloc de code , collez et exécutez ce qui suit :

    from delta.tables import *
        
    deltaTable = DeltaTable.forPath(spark, 'Tables/dimproduct_gold')
                
    dfUpdates = dfdimProduct_gold
                
    deltaTable.alias('silver') \
      .merge(
            dfUpdates.alias('updates'),
            'silver.ItemName = updates.ItemName AND silver.ItemInfo = updates.ItemInfo'
            ) \
            .whenMatchedUpdate(set =
            {
                   
            }
            ) \
            .whenNotMatchedInsert(values =
             {
              "ItemName": "updates.ItemName",
              "ItemInfo": "updates.ItemInfo",
              "ItemID": "updates.ItemID"
              }
              ) \
              .execute()
    

    Maintenant que vos dimensions sont définies, la dernière étape consiste à créer la table de faits.

15. Dans un nouveau bloc de code , collez et exécutez le code suivant pour créer la table de faits :

    from pyspark.sql.types import *
    from delta.tables import *
        
    DeltaTable.createIfNotExists(spark) \
        .tableName("sales.factsales_gold") \
        .addColumn("CustomerID", LongType()) \
        .addColumn("ItemID", LongType()) \
        .addColumn("OrderDate", DateType()) \
        .addColumn("Quantity", IntegerType()) \
        .addColumn("UnitPrice", FloatType()) \
        .addColumn("Tax", FloatType()) \
        .execute()
    

16. Dans un nouveau bloc de code , collez et exécutez le code suivant pour créer un nouveau cadre de données permettant de combiner les données de vente avec les informations sur le client et le produit, notamment l’ID client, l’ID d’article, la date de commande, la quantité, le prix unitaire et les taxes :

    from pyspark.sql.functions import col
        
    dfdimCustomer_temp = spark.read.table("Sales.dimCustomer_gold")
    dfdimProduct_temp = spark.read.table("Sales.dimProduct_gold")
        
    df = df.withColumn("ItemName",split(col("Item"), ", ").getItem(0)) \
        .withColumn("ItemInfo",when((split(col("Item"), ", ").getItem(1).isNull() | (split(col("Item"), ", ").getItem(1)=="")),lit("")).otherwise(split(col("Item"), ", ").getItem(1))) \
        
        
    # Create Sales_gold dataframe
        
    dffactSales_gold = df.alias("df1").join(dfdimCustomer_temp.alias("df2"),(df.CustomerName == dfdimCustomer_temp.CustomerName) & (df.Email == dfdimCustomer_temp.Email), "left") \
            .join(dfdimProduct_temp.alias("df3"),(df.ItemName == dfdimProduct_temp.ItemName) & (df.ItemInfo == dfdimProduct_temp.ItemInfo), "left") \
        .select(col("df2.CustomerID") \
            , col("df3.ItemID") \
            , col("df1.OrderDate") \
            , col("df1.Quantity") \
            , col("df1.UnitPrice") \
            , col("df1.Tax") \
        ).orderBy(col("df1.OrderDate"), col("df2.CustomerID"), col("df3.ItemID"))
        
    # Display the first 10 rows of the dataframe to preview your data
        
    display(dffactSales_gold.head(10))
    

17. Vous allez désormais vous assurer que les données de ventes restent à jour en exécutant le code suivant dans un nouveau bloc de code :

    from delta.tables import *
        
    deltaTable = DeltaTable.forPath(spark, 'Tables/factsales_gold')
        
    dfUpdates = dffactSales_gold
        
    deltaTable.alias('silver') \
      .merge(
        dfUpdates.alias('updates'),
        'silver.OrderDate = updates.OrderDate AND silver.CustomerID = updates.CustomerID AND silver.ItemID = updates.ItemID'
      ) \
       .whenMatchedUpdate(set =
        {
              
        }
      ) \
     .whenNotMatchedInsert(values =
        {
          "CustomerID": "updates.CustomerID",
          "ItemID": "updates.ItemID",
          "OrderDate": "updates.OrderDate",
          "Quantity": "updates.Quantity",
          "UnitPrice": "updates.UnitPrice",
          "Tax": "updates.Tax"
        }
      ) \
      .execute()
    

    Ici, vous utilisez l’opération de fusion de Delta Lake pour synchroniser et mettre à jour la table factales_gold avec de nouvelles données de ventes (dffactSales_gold). L’opération compare la date de commande, l’ID client et l’ID d’article entre les données existantes (table d’argent) et les nouvelles données (met à jour DataFrame), mettant à jour les enregistrements correspondants et en insérant de nouveaux enregistrements si nécessaire.

Vous disposez désormais d’une couche d’or organisée et modélisée qui peut être utilisée pour le reporting et l’analyse.

Dans votre espace de travail, vous pouvez désormais utiliser la couche Gold pour créer un rapport et analyser les données. Vous pouvez accéder à l’ensemble de données directement dans votre espace de travail pour créer des relations et des mesures à des fins de reporting.

Notez que vous ne pouvez pas utiliser l’ ensemble de données par défaut créé automatiquement lorsque vous créez une maison au bord d’un lac. Vous devez créer un nouvel ensemble de données incluant les tables d’or que vous avez créées dans cet exercice, à partir de l’explorateur Lakehouse.

1. Dans votre espace de travail, accédez à votre Lakehouse Sales .
2. Sélectionnez Nouveau jeu de données Power BI dans le ruban de la vue de l’explorateur Lakehouse.
3. Sélectionnez vos tables d’or transformées à inclure dans votre ensemble de données et sélectionnez Confirmer .
   • dimdate_gold
   • dimcustomer_gold
   • dimproduct_gold
   • faitsales_gold

   Cela ouvrira l’ensemble de données dans Fabric où vous pourrez créer des relations et des mesures, comme indiqué ici :

   

4. Renommez votre ensemble de données afin qu’il soit plus facile à identifier. Sélectionnez le nom de l’ensemble de données dans le coin supérieur gauche de la fenêtre. Renommez l’ensemble de données en Sales_Gold .

À partir de là, vous ou d’autres membres de votre équipe de données pouvez créer des rapports et des tableaux de bord basés sur les données de votre Lakehouse. Ces rapports seront connectés directement à la couche dorée de votre Lakehouse, ils refléteront donc toujours les dernières données.

Dans cet exercice, vous avez appris à créer une architecture en médaillon dans un Lakehouse 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 .