Utilitários Microsoft Spark (MSSparkUtils) para malha

O Microsoft Spark Utilities (MSSparkUtils) é um pacote incorporado que o ajuda a realizar tarefas comuns facilmente. Use MSSparkUtils para trabalhar com sistemas de arquivos, obter variáveis de ambiente, encadear blocos de anotações e trabalhar com segredos. O pacote MSSparkUtils está disponível em PySpark (Python), Scala, blocos de notas do SparkR e pipelines do Fabric.

Utilitários do sistema de arquivos

mssparkutils.fs fornece utilitários para trabalhar com vários sistemas de ficheiros, incluindo Azure Data Lake Storage Gen2 e Armazenamento de Blobs do Azure. Certifique-se de configurar 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.

MSSparkUtils funciona com o sistema de arquivos da mesma forma que as APIs do Spark. Veja-se o mssparkuitls.fs.mkdirs() e o uso em casas de lago, por exemplo:

Listar ficheiros

Para listar o conteúdo de um diretório, use mssparkutils.fs.ls('Seu caminho de 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 

Ver propriedades do ficheiro

Este método devolve propriedades do ficheiro, incluindo o nome do ficheiro, o caminho do ficheiro, o tamanho do ficheiro e se é um diretório ou um ficheiro.

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

Criar novo diretório

Este método cria o diretório especificado caso este 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 ficheiro

Esse método copia um arquivo ou diretório e oferece suporte à atividade de cópia entre 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

Cópia de ficheiro eficiente

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 conteúdo do arquivo

Este método retorna até aos primeiros maxBytes bytes do ficheiro 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 ficheiro

Esse método move um arquivo ou diretório e oferece 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

Este método grava a seqüência dada em um arquivo, codificado 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 ficheiro

Este método acrescenta a cadeia de caracteres dada a um arquivo, codificado 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 a utilização detalhada, consulte Montar e desmontar ficheiros.

Utilitários para notebook

Use o MSSparkUtils Notebook Utilities para executar um bloco de anotações ou sair de um bloco de anotações 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 bloco de notas

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

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

Por exemplo:

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

O caderno no Fabric também oferece suporte à referência de cadernos em vários espaços de trabalho ao especificar 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 do código e permite fazer debug facilmente de uma execução de referência específica.

Referência: executar vários blocos de anotações em paralelo

O método mssparkutils.notebook.runMultiple() permite executar vários blocos de anotações em paralelo ou com uma estrutura topológica predefinida. A API utiliza uma implementação com múltiplas threads para submeter, colocar em fila e monitorizar notebooks filhos que são executados em instâncias REPL isoladas (read-eval-print-loop) na sessão Spark existente. Os cadernos filhos referenciados partilham os recursos computacionais da sessão.

Com mssparkutils.notebook.runMultiple(), pode:

• Execute vários notebooks simultaneamente, sem esperar que cada um termine.

• Especifique as dependências e a ordem de execução de seus blocos de anotações, usando um formato JSON simples.

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

• Veja as capturas de cada registo de execução de notebook na saída e depure e monitorize as tarefas do seu notebook de forma prática.

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

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

Aqui está um exemplo simples de execução de uma lista de blocos de anotações em paralelo usando esse método:

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

O resultado da execução do bloco de anotações raiz é o seguinte:

O exemplo seguinte mostra cadernos em execução com uma estrutura topológica usando mssparkutils.notebook.runMultiple(). Use esse método para orquestrar facilmente blocos de anotações 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 bloco de anotações raiz é o seguinte:

Sair de um bloco de notas

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

• Quando se chama uma função exit() de um bloco de anotações interativamente, o bloco de anotações Fabric lança uma exceção, ignora a execução das células subsequentes e mantém a sessão do Spark ativa.

• Ao orquestrar um notebook num 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 encerra a sessão do Spark. Não coloque a função exit() dentro de um bloco try/catch, pois esta exceção NotebookExit tem de ser propagada para que o pipeline receba o valor de retorno.

• Quando chama uma função exit() num caderno que está a ser referenciada, o Fabric Spark interrompe a execução adicional do caderno referenciado e continua a executar as próximas células no caderno 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 as chamadas são executadas (notebook1) na terceira célula. Quando você executa o Notebook2, o Notebook1 para na segunda célula ao pressionar a função exit( ). O Notebook2 continua a executar a quarta célula e a quinta célula.

mssparkutils.notebook.exit("value string")

Por exemplo:

Exemplo1 bloco de notas com as seguintes duas células:

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

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

Você pode executar o Sample1 em outro bloco de anotações com valores padrão:

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

Saída:

Notebook executed successfully with exit value 10

Você pode executar o Sample1 em outro bloco de anotações 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

Pode usar as Utilidades de Credenciais MSSparkUtils para obter tokens de acesso e gerir 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 token

getToken devolve um token do Microsoft Entra para uma determinada audiência e nome (opcional). A lista a seguir mostra as chaves de audiência atualmente disponíveis:

• Recurso para os Destinatários 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')

Obter segredo utilizando as credenciais do utilizador

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

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

Montagem e desmontagem de ficheiros

O Fabric suporta os seguintes cenários de montagem no pacote Microsoft Spark Utilities. Pode usar as APIs mount, unmount, getMountPath() e mounts() para ligar armazenamento remoto (Azure Data Lake Storage Gen2) a todos os nós em funcionamento (nó driver e nó trabalhador). Depois que o ponto de montagem de armazenamento estiver instalado, use a API de arquivo local para acessar os dados como se estivessem armazenados no sistema de arquivos local.

Como montar uma conta Azure Data Lake Storage Gen2

O exemplo seguinte mostra como montar o Azure Data Lake Storage Gen2. A montagem do armazenamento Blob funciona de forma semelhante.

Este exemplo pressupõe que o/a utilizador(a) tenha uma conta do Data Lake Storage Gen2 chamada storegen2, e que a conta tenha um contentor chamado mycontainer que o/a utilizador(a) deseje montar em /test na sessão Spark do seu notebook.

Para montar o contentor chamado mycontainer, o mssparkutils verifica primeiro se tens permissão para aceder ao container. 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 por token do Microsoft Entra e a atual API notebookutils, consulte NotebookUtils: montar e desmontar ficheiros no Fabric.

Montar utilizando um token de assinatura de acesso partilhado ou a chave da conta

MSSparkUtils suporta a passagem explícita de uma chave de conta ou token de assinatura de acesso compartilhado (SAS) como um parâmetro para montar o destino.

Por motivos de segurança, recomendamos que você armazene chaves de conta ou tokens SAS no Cofre de Chaves do Azure (como mostra a captura de tela a seguir). Em seguida, você pode recuperá-los usando a API mssparkutils.credentials.getSecret . Para obter mais informações sobre o Azure Key Vault, consulte Sobre as chaves de conta de armazenamento gerenciado 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 armazenam-se na pasta temporária local durante 120 segundos por defeito. Durante este tempo, o blobfuse não verifica se o ficheiro está atualizado. Defina este parâmetro para alterar o timeout padrão. Quando vários clientes modificam ficheiros ao mesmo tempo, para evitar inconsistências entre ficheiros locais e remotos, recomendamos que encurta o tempo de cache, ou até o altere para 0, e que obtenha sempre os ficheiros mais recentes do servidor.
• timeout: O tempo limite da operação de montagem é de 120 segundos por defeito. Defina este parâmetro para alterar o timeout padrão. Quando há demasiados executores ou quando a operação de montagem excede o tempo limite, recomendamos que aumente o valor.

Você pode usar estes parâmetros assim:

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

Como montar uma casa no lago

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"
)

Acede aos ficheiros sob o ponto de montagem usando a API mssparkutils fs

O principal objetivo da operação de montagem é permitir-lhe aceder aos dados armazenados numa conta de armazenamento remoto usando uma API local do sistema de ficheiros. Você também pode acessar os dados usando a API mssparkutils fs com um caminho montado como parâmetro. Este formato de caminho é um pouco diferente.

Suponha que montou o contentor mycontainer do Data Lake Storage Gen2 em /test através da API de montagem. Quando acedes aos dados usando uma API local do sistema de ficheiros, o formato do caminho é o seguinte:

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

Quando quiser aceder aos dados ao utilizar a API mssparkutils fs, recomendamos que utilize getMountPath() para obter o caminho correto:

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

• Listar diretórios:

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

• Leia o conteúdo do arquivo:

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

• Crie um diretório:

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

Acesse arquivos no ponto de montagem através do caminho local

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

Como verificar os 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 a nível de trabalho. Recomendamos que utilize 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 terminar, para desmontar o ponto de montagem e liberar o espaço em disco, você precisará chamar explicitamente uma API de desmontagem em seu código. Caso contrário, o ponto de montagem ainda existe no nó após o término da execução da aplicação.

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

Utilitários Lakehouse

O mssparkutils.lakehouse módulo fornece utilidades para gerir itens da casa do lago. Estas utilidades facilitam a criação, recuperação, atualização e eliminaçã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 utilização

Para utilizar estes métodos de forma eficaz, considere os seguintes exemplos de utilização:

Criar um item de casa de lago

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

Recuperar um objeto da casa do lago

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

Atualizar um item da casa do lago

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

Eliminar 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 os seus parâmetros, use a mssparkutils.lakehouse.help("methodName") função.

Ao utilizar as utilidades Lakehouse da MSSparkUtils, pode gerir os seus itens lakehouse de forma mais eficiente e integrar esta gestão nos seus pipelines Fabric, melhorando a sua experiência global de gestão de dados.

Explore estas utilidades e incorpore-as nos seus fluxos de trabalho Fabric para uma gestão fluida de itens do lakehouse.

Utilitários de ambiente de execução

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

Ao usar mssparkutils.runtime.context, pode obter as informações de contexto da sessão ativa atual, incluindo o nome do notebook, o lakehouse predefinido, informações da área de trabalho, se se tratar de uma execução de pipeline, entre outras.

mssparkutils.runtime.context

Problema conhecido

Quando usas uma versão de runtime posterior à 1.2 e executasmssparkutils.help(), as APIsfabricClient, warehouse e workspace listadas não são atualmente suportadas.

Conteúdos relacionados

• Gestão de bibliotecas