Интерактивные обработки данных с помощью Apache Spark

Обработка данных является важным аспектом проектов машинного обучения. Из этой статьи вы узнаете, как выполнять интерактивные обработки данных, выполнив записные книжки Машинного обучения Azure на бессерверных вычислительных ресурсах Apache Spark, поддерживаемых Azure Synapse.

В этой статье объясняется, как подключить и настроить бессерверные вычисления Spark. В этой статье показано, как использовать бессерверные вычисления Spark для доступа к данным и их обработке из нескольких источников.

Предварительные требования

Разрешения владельца или назначения ролей в подписке Azure. Вы можете создать бесплатную учетную запись Azure.
Рабочая область Машинного обучения Azure. Дополнительные сведения см. в разделе Создание ресурсов рабочей области.
Набор данных titanic.csv, загруженный в общий доступ по умолчанию в рабочей области.
Учетная запись хранения Azure Data Lake Storage 2-го поколения. Дополнительные сведения см. в статье "Создание учетной записи хранения Azure Data Lake Storage 2-го поколения".
Предоставленные следующие назначения ролей :
- Для доступа к учетной записи Azure Storage необходимы роли Contributor и Storage Blob Data Contributor.
- Для доступа к секретам в Azure Key Vault необходима роль Пользователь секретов Key Vault в хранилище ключей.

Дополнительные сведения можно найти здесь

Использование бессерверных вычислений Spark в сеансах записной книжки

Использование бессерверных вычислений Spark — самый простой способ доступа к кластеру Spark для интерактивного обработки данных. Полностью управляемые бессерверные вычисления Spark, подключенные к пулу Synapse Spark , доступны непосредственно в записных книжках машинного обучения Azure.

Чтобы использовать любой из следующих источников и методов доступа к данным, подключите бессерверные вычисления Spark, выбрав Azure Machine Learning Бессерверные вычисления Spark>Бессерверные вычисления Spark — доступны рядом с Вычисления в верхней части страницы файла или записной книжки. Для подключения вычислительных ресурсов к сеансу может потребоваться несколько минут.

Настройка бессерверного сеанса Spark

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

Выберите "Настройка сеанса в левом верхнем углу" на странице файла или записной книжки.
На экране "Настройка сеанса" измените любой из следующих параметров:
- На панели вычислений :
  - Измените размер компьютера, выбрав другой размер в раскрывающемся меню в разделе "Размер узла".
  - Выберите, следует ли динамически выделять исполнителей.
  - Выберите количество исполнителей для сеанса Spark.
  - Выберите другой размер исполнителя , если он доступен в раскрывающемся меню.
- В панели параметров:
  - При наличии измените версию Apache Spark на версию, отличную от версии 3.5.
    
    Это важно
    
    Поддержка среды выполнения Azure Synapse для Apache Spark 3.4 заканчивается 31 марта 2026 г. Перейдите в Apache Spark 3.5 для продолжения поддержки. Дополнительные сведения см. в средах выполнения Azure Synapse.
  - Измените значение времени ожидания сеанса в минутах на более высокое число, чтобы предотвратить время ожидания сеанса.
  - В разделе "Параметры конфигурации" добавьте параметры имени или значения свойства , чтобы настроить сеанс по мере необходимости.
    
    Совет
    
    Если вы используете пакеты Conda уровня сеанса, добавление spark.hadoop.aml.enable_cache свойства конфигурации со значением true может улучшить время холодного запуска сеанса Spark. Холодный запуск сеанса с пакетами Conda, относящимися к уровню сеанса, обычно занимает от 10 до 15 минут в первый раз. При последующем холодном запуске сеанса, при установленной переменной конфигурации true, обычно уходит три–пять минут.
- В панели пакетов Python:
  - Чтобы использовать файл Conda для настройки сеанса, выберите "Отправить файл conda". Рядом с Select conda file выберите Обзор, затем перейдите к соответствующему файлу Conda YAML на вашем компьютере, чтобы загрузить его.
  - Чтобы использовать настраиваемую среду, выберите пользовательскую среду и выберите настраиваемую среду в разделе "Тип среды". Дополнительные сведения см. в разделе "Управление средами программного обеспечения".
Нажмите кнопку "Применить" , чтобы применить все конфигурации.

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

Импорт и обработка данных из Azure Data Lake Storage

Чтобы получить и обрабатывать данные, хранящиеся в учетных записях Azure Data Lake Storage, используйте URI-протокол с передачей удостоверения пользователя или доступом на основе учетной записи службы. Сквозное руководство пользователя не требует дополнительной конфигурации.

Чтобы использовать любой из методов, удостоверение пользователя или учетная запись службы должны иметь роль участника и роль Участника данных больших двоичных объектов хранилища в учетной записи Azure Data Lake Storage.

Для сквозной передачи идентификации пользователя выполните следующий пример кода для обработки данных, чтобы использовать URI данных в формате abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> с pyspark.pandas. Замените <STORAGE_ACCOUNT_NAME> заполнитель именем учетной записи Azure Data Lake Storage и <FILE_SYSTEM_NAME> именем контейнера данных.

import pyspark.pandas as pd

df = pd.read_csv(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv",
    index_col="PassengerId",
)
df.fillna(
    value={"Cabin": "None"}, inplace=True
)  # Fill Cabin column with value "None" if missing
df.dropna(inplace=True)  # Drop the rows which still have any missing value
df.to_csv(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled",
    index_col="PassengerId",
)

Использование служебного принципала

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

Создайте служебный принципал и назначьте ему необходимые роли участника данных BLOB-объектов хранилища и пользователя секретов Key Vault.
Получите идентификатор арендатора сервисного принципала, идентификатор клиента и секрет клиента из регистрации приложения и создайте секреты Azure Key Vault для этих значений.

Задайте идентификатор клиента субъекта-службы, идентификатор клиента и секрет клиента, добавив следующие пары свойств и значений в конфигурацию сеанса. Замените <STORAGE_ACCOUNT_NAME> именем учетной записи хранения и <TENANT_ID> идентификатором клиента субъекта-службы.

Название свойства	Ценность
`fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`	Значение идентификатора приложения (клиента)
`fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`	`https://login.microsoftonline.com/<TENANT_ID>/oauth2/token`
`fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`	Значение клиентского секрета

Выполните следующий код. Вызов get_secret() в коде зависит от имени Key Vault и имен секретов Key Vault, созданных для идентификатора арендатора принципала службы, идентификатора клиента и секрета клиента.

from pyspark.sql import SparkSession

sc = SparkSession.builder.getOrCreate()
token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary

# Set up service principal tenant ID, client ID, and secret from Azure Key Vault
client_id = token_library.getSecret("<KEY_VAULT_NAME>", "<CLIENT_ID_SECRET_NAME>")
tenant_id = token_library.getSecret("<KEY_VAULT_NAME>", "<TENANT_ID_SECRET_NAME>")
client_secret = token_library.getSecret("<KEY_VAULT_NAME>", "<CLIENT_SECRET_NAME>")

# Set up a service principal that has access to the data
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.auth.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net", "OAuth"
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth.provider.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    client_id,
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    client_secret,
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    "https://login.microsoftonline.com/" + tenant_id + "/oauth2/token",
)

Импорт и перебор данных titanic.csv с помощью URI данных в abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> формате, как показано в примере кода. Замените <STORAGE_ACCOUNT_NAME> заполнитель именем учетной записи Azure Data Lake Storage и <FILE_SYSTEM_NAME> именем контейнера данных.

import pyspark.pandas as pd

df = pd.read_csv(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv",
    index_col="PassengerId",
)
df.fillna(
    value={"Cabin": "None"}, inplace=True
)  # Fill Cabin column with value "None" if missing
df.dropna(inplace=True)  # Drop the rows which still have any missing value
df.to_csv(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled",
    index_col="PassengerId",
)

Импорт и обработка данных из хранилища BLOB-объектов Azure

Доступ к данным хранилища BLOB-объектов Azure можно получить с помощью ключа доступа к учетной записи хранения или маркера подписанной строки доступа (SAS). Сохраните учетные данные в Azure Key Vault в качестве секрета и задайте его в качестве свойства в конфигурации сеанса Spark.

Выполните один из следующих фрагментов кода. Вызовы get_secret() в фрагментах кода требуют имени хранилища ключей и имен секретов, созданных для ключа доступа к учетной записи хранения BLOB-объектов Azure или маркера SAS.
- Чтобы настроить ключ доступа к учетной записи хранения, задайте fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net свойство, как показано в следующем фрагменте кода:
```
from pyspark.sql import SparkSession

sc = SparkSession.builder.getOrCreate()
token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
access_key = token_library.getSecret("<KEY_VAULT_NAME>", "<ACCESS_KEY_SECRET_NAME>")
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net", access_key
)
```
- Чтобы настроить маркер SAS, задайте fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net свойство, как показано в следующем фрагменте кода:
```
from pyspark.sql import SparkSession

sc = SparkSession.builder.getOrCreate()
token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
sas_token = token_library.getSecret("<KEY_VAULT_NAME>", "<SAS_TOKEN_SECRET_NAME>")
sc._jsc.hadoopConfiguration().set(
    "fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net",
    sas_token,
)
```

Выполните следующий код обработки данных с помощью URI данных, отформатированного как wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/<PATH_TO_DATA>.

import pyspark.pandas as pd

df = pd.read_csv(
    "wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/titanic.csv",
    index_col="PassengerId",
)
df.fillna(
    value={"Cabin": "None"}, inplace=True
)  # Fill Cabin column with value "None" if missing
df.dropna(inplace=True)  # Drop the rows which still have any missing value
df.to_csv(
    "wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/wrangled",
    index_col="PassengerId",
)

Импорт и обработка данных из хранилища данных Машинного обучения Azure

Чтобы получить доступ к данным из хранилища данных Машинного обучения Azure, необходимо определить путь к данным в хранилище данных с форматомazureml://datastores/<DATASTORE_NAME>/paths/<PATH_TO_DATA> URI.

Выполните следующий пример кода для чтения и обработки данных titanic.csv из хранилища данных Azure Machine Learning, используя azureml:// URI хранилища данных и pyspark.pandas.

import pyspark.pandas as pd

df = pd.read_csv(
    "azureml://datastores/<DATASTORE_NAME>/paths/data/titanic.csv",
    index_col="PassengerId",
)
df.fillna(
    value={"Cabin": "None"}, inplace=True
)  # Fill Cabin column with value "None" if missing
df.dropna(inplace=True)  # Drop the rows which still have any missing value
df.to_csv(
    "azureml://datastores/<DATASTORE_NAME>/paths/data/wrangled",
    index_col="PassengerId",
)

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

В следующей таблице приведены механизмы проверки подлинности для доступа к данным в хранилищах данных Машинного обучения Azure:

Тип учётной записи хранилища	Доступ к данным без учетных данных	Механизм доступа к данным	Назначения ролей
Azure Blob	нет	Ключ доступа или маркер SAS	Нет необходимых назначений ролей.
Azure Blob	Да	Сквозная идентификация пользователя*	Идентификация пользователя должна иметь соответствующие назначения ролей в учетной записи хранения BLOB-объектов Azure.
Azure Data Lake Storage	нет	Субъект-служба	Принципал службы должен иметь соответствующие назначения ролей в аккаунте хранения Azure Data Lake Storage.
Azure Data Lake Storage	Да	Сквозная передача идентификации пользователя	Идентификатор пользователя должен иметь соответствующие назначения ролей в учетной записи хранения Azure Data Lake Storage.

* Сквозная идентификация пользователя работает для хранилищ данных без учетных данных, указывающих на учетные записи хранения BLOB-объектов Azure, только если мягкое удаление не включено.

В Студии машинного обучения Azure общая папка рабочей области по умолчанию — это дерево каталогов на вкладке "Файлы " в записных книжках. Код записной книжки может напрямую получить доступ к файлам, хранящимся в этой общей папке, используя протокол file:// и абсолютный путь к файлу без какой-либо другой настройки. Файловое хранилище по умолчанию подключено к бессерверным вычислительным ресурсам Spark и смонтировано в подключённых пулах Synapse Spark.

Следующий фрагмент кода обращается к данным и выполняет обработку данных из файла titanic.csv , хранящегося в папке данных непосредственно под именем пользователя в общей папке по умолчанию. Замените <USER> заполнитель именем пользователя.

import os
import pyspark.pandas as pd

abspath = os.path.abspath(".")
file = "file://" + abspath + "/Users/<USER>/data/titanic.csv"
print(file)
df = pd.read_csv(file, index_col="PassengerId")
df.fillna(value={"Cabin" : "None"}, inplace=True) # Fill Cabin column with value "None" if missing
df.dropna(inplace=True) # Drop the rows which still have any missing value
output_path = "file://" + abspath + "/Users/<USER>/data/wrangled"
df.to_csv(output_path, index_col="PassengerId")

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-03-11