Utilidades de Microsoft Spark (MSSparkUtils) para Fabric

Microsoft Spark Utilities (MSSparkUtils) es un paquete integrado que te ayuda a realizar tareas comunes con facilidad. Use MSSparkUtils para trabajar con sistemas de archivos, obtener variables de entorno, encadenar cuadernos y trabajar con secretos. El paquete MSSparkUtils está disponible en PySpark (Python), Scala, cuadernos SparkR y pipelines Fabric.

Utilidades del sistema de archivos

mssparkutils.fs ofrece utilidades para trabajar con diversos sistemas de archivos, incluyendo Azure Data Lake Storage Gen2 y Azure Blob Storage. Asegúrese de configurar el acceso a Azure Data Lake Storage Gen2 y Azure Blob Storage adecuadamente.

Ejecute los siguientes comandos para obtener información general sobre los métodos disponibles:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Salida

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 funciona con el sistema de archivos de la misma manera que las API de Spark. Tomemos como ejemplo mssparkuitls.fs.mkdirs() y el uso de la casa del lago:

Enumerar archivos

Para enumerar el contenido de un directorio, use mssparkutils.fs.ls("Ruta de acceso del directorio"). Por ejemplo:

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 

Vea las propiedades del archivo.

Este método devuelve las propiedades del archivo, incluyendo el nombre del archivo, la ruta del archivo, el tamaño del archivo y si es un directorio o un archivo.

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

Creación de un directorio

Este método crea el directorio especificado si no existe, y crea los directorios padres necesarios.

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 

Copiar archivo

Este método copia un archivo o directorio y admite la actividad de copia entre sistemas de archivos.

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

Copiar archivo con rendimiento óptimo

Este método proporciona una manera más rápida de copiar o mover archivos, especialmente grandes volúmenes de datos.

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

Vista previa del contenido del archivo

Este método devuelve hasta los primeros maxBytes bytes del archivo especificado como una cadena codificada en UTF-8.

# Set the second parameter as an integer for the maxBytes to read
mssparkutils.fs.head('file path', <maxBytes>)

Mover archivo

Este método mueve un archivo o directorio y admite movimientos entre sistemas de archivos.

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.

Escribir archivo

Este método escribe la cadena especificada en un archivo, codificada en formato 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

Adición de contenido a un archivo

Este método anexa la cadena especificada a un archivo, codificada en formato 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

Eliminación de un archivo o directorio

Este método quita un archivo o directorio.

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

Montaje o desmontaje del directorio

Para obtener más información sobre el uso en detalle, consulte Montar y desmontar archivos.

Utilidades de portátil

Use las utilidades de cuaderno MSSparkUtils para ejecutar un cuaderno o salir de uno con un valor. Ejecute el siguiente comando para obtener información general sobre los métodos disponibles:

mssparkutils.notebook.help()

Salida:

exit(value: String): Raises NotebookExit Exception -> 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.

Referencia a un cuaderno

Este método hace referencia a un cuaderno y devuelve su valor de salida. Puede ejecutar llamadas de función anidadas en una libreta de manera interactiva o en una canalización. El cuaderno al que se hace referencia se ejecuta en el grupo de Spark del cuaderno que invoca esta función.

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

Por ejemplo:

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

El cuaderno de Fabric también admite la referencia a cuadernos en varias áreas de trabajo especificando el ID del área de trabajo.

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

Puede abrir el enlace de la instantánea de la ejecución de referencia en la celda de salida. La instantánea captura los resultados de la ejecución del código y permite depurar fácilmente una ejecución de referencia.

Referencia para ejecutar múltiples notebooks en paralelo

El método mssparkutils.notebook.runMultiple() permite ejecutar varios cuadernos en paralelo o con una estructura topológica predefinida. La API utiliza una implementación multihilo para enviar, poner en cola y supervisar cuadernos secundarios que se ejecutan en instancias de REPL aisladas (bucle de lectura, evaluación e impresión) dentro de la sesión de Spark existente. Los cuadernos infantiles referenciados comparten los recursos de cómputo de la sesión.

Con mssparkutils.notebook.runMultiple(), puede:

• Ejecute varios cuadernos simultáneamente, sin esperar a que finalice cada uno.

• Especifique las dependencias y el orden de ejecución de los cuadernos mediante un formato JSON simple.

• Optimice el uso de recursos de proceso de Spark y reduzca el costo de los proyectos de Fabric.

• Visualiza las instantáneas de cada registro de ejecución de cada notebook en la salida, y depura y supervisa las tareas de tu notebook de forma cómoda.

• Obtenga el valor de salida de cada actividad ejecutiva y úselos en tareas descendentes.

También puede intentar ejecutar mssparkutils.notebook.help("runMultiple") para buscar el ejemplo y el uso detallado.

Este es un ejemplo sencillo de ejecutar una lista de cuadernos en paralelo mediante este método:

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

El resultado de la ejecución del cuaderno raíz es el siguiente:

El siguiente ejemplo muestra cuadernos en ejecución con una estructura topológica usando mssparkutils.notebook.runMultiple(). Use este método para organizar fácilmente cuadernos a través de una experiencia de código.

# 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, defaults to 50 but ultimately constrained by the number of driver cores
}
mssparkutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

El resultado de la ejecución del cuaderno raíz es el siguiente:

Salir de un cuaderno

Este método sale de un cuaderno con un valor. Puede ejecutar llamadas de función anidadas en una libreta de manera interactiva o en una canalización.

• Cuando llama a una función exit() desde un cuaderno de forma interactiva, el cuaderno de Fabric lanza una excepción, omite la ejecución de las celdas siguientes y mantiene activa la sesión de Spark.

• Cuando se orquesta un cuaderno en una canalización que invoca una función exit(), la actividad del cuaderno devuelve un valor de salida, completa la ejecución de la canalización y detiene la sesión de Spark. No encierres la función exit() alrededor de un try/catch, ya que esta excepción NotebookExit debe propagarse para que la tubería obtenga el valor de retorno.

• Cuando llamas a una función exit() en un cuaderno que se está referenciando, Fabric Spark detiene la ejecución adicional del cuaderno referenciado y continúa ejecutando las siguientes celdas en el cuaderno principal que llama a la función run(). Por ejemplo: Notebook1 tiene tres celdas y llama a una función exit() en la segunda celda. Notebook2 tiene cinco celdas y llama a run(notebook1) en la tercera celda. Al ejecutar Notebook2, Notebook1 se detiene en la segunda celda al pulsar la función exit(). Notebook2 sigue ejecutando su cuarta y quinta celda.

mssparkutils.notebook.exit("value string")

Por ejemplo:

El cuaderno Sample1 con las dos celdas siguientes:

• La celda 1 define un parámetro input con un valor predeterminado establecido en 10.

• La celda 2 sale del cuaderno con el valor de input como valor de salida.

Puede ejecutar Sample1 en otro cuaderno con los valores predeterminados:

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

Salida:

Notebook executed successfully with exit value 10

Puede ejecutar Sample1 en otro cuaderno y establecer el valor de input en 20:

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

Salida:

Notebook executed successfully with exit value 20

Utilidades de credenciales

Puedes usar las utilidades de credenciales MSSparkUtils para obtener tokens de acceso y gestionar secretos en Azure Key Vault.

Ejecute el siguiente comando para obtener información general sobre los métodos disponibles:

mssparkutils.credentials.help()

Salida:

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

Obtener el token

getToken devuelve un token de Microsoft Entra para la audiencia especificada y un nombre (opcional). En la lista siguiente se muestran las claves de público disponibles actualmente:

• Recurso para el público de almacenamiento: storage
• Recurso de Power BI:pbi
• Azure Key Vault Resource:keyvault
• Recurso de bases de datos Synapse RTA KQL: kusto

Ejecute el siguiente comando para obtener el token:

mssparkutils.credentials.getToken('audience Key')

Consigue el secreto usando las credenciales de usuario

getSecret devuelve un secreto de Azure Key Vault para un punto de conexión de Azure Key Vault y un nombre del secreto especificados mediante credenciales de usuario.

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

Montaje y desmontaje de archivos

Fabric admite los siguientes escenarios de montaje en el paquete de utilidades de Microsoft Spark. Puedes usar las APIs mount, unmount, getMountPath() y mounts() para conectar almacenamiento remoto (Azure Data Lake Storage Gen2) a todos los nodos en funcionamiento (nodos driver y nodos de trabajo). Una vez instalado el punto de montaje de almacenamiento, utilice la API de archivos locales para acceder a los datos como si estuvieran almacenados en el sistema de archivos local.

Cómo montar una cuenta de Azure Data Lake Storage Gen2

El siguiente ejemplo muestra cómo montar Azure Data Lake Storage Gen2. El montaje de almacenamiento en blobs funciona de manera similar.

En este ejemplo se supone que tiene una cuenta de Data Lake Storage Gen2 denominada storegen2 y que la cuenta tiene un contenedor denominado mycontainer que desea montar en /test en la sesión de Spark del cuaderno.

Para montar el contenedor llamado mycontainer, mssparkutils primero comprueba si tienes permiso para acceder al contenedor. Fabric admite tres métodos de autenticación para la operación de montaje del desencadenador: token de Microsoft Entra (predeterminado y recomendado), accountKey y sastoken. Para más información sobre la autenticación de tokens de Microsoft Entra y la API notebookutils actual, consulte Montaje y desmontaje de archivos con NotebookUtils para Fabric.

Montar usando un token de firma de acceso compartido o una clave de cuenta

MSSparkUtils permite pasar explícitamente una clave de cuenta o un token de firma de acceso compartido (SAS) como parámetro para montar el recurso de destino.

Por motivos de seguridad, se recomienda almacenar las claves de cuenta o los tokens de SAS en Azure Key Vault (como se muestra en la siguiente captura de pantalla). A continuación, puede recuperarlos mediante la API mssparkutils.credentials.getSecret API. Para obtener más información sobre Azure Key Vault, consulte Acerca de las claves de cuenta de almacenamiento administradas de Azure Key Vault.

Código de ejemplo para el método 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}
)

Código de ejemplo para 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

Parámetros de montaje:

• fileCacheTimeout: Los blobs se almacenan en caché en la carpeta temporal local durante 120 segundos de forma predeterminada. Durante este tiempo, blobfuse no comprueba si el archivo está actualizado. Configura este parámetro para cambiar el tiempo de espera por defecto. Cuando varios clientes modifican los archivos al mismo tiempo, para evitar inconsistencias entre archivos locales y remotos, recomendamos acortar el tiempo de caché, o incluso cambiarlo a 0, y obtener siempre los archivos más recientes del servidor.
• timeout: El tiempo de espera de la operación de montaje es de 120 segundos por defecto. Configura este parámetro para cambiar el tiempo de espera por defecto. Cuando hay demasiados ejecutores o cuando se agota el tiempo de espera del montaje, recomendamos aumentar el valor.

Puede usar estos parámetros de la siguiente manera:

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

Cómo montar un lakehouse

Código de ejemplo para montar un lakehouse en /test:

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

Accede a los archivos en el punto de montaje mediante la API mssparkutils fs

El objetivo principal de la operación de montaje es permitirte acceder a los datos almacenados en una cuenta de almacenamiento remoto usando una API local del sistema de archivos. También puede acceder a los datos por medio de la API mssparkutils fs con una ruta de acceso montada como parámetro. Este formato de ruta de acceso es un poco diferente.

Supongamos que montaste el contenedor de Data Lake Storage Gen2 mycontainer en /test mediante la API de montaje. Cuando accedes a los datos usando una API local del sistema de archivos, el formato de ruta es así:

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

Cuando quieres acceder a los datos usando la API mssparkutils fs , te recomendamos que uses getMountPath() para obtener la ruta precisa:

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

• Enumerar directorios:

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

• Leer el contenido de archivos:

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

• Crear un directorio:

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

Acceder a los archivos en el punto de montaje mediante la ruta de acceso local

Puede leer y escribir fácilmente los archivos en el punto de montaje mediante el sistema de archivos estándar. Este es un ejemplo de 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"))

Cómo comprobar los puntos de montaje existentes

Puede usar la API mssparkutils.fs.mounts() para comprobar toda la información existente sobre los puntos de montaje:

mssparkutils.fs.mounts()

Cómo desmontar el punto de montaje.

Use el siguiente código para desmontar el punto de montaje (/test en este ejemplo):

mssparkutils.fs.unmount("/test")

Limitaciones conocidas

• La montura actual es una configuración a nivel de trabajo. Recomendamos que utilices la API de montajes para comprobar si existe o no existe un punto de montaje.

• El mecanismo de desmontaje no es automático. Cuando termine la ejecución de la aplicación, para desmontar el punto de montaje y liberar el espacio en disco, debe llamar explícitamente a una API de desmontaje en el código. De lo contrario, el punto de montaje sigue existiendo en el nodo una vez finalizada la ejecución de la aplicación.

• No se soporta montar una cuenta de almacenamiento de Azure Data Lake Storage Gen1.

Utilidades de Lakehouse

El mssparkutils.lakehouse módulo proporciona utilidades para gestionar los elementos de la casa del lago. Estas utilidades facilitan crear, recuperar, actualizar y eliminar elementos de la casa del lago.

Información general de los métodos

Los siguientes métodos están disponibles en el mssparkutils.lakehouse módulo:

# 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]

Ejemplos de uso

Para utilizar estos métodos de forma eficaz, consideren los siguientes ejemplos de uso:

Crear un objeto de casa de lago

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

Recuperar un objeto de la casa del lago

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

Actualizar un elemento de la casa del lago

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

Eliminar un elemento de la casa del lago

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

Listado de objetos de la casa del lago

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

Información adicional

Para obtener información más detallada sobre cada método y sus parámetros, utilice la mssparkutils.lakehouse.help("methodName") función.

Al utilizar las utilidades Lakehouse de MSSparkUtils, puedes gestionar tus elementos de lakehouse de forma más eficiente e integrar esta gestión en tus pipelines de Fabric, mejorando así tu experiencia global en la gestión de datos.

Explora estas utilidades e intégralas en tus flujos de trabajo de Fabric para una gestión fluida de los artículos en la casa del lago.

Utilidades en tiempo de ejecución

Mostrar la información de contexto de sesión

Al usar mssparkutils.runtime.context, puedes obtener la información de contexto de la sesión activa actual, incluido el nombre del cuaderno de notas, el lakehouse predeterminado, la información del espacio de trabajo, si se trata de una ejecución de canalización, entre otros datos.

mssparkutils.runtime.context

Problema conocido

Cuando usas una versión de runtime posterior a la 1.2 y ejecutasmssparkutils.help(), las APIsde fabricClient, warehouse y workspace listadas no están actualmente soportadas.

Contenido relacionado

• Administración de bibliotecas