Служебные программы Microsoft Spark (MSSparkUtils) для Fabric

Служебные программы Microsoft Spark (MSSparkUtils) — это встроенный пакет, помогающий легко выполнять распространенные задачи. С помощью MSSparkUtils можно работать с файловыми системами, получать переменные среды, связывать записные книжки и работать с секретами. Пакет MSSparkUtils доступен в конвейерах PySpark (Python), Scala, SparkR notebooks и Fabric.

Служебные программы файловой системы

mssparkutils.fs предоставляет служебные программы для работы с различными файловыми системами, включая Azure Data Lake Storage (ADLS) 2-го поколения и Хранилище BLOB-объектов Azure. Убедитесь, что доступ к Azure Data Lake Storage 2-го поколения и Хранилищу BLOB-объектов Azure настроен правильно.

Чтобы получить общие сведения о доступных методах, выполните следующие команды:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Выходные данные

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 работает с файловой системой так же, как и с API Spark. Возьмем, к примеру, использование mssparkuitls.fs.mkdirs() и Fabric lakehouse:

Перечень файлов

Чтобы получить список содержимого каталога, используйте mssparkutils.fs.ls("Путь к каталогу") Например:

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 

Просмотр свойств файла.

Этот метод возвращает свойства файла, включая имя файла, путь к файлу, размер файла и то, является ли он каталогом и файлом.

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

Создать новый каталог

Этот метод создает указанный каталог, если он не существует, и создает все необходимые родительские каталоги.

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 

Копировать файл

Этот метод копирует файл или каталог и поддерживает действие копирования в файловых системах.

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

Эффективное копирование файлов

Этот метод обеспечивает более быстрый способ копирования или перемещения файлов, особенно больших объемов данных.

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

Предварительный просмотр содержимого файла

Этот метод возвращает до первых байтов maxBytes заданного файла в виде строки, закодированной в UTF-8.

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

Переместить файл

Этот метод перемещает файл или каталог и поддерживает перемещение между файловыми системами.

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.

Записать в файл

Этот метод записывает указанную строку в файл, закодированный в 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

Добавить содержимое в файл

Этот метод добавляет заданную строку в файл, закодированный в 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

Удалить файл или каталог

Этот метод удаляет файл или каталог.

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

Каталог mount/unmount

Дополнительные сведения о подробном использовании приведены в разделе Подключение и отключение файлов.

Служебные программы для ноутбуков

Используйте служебные программы записной книжки MSSparkUtils для запуска записной книжки или выхода из записной книжки со значением. Чтобы получить общие сведения о доступных методах, используйте следующую команду:

mssparkutils.notebook.help()

Выходные данные:

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.

Ссылка на записную книжку

Этот метод ссылается на записную книжку и возвращает значение выхода. Вызовы вложенных функций можно запускать в записной книжке в интерактивном режиме или в конвейере. Записная книжка, на которую ссылаются, выполняется в Spark-пуле той же записной книжки, которая вызывает эту функцию.

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

Например:

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

Записная книжка Fabric также поддерживает ссылки на записные книжки в нескольких рабочих областях, указав идентификатор рабочей области.

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

Вы можете открыть ссылку моментального снимка эталонного запуска в выводе ячейки. Моментальный снимок записывает результаты выполнения кода и позволяет легко отлаживать эталонный запуск.

Ссылка на выполнение нескольких блокнотов в параллельном режиме

Этот метод mssparkutils.notebook.runMultiple() позволяет выполнять несколько записных книжек параллельно или с предопределенной топологической структурой. API использует многопотоковую реализацию для отправки, постановки в очередь и отслеживания дочерних ноутбуков, которые выполняются в изолированных экземплярах REPL (чтение – оценка – печать – цикл) в существующем сеансе Spark. Вычислительные ресурсы сессий разделяются упомянутыми дочерними тетрадями.

С помощью mssparkutils.notebook.runMultiple():

• Одновременно выполняйте несколько блокнотов, не дожидаясь завершения каждого из них.

• Укажите зависимости и порядок выполнения записных книжек с помощью простого формата JSON.

• Оптимизируйте использование вычислительных ресурсов Spark и уменьшите затраты на проекты Fabric.

• Просматривайте снимки выполнения каждой записной книжки и удобно отлаживайте и контролируйте задачи в ней.

• Получите результат выполнения каждой выполняемой задачи и используйте их в зависимых задачах.

Вы также можете попытаться запустить mssparkutils.notebook.help("runMultiple"), чтобы найти пример и подробное использование.

Ниже приведен простой пример запуска списка записных книжек параллельно с помощью этого метода:

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

Результат выполнения корневой записной книжки выглядит следующим образом:

Ниже приведен пример запуска записных книжек с топологической структурой с помощью mssparkutils.notebook.runMultiple(). Используйте этот метод для легкой оркестрации записных книжек путем взаимодействия через код.

# 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})

Результат выполнения корневой записной книжки выглядит следующим образом:

Выход из ноутбука

Этот метод закрывает записную книжку со значением. Вызовы вложенных функций можно запускать в записной книжке в интерактивном режиме или в конвейере.

• При вызове функции exit() из записной книжки в интерактивном режиме записная книжка Fabric создает исключение, пропускает выполнение последующих ячеек и сохраняет сеанс Spark активным.

• При оркестрации записной книжки в конвейере, который вызывает функцию exit(), действие записной книжки возвращается со значением выхода, завершает выполнение конвейера и останавливает сеанс Spark.

• При вызове функции exit() в записной книжке, на которую ссылается ссылка, Fabric Spark остановит дальнейшее выполнение записной книжки, на которую ссылается ссылка, и продолжит выполнять следующие ячейки в главной записной книжке, которая вызывает функцию run(). Например: Notebook1 имеет три ячейки и вызывает функцию exit() во второй ячейке. Notebook2 содержит пять ячеек и вызывает run(notebook1) в третьей ячейке. При запуске Notebook2 Записная книжка1 останавливается во второй ячейке при нажатии функции exit(). Notebook2 продолжает выполнять свои четвертую и пятую ячейки.

mssparkutils.notebook.exit("value string")

Например:

Пример1 записной книжки со следующими двумя ячейками:

• Ячейка 1 определяет входной параметр со значением по умолчанию, равным 10.

• Ячейка 2 выходит из записной книжки с входным значением.

Вы можете запустить Sample1 в другой записной книжке со значениями по умолчанию:

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

Выходные данные:

Notebook executed successfully with exit value 10

Вы можете запустить Sample1 в другом ноутбуке и установить значение input на 20:

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

Выходные данные:

Notebook executed successfully with exit value 20

Утилиты для учётных записей

Служебные программы учетных данных MSSparkUtils можно использовать для получения маркеров доступа и управления секретами в Azure Key Vault.

Чтобы получить общие сведения о доступных методах, используйте следующую команду:

mssparkutils.credentials.help()

Выходные данные:

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

Получение токена

getToken возвращает токен Microsoft Entra для заданной аудитории и имени (необязательно). В следующем списке показаны доступные в настоящее время ключи аудитории:

• Ресурс аудитории для хранения: "хранилище"
• Ресурс Power BI: "pbi"
• Azure Key Vault Ресурс: keyvault
• Synapse RTA KQL DB Ресурс: "kusto"

Выполните следующую команду, чтобы получить маркер:

mssparkutils.credentials.getToken('audience Key')

Получить секрет с помощью учетных данных пользователя

getSecret возвращает секрет Azure Key Vault для заданной конечной точки Azure Key Vault и имени секрета с использованием учетных данных пользователя.

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

Монтирование и размонтирование файлов

Fabric поддерживает следующие сценарии подключения в пакете служебных программ Microsoft Spark. Вы можете использовать API mount, unmount, getMountPath() и mounts() для подключения удаленного хранилища (ADLS Gen2) ко всем рабочим узлам (узлу драйвера и рабочим узлам). После установки точки подключения хранилища используйте локальный API файлов для доступа к данным, как будто он хранится в локальной файловой системе.

Подключение учетной записи ADLS 2-го поколения

В следующем примере показано, как подключить Azure Data Lake Storage 2-го поколения. Подключение облачного хранилища осуществляется аналогичным образом.

В этом примере предполагается, что у вас есть одна учетная запись Data Lake Storage 2-го поколения с именем storegen2, и в этой учетной записи есть один контейнер с именем mycontainer, который вы хотите подключить к /test в вашем сеансе Spark в записной книжке.

Чтобы подключить контейнер с именем mycontainer, mssparkutils сначала необходимо проверить, есть ли у вас разрешение на доступ к контейнеру. В настоящее время Fabric поддерживает два метода проверки подлинности для операции подключения триггера: accountKey и sastoken.

Подключение с помощью токена подписи для общего доступа или ключа учетной записи

MSSparkUtils поддерживает явную передачу ключа учетной записи или маркера общего доступа (SAS) в качестве параметра для подключения к целевому объекту.

По соображениям безопасности рекомендуется хранить ключи учетной записи или маркеры SAS в Azure Key Vault (как показано на следующем снимке экрана). Затем их можно получить с помощью API mssparkutils.credentials.getSecret . Дополнительные сведения об Azure Key Vault см. в статье "Сведения о ключах управляемой учетной записи хранения Azure Key Vault".

Пример кода для метода 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}
)

Пример кода для 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

Параметры подключения:

• fileCacheTimeout: Blobs будут по умолчанию кэшироваться в локальной временной папке в течение 120 секунд. В течение этого времени blobfuse не будет проверять, обновлен ли файл. Параметр может быть задан для изменения времени ожидания по умолчанию. При одновременном изменении нескольких клиентов файлов, чтобы избежать несоответствий между локальными и удаленными файлами, рекомендуется сократить время кэша или даже изменить его на 0 и всегда получать последние файлы с сервера.
• время ожидания: время ожидания операции подключения составляет 120 секунд по умолчанию. Параметр может быть задан для изменения времени ожидания по умолчанию. Если количество исполнителей слишком велико или происходит тайм-аут при монтировании, рекомендуется увеличить значение.

Эти параметры можно использовать следующим образом:

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

Как подключить озеро

Пример кода для подключения lakehouse к /test:

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

Доступ к файлам в точке подключения с помощью API mssparktuils fs

Основной целью операции подключения является предоставление клиентам доступа к данным, хранящимся в удаленной учетной записи хранения, с помощью API локальной файловой системы. Вы также можете получить доступ к данным с помощью API mssparkutils fs с монтированным путем в виде параметра. Этот формат пути немного отличается.

Предположим, что вы подключили контейнер Data Lake Storage 2-го поколения mycontainer к /test с помощью API подключения. При доступе к данным с помощью локального API файловой системы формат пути выглядит следующим образом:

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

Если вы хотите получить доступ к данным с помощью API mssparkutils fs, рекомендуется использовать getMountPath(), чтобы получить точный путь:

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

• Список каталогов:

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

• Прочитать содержимое файла:

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

• Создание каталога:

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

Доступ к файлам под точкой подключения через локальный путь

Вы можете легко считывать и записывать файлы в точке подключения с помощью стандартной файловой системы. Ниже приведен пример 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"))

Как проверить существующие точки подключения

Api mssparkutils.fs.mounts() можно использовать для проверки всех существующих сведений о точке подключения:

mssparkutils.fs.mounts()

Отмонтирование точки монтирования

Используйте следующий код, чтобы отключить точку подключения (/test в этом примере):

mssparkutils.fs.unmount("/test")

Известные ограничения

• Текущая точка монтирования — это конфигурация уровня задания; рекомендуется использовать API точек монтирования, чтобы проверить, существует ли точка монтирования или недоступна.

• Механизм демонтажа не является автоматическим. Когда приложение завершит работу, чтобы отключить точку подключения и освободить место на диске, необходимо явно вызвать API отключения в коде. В противном случае точка подключения по-прежнему будет существовать в узле после завершения работы приложения.

• Подключение учетной записи хранения ADLS 1-го поколения не поддерживается.

Служебные программы Lakehouse

mssparkutils.lakehouse предоставляет утилиты, специально предназначенные для управления артефактами Lakehouse. Эти служебные программы позволяют пользователям создавать, извлекать, обновлять и удалять артефакты Lakehouse легко.

Обзор методов

Ниже приведен обзор доступных методов, предоставляемых 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]

Примеры использования

Чтобы эффективно использовать эти методы, рассмотрим следующие примеры использования:

Создание артефакта Lakehouse

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

Получение артефакта Lakehouse

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

Обновление артефакта Lakehouse

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

Удаление артефакта Lakehouse

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

Перечисление артефактов Lakehouse

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

Дополнительная информация:

Для получения дополнительных сведений о каждом методе и его параметрах используйте функцию mssparkutils.lakehouse.help("methodName") .

С помощью служебных программ MSSparkUtils' Lakehouse управление артефактами Lakehouse становится более эффективным и интегрированным в потоки Fabric, улучшая общий опыт управления данными.

Вы можете изучить эти служебные программы и включить их в рабочие процессы Fabric для простого управления артефактами Lakehouse.

Служебные программы среды выполнения

Отображение сведений о контексте сеанса

С mssparkutils.runtime.context помощью контекста текущего динамического сеанса можно получить сведения о контексте, включая имя записной книжки, озеро по умолчанию, сведения о рабочей области, если это запуск конвейера и т. д.

mssparkutils.runtime.context

Известная проблема

При использовании версии среды выполнения выше 1.2 и запуска mssparkutils.help(), перечисленные API fabricClient, хранилища и рабочей области пока не поддерживаются, но будут доступны в будущем.

Связанный контент

• Управление библиотеками