Microsoft Spark Utilities (MSSparkUtils) para Fabric

O Microsoft Spark Utilities (MSSparkUtils) é um pacote embutido que ajuda você a realizar tarefas comuns com facilidade. Use o MSSparkUtils para trabalhar com sistemas de arquivos, obter variáveis de ambiente, encadear notebooks e trabalhar com segredos. O pacote MSSparkUtils está disponível em PySpark (Python), Scala, nos notebooks SparkR e nos pipelines do Fabric.

Utilitários do sistema de arquivos

mssparkutils.fs oferece utilitários para trabalhar com vários sistemas de arquivos, incluindo Azure Data Lake Storage Gen2 e Armazenamento de Blobs do Azure. Configure o acesso ao Azure Data Lake Storage Gen2 e ao Armazenamento de Blobs do Azure adequadamente.

Execute os seguintes comandos para obter uma visão geral dos métodos disponíveis:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Saída

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.

O MSSparkUtils funciona com o sistema de arquivos da mesma forma que as APIs do Spark. Pegue como exemplo mssparkuitls.fs.mkdirs() e o uso de casas de lago:

Listar arquivos

Para listar o conteúdo de um diretório, use mssparkutils.fs.ls("O caminho do seu diretório"). Por exemplo:

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 

Exibir propriedades de arquivo

Esse método retorna propriedades do arquivo, incluindo o nome do arquivo, caminho do arquivo, tamanho do arquivo e se é um diretório ou um arquivo.

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

Criar um diretório

Esse método cria o diretório especificado caso ele não exista, e cria quaisquer diretórios pais necessários.

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 arquivo

Esse método copia um arquivo ou diretório e dá suporte à atividade de cópia em sistemas de arquivos.

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

Arquivo de cópia com desempenho

Esse método fornece uma maneira mais rápida de copiar ou mover arquivos, especialmente grandes volumes de dados.

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

Visualizar o conteúdo do arquivo

Esse método retorna até os primeiros maxBytes bytes do arquivo especificado como uma string codificada em UTF-8.

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

Mover arquivo

Esse método move um arquivo ou diretório e dá suporte a movimentações entre sistemas de arquivos.

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.

Gravar arquivo

Esse método grava a cadeia de caracteres fornecida em um arquivo, codificada em 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

Acrescentar conteúdo a um arquivo

Esse método acrescenta a cadeia de caracteres especificada a um arquivo, codificada em 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

Excluir arquivo ou diretório

Esse método remove um arquivo ou diretório.

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

Diretório de montagem/desmontagem

Para mais informações sobre uso detalhado, veja Montagem e desmontagem de arquivos.

Utilitários de notebook

Use os Utilitários de Notebook MSSparkUtils para executar ou sair de um notebook com um valor. Execute o seguinte comando para obter uma visão geral dos métodos disponíveis:

mssparkutils.notebook.help()

Saída:

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.

Referenciar um notebook

Esse método faz referência a um notebook e retorna seu valor de saída. Você pode executar o aninhamento de chamadas de função em um notebook interativamente ou em um pipeline. O notebook que está sendo referenciado é executado no pool do Spark em que o notebook chama essa função.

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

Por exemplo:

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

O Fabric Notebook também oferece suporte ao referenciamento de notebooks em vários espaços de trabalho, especificando a ID do espaço de trabalho.

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

Você pode abrir o link de instantâneo da execução de referência na saída da célula. O instantâneo captura os resultados da execução de código e permite que você debugue facilmente uma execução de referência.

A referência executa vários notebooks paralelamente

O método mssparkutils.notebook.runMultiple() permite executar vários notebooks paralelamente ou com uma estrutura topológica predefinida. A API utiliza uma implementação multithread para enviar, enfileirar e monitorar notebooks filhos que executam em instâncias REPL isoladas (read-eval-print-loop) dentro da sessão Spark existente. Os cadernos filhos referenciados compartilham os recursos computacionais da sessão.

Com mssparkutils.notebook.runMultiple(), você pode:

• Execute vários notebooks simultaneamente, sem aguardar a conclusão de cada um.

• Especifique as dependências e a ordem de execução para seus notebooks usando um formato JSON simples.

• Otimize o uso de recursos de computação do Spark e reduza o custo dos seus projetos do Fabric.

• Visualize os snapshots de cada registro de execução do notebook na saída, e depure e monitore suas tarefas do notebook de forma conveniente.

• Obtenha o valor de saída de cada atividade executiva e use-as em tarefas downstream.

Você também pode tentar executar o mssparkutils.notebook.help("runMultiple") para localizar o exemplo e o uso detalhado.

Este é um exemplo simples de como executar uma lista de notebooks paralelamente usando este método:

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

O resultado da execução do notebook raiz é o seguinte:

O exemplo a seguir mostra cadernos em execução com uma estrutura topológica usando mssparkutils.notebook.runMultiple(). Use esse método para orquestrar facilmente notebooks por meio de uma experiência 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})

O resultado da execução do notebook raiz é o seguinte:

Sair de um notebook

Esse método sai de um notebook com um valor. Você pode executar o aninhamento de chamadas de função em um notebook interativamente ou em um pipeline.

• Quando você chama uma função exit() de um notebook interativamente, o notebook do Fabric gera uma exceção, ignora as células seguintes em execução e mantém a sessão do Spark ativa.

• Quando você orquestra um notebook no pipeline que chama uma função exit(), a atividade do Notebook retorna com um valor de saída, conclui a execução do pipeline e interrompe a sessão do Spark. Não coloque a função exit() dentro de um bloco try/catch, pois essa exceção NotebookExit precisa se propagar para que o pipeline obtenha o valor de retorno.

• Quando você chama uma função exit() em um notebook que está sendo referenciado, o Fabric Spark interrompe a execução subsequente do notebook referenciado e continua a executar as próximas células no notebook principal que chama a função run(). Por exemplo: Notebook1 tem três células e chama uma função exit() na segunda célula. O Notebook2 tem cinco células e chama run(notebook1) na terceira célula. Quando você executa o Notebook2, o Notebook1 para na segunda célula quando a função exit() é executada. Notebook2 continua a executar a quarta célula e a quinta célula.

mssparkutils.notebook.exit("value string")

Por exemplo:

Notebook Sample1 com as duas células seguintes:

• A célula 1 define um parâmetro de entrada com valor padrão 10.

• A célula 2 sai do notebook com entrada como valor de saída.

Você pode executar Sample1 em outro notebook com valores padrão:

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

Saída:

Notebook executed successfully with exit value 10

Você pode executar Sample1 em outro notebook e definir o valor de entrada como 20:

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

Saída:

Notebook executed successfully with exit value 20

Utilitários de credenciais

Você pode usar as Utilidades de Credenciais MSSparkUtils para obter tokens de acesso e gerenciar segredos no Azure Key Vault.

Execute o seguinte comando para obter uma visão geral dos métodos disponíveis:

mssparkutils.credentials.help()

Saída:

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

Obter o token

getTokenretorna um token Microsoft Entra para um determinado público e nome (opcional). A lista a seguir mostra as chaves de público disponíveis no momento:

• Recurso de público de armazenamento: storage
• Recurso Power BI:pbi
• Azure Key Vault Resource:keyvault
• Synapse RTA KQL DB Resource: kusto

Execute o seguinte comando para obter o token:

mssparkutils.credentials.getToken('audience Key')

Obtenha o segredo usando as credenciais do usuário

getSecretretorna um segredo do Azure Key Vault para um dado endpoint e nome secreto do Azure Key Vault usando credenciais de usuário.

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

Montagem e desmontagem de arquivos

O Fabric dá suporte aos seguintes cenários de montagem no pacote Microsoft Spark Utilities. Você pode usar as APIs mount, unmount, getMountPath() e mounts() para anexar o armazenamento remoto (Azure Data Lake Storage Gen2) a todos os nós de trabalho (nó do driver e nós de trabalho). Depois que o ponto de montagem do armazenamento estiver estabelecido, use a API de arquivo local para acessar dados como se estivessem armazenados no sistema de arquivos local.

Como montar uma conta Azure Data Lake Storage Gen2

O exemplo a seguir mostra como montar o Azure Data Lake Storage Gen2. O processo de montagem do Armazenamento de Blobs funciona de maneira semelhante.

Esse exemplo pressupõe que você tenha uma conta do Data Lake Storage Gen2 chamada storegen2 e que a conta tenha um contêiner chamado mycontainer que você deseja montar para /test em sua sessão do Notebook Spark.

Para montar o contêiner chamado mycontainer, o mssparkutils primeiro verifica se você tem permissão para acessar o contêiner. O Fabric suporta três métodos de autenticação para a operação de montagem do gatilho: token Microsoft Entra (padrão e recomendado), accountKey e sastoken. Para mais informações sobre a autenticação de token do Microsoft Entra e a atual notebookutils API, consulte Montagem e desmontagem de arquivos com o NotebookUtils para o Fabric.

Monte usando um token de assinatura de acesso compartilhado ou chave de conta

O MSSparkUtils suporta a passagem explícita de uma chave de conta ou de um token de Assinatura de Acesso Compartilhado (SAS) como um parâmetro para montar o alvo.

Por motivos de segurança, recomendamos que você armazene chaves de conta ou tokens SAS no Azure Key Vault (como mostra a captura de tela de exemplo a seguir). Você pode recupera-los usando a API mssparkutils.credentials.getSecret. Para obter mais informações, consulte Sobre as chaves de conta de armazenamento gerenciadas do Azure Key Vault.

Código de exemplo para o 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 exemplo 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 montagem:

• fileCacheTimeout: Os blobs ficam em cache na pasta temporária local por 120 segundos por padrão. Durante esse tempo, o blobfuse não verifica se o arquivo está atualizado. Defina esse parâmetro para alterar o tempo padrão de expiração. Quando vários clientes modificam arquivos ao mesmo tempo, para evitar inconsistências entre arquivos locais e remotos, recomendamos que você reduza o tempo de cache, ou até mesmo o mude para 0, e sempre obtenha os arquivos mais recentes do servidor.
• timeout: O tempo limite da operação de montagem é de 120 segundos por padrão. Defina esse parâmetro para alterar o tempo padrão de expiração. Quando há muitos executores ou quando a montagem expira por tempo limite, recomendamos que você aumente esse valor.

Você pode usar esses parâmetros como este:

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

Como montar um lakehouse

Código de exemplo para montar uma casa de lago para /test:

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

Acesse arquivos no ponto de montagem por meio da API mssparkutils fs

O principal objetivo da operação de montagem é permitir que você acesse os dados armazenados em uma conta de armazenamento remoto usando uma API local do sistema de arquivos. Você também pode acessar os dados usando a API mssparktuils fs com um caminho montado como parâmetro. Esse formato de caminho é um pouco diferente.

Suponha que você montou o container mycontainer/test do Data Lake Storage Gen2 usando a API mount. Quando você acessa os dados usando uma API de sistema de arquivos local, o formato do caminho é o seguinte:

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

Quando quiser acessar os dados usando a API mssparkutils fs, recomendamos que você use getMountPath() para obter o caminho correto:

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

• Listar diretórios:

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

• Ler conteúdo do arquivo:

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

• Criar um diretório:

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

Acessar arquivos no ponto de montagem por meio do caminho local

Você pode ler e gravar facilmente os arquivos no ponto de montagem usando o sistema de arquivos padrão. Aqui está um exemplo do 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"))

Como verificar pontos de montagem existentes

Você pode usar a API mssparkutils.fs.mounts() para verificar todas as informações de ponto de montagem existentes:

mssparkutils.fs.mounts()

Como desmontar o ponto de montagem

Use o código a seguir para desmontar o ponto de montagem (/test neste exemplo):

mssparkutils.fs.unmount("/test")

Limitações conhecidas

• A montagem atual é uma configuração em nível de trabalho. Recomendamos que você use a API mounts para verificar se existe ou não existe um ponto de montagem.

• O mecanismo de desmontagem não é automático. Quando a execução do aplicativo for concluída, para liberar o espaço em disco, você precisará chamar explicitamente uma API de desmontagem em seu código para desmontar o ponto de montagem. Caso contrário, o ponto de montagem ainda permanece no nó após a conclusão da execução do aplicativo.

• A montagem de uma conta de armazenamento Azure Data Lake Storage Gen1 não é suportada.

Utilitários do Lakehouse

O mssparkutils.lakehouse módulo fornece utilidades para gerenciar itens da casa do lago. Essas utilidades facilitam a criação, recuperação, atualização e exclusão de itens do lakehouse.

Visão geral dos métodos

Os seguintes métodos estão disponíveis no 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]

Exemplos de uso

Para usar esses métodos de forma eficaz, considere os seguintes exemplos de uso:

Criando um item de casa de lago

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

Recuperando um item da casa do lago

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

Atualizando um item de casa do lago

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

Deletando um item da casa do lago

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

Listagem de itens da casa do lago

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

Informações adicionais

Para informações mais detalhadas sobre cada método e seus parâmetros, use a mssparkutils.lakehouse.help("methodName") função.

Ao usar as utilidades Lakehouse do MSSparkUtils, você pode gerenciar seus itens do lakehouse de forma mais eficiente e integrar esse gerenciamento aos seus pipelines Fabric, melhorando sua experiência geral de gerenciamento de dados.

Explore essas utilidades e incorpore-as aos seus fluxos de trabalho do Fabric para gerenciar itens de lakehouse sem falhas.

Utilitários de runtime

Mostrar as informações de contexto da sessão

Ao usar mssparkutils.runtime.context, você pode obter informações de contexto da sessão ativa atual, incluindo o nome do notebook, o lakehouse padrão, informações do espaço de trabalho, se é uma execução de pipeline, entre outras.

mssparkutils.runtime.context

Problema conhecido

Quando você usa uma versão de runtime posterior à 1.2 e executa mssparkutils.help(), as APIs listadas do fabricClient, warehouse e workspace não são suportadas atualmente.

Conteúdo relacionado

• Gerenciamento de biblioteca