Utilitaires Microsoft Spark (MSSparkUtils) pour Fabric

Microsoft Spark Utilities (MSSparkUtils) est un package intégré qui vous permet d’effectuer aisément des tâches courantes. Vous pouvez utiliser MSSparkUtils pour travailler efficacement avec des systèmes de fichiers, pour récupérer des variables d’environnement, chaîner des notebooks et utiliser des secrets. Le package MSSparkUtils est disponible dans PySpark (Python) Scala, les notebooks SparkR et les pipelines Fabric.

Utilitaires du système de fichiers

mssparkutils.fs fournit des utilitaires pour travailler avec différents systèmes de fichiers, notamment Azure Data Lake Storage (ADLS) Gen2 et Stockage Blob Azure. Veillez à configurer l’accès à Azure Data Lake Storage Gen2 et Stockage Blob Azure de manière appropriée.

Exécutez les commandes suivantes pour obtenir une vue d’ensemble des méthodes disponibles :

from notebookutils import mssparkutils
mssparkutils.fs.help()

Sortie

mssparkutils.fs provides utilities for working with various FileSystems.

Below is overview about the available methods:

cp(from: String, to: String, recurse: Boolean = false): Boolean -> Copies a file or directory, possibly across FileSystems
mv(from: String, to: String, recurse: Boolean = false): Boolean -> Moves a file or directory, possibly across FileSystems
ls(dir: String): Array -> Lists the contents of a directory
mkdirs(dir: String): Boolean -> Creates the given directory if it does not exist, also creating any necessary parent directories
put(file: String, contents: String, overwrite: Boolean = false): Boolean -> Writes the given String out to a file, encoded in UTF-8
head(file: String, maxBytes: int = 1024 * 100): String -> Returns up to the first 'maxBytes' bytes of the given file as a String encoded in UTF-8
append(file: String, content: String, createFileIfNotExists: Boolean): Boolean -> Append the content to a file
rm(dir: String, recurse: Boolean = false): Boolean -> Removes a file or directory
exists(file: String): Boolean -> Check if a file or directory exists
mount(source: String, mountPoint: String, extraConfigs: Map[String, Any]): Boolean -> Mounts the given remote storage directory at the given mount point
unmount(mountPoint: String): Boolean -> Deletes a mount point
mounts(): Array[MountPointInfo] -> Show information about what is mounted
getMountPath(mountPoint: String, scope: String = ""): String -> Gets the local path of the mount point

Use mssparkutils.fs.help("methodName") for more info about a method.

MSSparkUtils fonctionne avec le système de fichiers de la même façon que les API Spark. Prenons l’utilisation de mssparkutils.fs.mkdirs() et Fabric Lakehouse, par exemple :

Énumérer des fichiers

Pour répertorier le contenu d’un répertoire, utilisez mssparkutils.fs.ls(’Le chemin d’accès à votre répertoire’). Par exemple :

mssparkutils.fs.ls("Files/tmp") # works with the default lakehouse files using relative path 
mssparkutils.fs.ls("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<path>")  # based on ABFS file system 
mssparkutils.fs.ls("file:/tmp")  # based on local file system of driver node 

Affichez les propriétés de fichier

Cette méthode retourne les propriétés du fichier, notamment le nom du fichier, son chemin d’accès, sa taille, s’il s’agit d’un répertoire et s’il s’agit d’un fichier.

files = mssparkutils.fs.ls('Your directory path')
for file in files:
    print(file.name, file.isDir, file.isFile, file.path, file.size)

Créer un répertoire

Cette méthode crée le répertoire donné s’il n’existe pas, ainsi que les répertoires parents nécessaires.

mssparkutils.fs.mkdirs('new directory name')  
mssparkutils.fs. mkdirs("Files/<new_dir>")  # works with the default lakehouse files using relative path 
mssparkutils.fs.ls("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<new_dir>")  # based on ABFS file system 
mssparkutils.fs.ls("file:/<new_dir>")  # based on local file system of driver node 

Copier un fichier

Cette méthode copie un fichier ou un répertoire et prend en charge l’activité de copie entre les systèmes de fichiers.

mssparkutils.fs.cp('source file or directory', 'destination file or directory', True)# Set the third parameter as True to copy all files and directories recursively

Fichier de copie performant

Cette méthode fournit un moyen plus rapide de copier ou de déplacer des fichiers, en particulier de gros volumes de données.

mssparkutils.fs.fastcp('source file or directory', 'destination file or directory', True)# Set the third parameter as True to copy all files and directories recursively

Afficher un aperçu du contenu du fichier

Cette méthode renvoie jusqu’aux premiers octets « maxBytes » du fichier donné sous la forme d’une chaîne encodée au format UTF-8.

mssparkutils.fs.head('file path', maxBytes to read)

Déplacer un fichier

Cette méthode déplace un fichier ou un répertoire et prend en charge les déplacements entre les systèmes de fichiers.

mssparkutils.fs.mv('source file or directory', 'destination directory', True) # Set the last parameter as True to firstly create the parent directory if it does not exist
mssparkutils.fs.mv('source file or directory', 'destination directory', True, True) # Set the third parameter to True to firstly create the parent directory if it does not exist. Set the last parameter to True to overwrite the updates.

Écrire dans un fichier

Cette méthode écrit la chaîne donnée dans un fichier encodé au format UTF-8.

mssparkutils.fs.put("file path", "content to write", True) # Set the last parameter as True to overwrite the file if it existed already

Ajouter du contenu à un fichier

Cette méthode ajoute la chaîne donnée à un fichier, encodé au format UTF-8.

mssparkutils.fs.append("file path", "content to append", True) # Set the last parameter as True to create the file if it does not exist

Supprimer un fichier ou un répertoire

Cette méthode déplace un fichier ou un répertoire.

mssparkutils.fs.rm('file path', True) # Set the last parameter as True to remove all files and directories recursively

Monter/démonter le répertoire

Pour plus d’informations sur l’utilisation détaillée dans Montage de fichiers et démonter.

Utilitaires de notebook

Utilisez les utilitaires de notebook MSSparkUtils pour exécuter un notebook ou quitter un notebook avec une valeur. Exécutez la commande suivante pour obtenir une vue d’ensemble des méthodes disponibles :

mssparkutils.notebook.help()

Sortie :

exit(value: String): void -> This method lets you exit a notebook with a value.
run(path: String, timeoutSeconds: int, arguments: Map): String -> This method runs a notebook and returns its exit value.

Référencer un notebook

Cette méthode référence un notebook et renvoie sa valeur de sortie. Vous pouvez exécuter des appels de fonction d’imbrication dans un notebook de manière interactive ou dans un pipeline. Le notebook référencé s’exécute sur le pool Spark du notebook qui appelle cette fonction.

mssparkutils.notebook.run("notebook name", <timeoutSeconds>, <parameterMap>, <workspaceId>)

Par exemple :

mssparkutils.notebook.run("Sample1", 90, {"input": 20 })

Le notebook Fabric prend également en charge le référencement de blocs-notes sur plusieurs espaces de travail en spécifiant l’ID de l’espace de travail.

mssparkutils.notebook.run("Sample1", 90, {"input": 20 }, "fe0a6e2a-a909-4aa3-a698-0a651de790aa")

Vous pouvez ouvrir le lien d’instantané de l’exécution de référence dans la sortie de cellule. L’instantané capture les résultats de l’exécution du code et vous permet de déboguer facilement une exécution de référence.

Exécution de référence de plusieurs notebooks en parallèle

La méthode mssparkutils.notebook.runMultiple() vous permet d’exécuter plusieurs notebooks en parallèle ou avec une structure topologique prédéfinie. L’API utilise un mécanisme d’implémentation multithread au sein d’une session Spark, ce qui signifie que les ressources de calcul sont partagées par les exécutions du notebook de référence.

Avec mssparkutils.notebook.runMultiple(), vous pouvez :

• Exécutez plusieurs notebooks simultanément, sans attendre que chacun d’eux se termine.

• Spécifiez les dépendances et l’ordre d’exécution de vos notebooks à l’aide d’un format JSON simple.

• Optimisez l’utilisation des ressources de calcul Spark et réduisez le coût de vos projets Fabric.

• Affichez les instantanés de chaque enregistrement d’exécution de notebook dans la sortie, et déboguez/surveillez vos tâches de notebook de manière pratique.

• Obtenez la valeur de sortie de chaque activité exécutive et utilisez-les dans les tâches en aval.

Vous pouvez également essayer d’exécuter mssparkutils.notebook.help("runMultiple") pour rechercher l’exemple et l’utilisation détaillée.

Voici un exemple simple d’exécution d’une liste de notebooks en parallèle à l’aide de cette méthode :

mssparkutils.notebook.runMultiple(["NotebookSimple", "NotebookSimple2"])

Le résultat d’exécution du notebook racine est le suivant :

Voici un exemple d’exécution de notebooks avec une structure topologique à l’aide de mssparkutils.notebook.runMultiple(). Utilisez cette méthode pour orchestrer facilement les notebooks via une expérience de code.

# run multiple notebooks with parameters
DAG = {
    "activities": [
        {
            "name": "NotebookSimple", # activity name, must be unique
            "path": "NotebookSimple", # notebook path
            "timeoutPerCellInSeconds": 90, # max timeout for each cell, default to 90 seconds
            "args": {"p1": "changed value", "p2": 100}, # notebook parameters
        },
        {
            "name": "NotebookSimple2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 2", "p2": 200}
        },
        {
            "name": "NotebookSimple2.2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 3", "p2": 300},
            "retry": 1,
            "retryIntervalInSeconds": 10,
            "dependencies": ["NotebookSimple"] # list of activity names that this activity depends on
        }
    ],
    "timeoutInSeconds": 43200, # max timeout for the entire DAG, default to 12 hours
    "concurrency": 50 # max number of notebooks to run concurrently, default to 50
}
mssparkutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

Le résultat d’exécution du notebook racine est le suivant :

Quitter un notebook

Cette méthode quitte un notebook avec une valeur. Vous pouvez exécuter des appels de fonction d’imbrication dans un notebook de manière interactive ou dans un pipeline.

• Lorsque vous appelez une fonction exit() à partir d’un notebook de manière interactive, le notebook Fabric lève une exception, ignore l’exécution des cellules de la sous-séquence et maintient la session Spark active.

• Lorsque vous orchestrez un notebook qui appelle une fonction exit(), l’activité de notebook renvoie une valeur de sortie, termine l’exécution du pipeline et arrête la session Spark.

• Lorsque vous appelez une fonction exit() dans un notebook référencé, Fabric Spark arrête l’exécution supplémentaire dans le notebook référencé et continue à exécuter les cellules suivantes du notebook principal qui appelle la fonction run(). Par exemple : Notebook1 possède trois cellules et appelle une fonction exit() dans la deuxième cellule. Notebook2 possède cinq cellules et appelle run(notebook1) dans la troisième cellule. Lorsque vous exécutez Notebook2, Notebook1 s’arrête à la deuxième cellule lorsque vous atteignez la fonction exit(). Notebook2 continue à exécuter ses quatrième et cinquième cellules.

mssparkutils.notebook.exit("value string")

Par exemple :

Le notebook Sample1 avec les deux cellules suivantes :

• La cellule 1 définit un paramètre d’entrée dont la valeur par défaut est définie sur 10.

• La cellule 2 quitte le notebook avec l'entrée comme valeur de sortie.

Vous pouvez exécuter Sample1 dans un autre bloc-notes avec les valeurs par défaut :

exitVal = mssparkutils.notebook.run("Sample1")
print (exitVal)

Output:

Notebook executed successfully with exit value 10

Vous pouvez exécuter Sample1 dans un autre notebook et définir la valeur d’entrée sur 20 :

exitVal = mssparkutils.notebook.run("Sample1", 90, {"input": 20 })
print (exitVal)

Output:

Notebook executed successfully with exit value 20

Utilitaires d’informations d’identification

Vous pouvez utiliser les utilitaires d’informations d’identification MSSparkUtils pour récupérer les jetons d’accès et gérer les secrets dans Azure Key Vault.

Exécutez la commande suivante pour obtenir une vue d’ensemble des méthodes disponibles :

mssparkutils.credentials.help()

Output:

getToken(audience, name): returns AAD token for a given audience, name (optional)
getSecret(keyvault_endpoint, secret_name): returns secret for a given Key Vault and secret name

Obtenir un jeton

getToken renvoie un jeton Microsoft Entra pour une audience et un nom donnés (facultatif). La liste suivante présente les touches d’audience actuellement disponibles :

• Ressource d’audience de stockage : « storage »
• Ressource Power BI : « pbi »
• Ressource Azure Key Vault : « keyvault »
• Ressource de base de données KQL Synapse RTA : « kusto »

Exécutez la commande suivante pour obtenir le jeton :

mssparkutils.credentials.getToken('audience Key')

Obtenir le secret à l’aide des informations d’identification de l’utilisateur

getSecret renvoie un secret Azure Key Vault pour un point de terminaison Azure Key Vault donné et un nom de secret à l’aide des informations d’identification d’utilisateur.

mssparkutils.credentials.getSecret('https://<name>.vault.azure.net/', 'secret name')

Montage et démontage de fichiers

Fabric prend en charge les scénarios de montage suivants dans le package Utilitaires Microsoft Spark. Vous pouvez utiliser les API mount, unmount, getMountPath() et mounts() pour attacher un stockage distant (ADLS Gen2) à tous les nœuds de travail (nœud pilote et nœuds Worker). Une fois le point de montage de stockage en place, utilisez l’API de fichier local pour accéder aux données comme si elles étaient stockées dans le système de fichiers local.

Guide pratique pour monter un compte ADLS Gen2

L’exemple suivant illustre comment monter Azure Data Lake Storage Gen2. Le montage de Stockage Blob fonctionne de la même façon.

Cet exemple suppose que vous disposez d’un compte Data Lake Storage Gen2 nommé storegen2et que le compte possède un conteneur nommé mycontainer que vous souhaitez monter sur /test dans votre session Spark de notebook.

Pour monter le conteneur appelé mycontainer, mssparkutils doit d’abord vérifier si vous disposez de l’autorisation d’accéder au conteneur. Actuellement, Fabric prend en charge deux méthodes d’authentification pour l’opération de montage du déclencheur : accountKey et sastoken.

Montage via un jeton de signature d’accès partagé ou une clé de compte

MSSparkUtils prend en charge la transmission explicite d’une clé de compte ou d’un jeton de signature d’accès partagé (SAP) en tant que paramètre pour monter la cible.

Pour des raisons de sécurité, nous vous recommandons de stocker des clés de compte ou des jetons SAP dans Azure Key Vault (comme l’illustre la capture d’écran suivant). Vous pouvez ensuite les récupérer à l’aide de l’API mssparkutils.credentials.getSecret. Pour plus d’informations sur Azure Key Vault, consultez À propos des clés de compte de stockage managées Azure Key Vault.

Exemple de code pour la méthode accountKey :

from notebookutils import mssparkutils  
# get access token for keyvault resource
# you can also use full audience here like https://vault.azure.net
accountKey = mssparkutils.credentials.getSecret("<vaultURI>", "<secretName>")
mssparkutils.fs.mount(  
    "abfss://mycontainer@<accountname>.dfs.core.windows.net",  
    "/test",  
    {"accountKey":accountKey}
)

Exemple de code pour sastoken :

from notebookutils import mssparkutils  
# get access token for keyvault resource
# you can also use full audience here like https://vault.azure.net
sasToken = mssparkutils.credentials.getSecret("<vaultURI>", "<secretName>")
mssparkutils.fs.mount(  
    "abfss://mycontainer@<accountname>.dfs.core.windows.net",  
    "/test",  
    {"sasToken":sasToken}
)

from notebookutils import mssparkutils

Paramètres de montage :

• fileCacheTimeout : les objets blob sont mis en cache dans le dossier temporaire local pendant 120 secondes par défaut. Pendant ce temps, blobfuse ne vérifie pas si le fichier est à jour ou non. Le paramètre peut être défini pour modifier le délai d’expiration par défaut. Lorsque plusieurs clients modifient des fichiers en même temps, afin d’éviter les incohérences entre les fichiers locaux et distants, nous vous recommandons de raccourcir le temps de mise en cache, ou même de le remplacer par 0, et de toujours obtenir les derniers fichiers à partir du serveur.
• timeout : le délai d’expiration de l’opération de montage est de 120 secondes par défaut. Le paramètre peut être défini pour modifier le délai d’expiration par défaut. Lorsqu’il y a trop d’exécuteurs ou lorsque le montage expire, nous vous recommandons d’augmenter la valeur.

Vous pouvez utiliser ces paramètres comme suit :

mssparkutils.fs.mount(
   "abfss://mycontainer@<accountname>.dfs.core.windows.net",
   "/test",
   {"fileCacheTimeout": 120, "timeout": 120}
)

Guide pratique pour monter un lakehouse

Exemple de code de montage pour un lakehouse sur /test :

from notebookutils import mssparkutils 
mssparkutils.fs.mount( 
 "abfss://<workspace_id>@onelake.dfs.fabric.microsoft.com/<lakehouse_id>", 
 "/test"
)

Accéder aux fichiers sous le point de montage à l’aide de l’API mssparktuils fs

L’objectif principal de l’opération de montage est de permettre aux clients d’accéder aux données stockées dans un compte de stockage distant avec une API de système de fichiers local. Vous pouvez également accéder aux données à l’aide de l’API mssparkutils fs avec un chemin d’accès monté en tant que paramètre. Le format de chemin d’accès utilisé ici est un peu différent.

Supposons que vous avez monté le conteneur Data Lake Storage Gen2 mycontainer sur /test à l’aide de l’API mount. Lorsque vous accédez aux données avec l’API de système de fichiers local, le format du chemin d’accès est le suivant :

/synfs/notebook/{sessionId}/test/{filename}

Lorsque vous souhaitez accéder aux données à l’aide de l’API mssparkutils fs, nous vous recommandons d’utiliser un getMountPath() pour obtenir le chemin d’accès précis :

path = mssparkutils.fs.getMountPath("/test")

• Lister des répertoires :

  mssparkutils.fs.ls(f"file://{mssparkutils.fs.getMountPath('/test')}")
  

• Lisez le contenu du fichier :

  mssparkutils.fs.head(f"file://{mssparkutils.fs.getMountPath('/test')}/myFile.txt")
  

• Créer un répertoire :

  mssparkutils.fs.mkdirs(f"file://{mssparkutils.fs.getMountPath('/test')}/newdir")
  

Accéder aux fichiers sous le point de montage via le chemin d’accès local

Vous pouvez facilement lire et écrire les fichiers au point de montage à l’aide du système de fichiers standard. Voici un exemple Python :

#File read
with open(mssparkutils.fs.getMountPath('/test2') + "/myFile.txt", "r") as f:
    print(f.read())
#File write
with open(mssparkutils.fs.getMountPath('/test2') + "/myFile.txt", "w") as f:
    print(f.write("dummy data"))

Guide pratique pour vérifier les points de montage existants

Vous pouvez utiliser l’API mssparkutils.fs.mounts() pour vérifier toutes les informations de point de montage existantes :

mssparkutils.fs.mounts()

Comment démonter le point de montage

Utilisez le code suivant pour démonter votre point de montage (/test dans cet exemple) :

mssparkutils.fs.unmount("/test")

Limitations connues

• Le montage actuel étant une configuration au niveau du travail, nous vous recommandons d’utiliser l’API mounts pour vérifier si un point de montage existe ou n’est pas disponible.

• Le mécanisme de démontage n’est pas automatique. Une fois l’exécution de l’application terminée, pour démonter le point de montage et libérer l’espace disque, vous devez appeler explicitement une API de démontage dans votre code. Sinon, le point de montage existera toujours dans le nœud une fois l’exécution de l’application terminée.

• Le montage d’un compte de stockage ADLS Gen1 n’est pas pris en charge.

Utilitaires Lakehouse

mssparkutils.lakehouse fournit des utilitaires spécifiquement adaptés à la gestion des artefacts Lakehouse. Ces utilitaires permettent aux utilisateurs de créer, récupérer, mettre à jour et supprimer facilement des artefacts Lakehouse.

Vue d’ensemble des méthodes

Voici une vue d’ensemble des méthodes disponibles fournies par mssparkutils.lakehouse :

# Create a new Lakehouse artifact
create(name: String, description: String = "", workspaceId: String = ""): Artifact

# Retrieve a Lakehouse artifact
get(name: String, workspaceId: String = ""): Artifact

# Update an existing Lakehouse artifact
update(name: String, newName: String, description: String = "", workspaceId: String = ""): Artifact

# Delete a Lakehouse artifact
delete(name: String, workspaceId: String = ""): Boolean

# List all Lakehouse artifacts
list(workspaceId: String = ""): Array[Artifact]

Exemples d'utilisation

Pour utiliser efficacement ces méthodes, tenez compte des exemples d’utilisation suivants :

Création d’un artefact Lakehouse

artifact = mssparkutils.lakehouse.create("artifact_name", "Description of the artifact", "optional_workspace_id")

Récupération d’un artefact Lakehouse

artifact = mssparkutils.lakehouse.get("artifact_name", "optional_workspace_id")

Mise à jour d’un artefact Lakehouse

updated_artifact = mssparkutils.lakehouse.update("old_name", "new_name", "Updated description", "optional_workspace_id")

Suppression d’un artefact Lakehouse

is_deleted = mssparkutils.lakehouse.delete("artifact_name", "optional_workspace_id")

Liste des artefacts Lakehouse

artifacts_list = mssparkutils.lakehouse.list("optional_workspace_id")

Informations supplémentaires

Pour plus d’informations sur chaque méthode et ses paramètres, utilisez la fonction mssparkutils.lakehouse.help("methodName").

Avec les utilitaires Lakehouse de MSSparkUtils, la gestion de vos artefacts Lakehouse devient plus efficace et intégrée à vos pipelines Fabric, ce qui améliore votre expérience globale de gestion des données.

N’hésitez pas à explorer ces utilitaires et à les incorporer dans vos workflows Fabric pour une gestion transparente des artefacts Lakehouse.

Utilitaires de runtime

Afficher les informations de contexte de la session

Avec mssparkutils.runtime.context, vous pouvez obtenir les informations contextuelles sur la session actuelle, notamment le nom du notebook, le lakehouse par défaut, les informations de l’espace de travail, s’il s’agit d’une exécution de pipeline, etc.

mssparkutils.runtime.context

Problème connu

Lors de l’utilisation d’une version de runtime supérieure à la version 1.2 et exécutant mssparkutils.help(), les API fabricClient, warehouse et workspace répertoriées ne sont pas prises en charge pour l’instant, mais le seront disponibles plus tard.

Contenu connexe

• Gestion des bibliothèques