Учебное руководство: запуск пакетного задания с помощью Data Factory, Batch Explorer, Storage Explorer и Python.

В этом руководстве описывается, как создать и запустить конвейер в Azure Data Factory, который выполняет рабочую нагрузку в Azure Batch. Скрипт Python запускается на узлах пакетной обработки, чтобы получить входные данные, разделенные запятыми (CSV), из контейнера Blob-хранилища Azure, обрабатывать данные и записать выходные данные в другой контейнер хранилища. Пакетный обозреватель Batch Explorer используется для создания пула и узлов пакетной службы, а Azure Storage Explorer — для работы с контейнерами и файлами хранилища.

В этом руководстве описано следующее:

Требования

• Учетная запись Azure с активной подпиской. Если ее нет, создайте бесплатную учетную запись.
• Учетная запись службы Batch со связанной учетной записью Azure Storage. Учетные записи можно создать с помощью любого из следующих методов: Azure портал | Azure CLI | Bicep | ARM template | Terraform.
• Экземпляр фабрики данных. Чтобы создать фабрику данных, следуйте инструкциям в статье "Создание фабрики данных".
• Batch Explorer скачан и установлен.
• Storage Explorer загружен и установлен.
• Python 3.8 или более поздней версии с пакетом azure-storage-blob , установленным с помощью pip.
• Набор данных iris.csv, загруженный с GitHub.

Используйте Batch Explorer для создания пула и узлов пакетной обработки

Используйте Batch Explorer для создания пула вычислительных узлов для выполнения рабочей нагрузки.

1. Войдите в Batch Explorer с помощью учетных данных Azure.

2. Выберите учетную запись Batch.

3. Выберите пулы на левой боковой панели и щелкните + значок, чтобы добавить пул.

   

4. Заполните форму добавления пула в форму учетной записи следующим образом:

   • В поле "Идентификатор" введите custom-activity-pool.
   • В разделе "Выделенные узлы" введите 2.
   • Чтобы выбрать конфигурацию операционной системы, перейдите на вкладку "Обработка и анализ данных", а затем выберите Dsvm Win 2019.
   • Для выбора размера виртуальной машины выберите Standard_F2s_v2.
   • Для задачи 'Начать' выберите Добавить задачу начала. На начальном экране задачи в командной строке введите cmd /c "pip install azure-storage-blob pandas"и нажмите кнопку "Выбрать". Эта команда устанавливает azure-storage-blob пакет на каждом узле по мере запуска.

5. Выберите Сохранить и закрыть.

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

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

1. Войдите в Обозреватель службы хранилища с помощью учетных данных Azure.
2. На левой панели найдите и разверните учетную запись хранилища, связанную с вашим Batch-аккаунтом.
3. Щелкните правой кнопкой мыши контейнеры BLOB-объектов и выберите «Создать контейнер BLOB-объектов», или выберите «Создать контейнер BLOB-объектов» в разделе Действия в нижней части боковой панели.
4. Введите текст в поле ввода.
5. Создайте другой контейнер объектов с именем output.
6. Выберите входнойконтейнер и нажмите кнопку >" в правой области.
7. На экране "Отправить файлы" в разделе "Выбранные файлы" выберите многоточие ... рядом с полем записи.
8. Перейдите к расположению скачаемого файла iris.csv , выберите "Открыть" и нажмите кнопку "Отправить".

Разработка скрипта Python

Следующий скрипт Python загружает файл набора данных iris.csv из контейнера ввода в Обозревателе хранилища, обрабатывает данные и сохраняет результаты в контейнере выхода.

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

1. В портале Azure найдите и выберите имя учетной записи хранения, связанной с вашей учетной записью Batch.
2. На странице учетной записи хранения выберите ключи доступа в левой области навигации в разделе "Безопасность и сеть".
3. В разделе key1 выберите "Показать рядом со строкой подключения", а затем щелкните значок "Копировать", чтобы скопировать строка подключения.

Вставьте строку подключения в следующий скрипт, заменив этот <storage-account-connection-string> заполнитель. Сохраните скрипт в виде файла с именем main.py.

# Load libraries
# from azure.storage.blob import BlobClient
from azure.storage.blob import BlobServiceClient
import pandas as pd
import io

# Define parameters
connectionString = "<storage-account-connection-string>"
containerName = "output"
outputBlobName	= "iris_setosa.csv"

# Establish connection with the blob storage account
blob = BlobClient.from_connection_string(conn_str=connectionString, container_name=containerName, blob_name=outputBlobName)

# Initialize the BlobServiceClient (This initializes a connection to the Azure Blob Storage, downloads the content of the 'iris.csv' file, and then loads it into a Pandas DataFrame for further processing.)
blob_service_client = BlobServiceClient.from_connection_string(conn_str=connectionString)
blob_client = blob_service_client.get_blob_client(container_name=containerName, blob_name=outputBlobName)

# Download the blob content
blob_data = blob_client.download_blob().readall()

# Load iris dataset from the task node
# df = pd.read_csv("iris.csv")
df = pd.read_csv(io.BytesIO(blob_data))

# Take a subset of the records
df = df[df['Species'] == "setosa"]

# Save the subset of the iris dataframe locally in the task node
df.to_csv(outputBlobName, index = False)

with open(outputBlobName, "rb") as data:
    blob.upload_blob(data, overwrite=True)

Ознакомьтесь с документацией по хранилищу blob-объектов Azure для получения дополнительной информации о его использовании.

Запустите скрипт локально для тестирования и проверки функциональности.

python main.py

Скрипт должен создать выходной файл с именем iris_setosa.csv , который содержит только записи данных с типом = setosa. Убедившись, что он работает правильно, загрузите файл скрипта main.py в входной контейнер Обозревателя службы хранилища.

Настройка конвейера в Data Factory

Создайте и проверьте конвейер фабрики данных, использующий скрипт Python.

Получение данных об учетной записи

Конвейер Фабрики данных использует имена учетных записей Batch и хранения, значения ключей учетной записи и конечную точку учетной записи Batch. Чтобы получить эти сведения из портал Azure, выполните указанные ниже действия.

1. На панели поиска Azure найдите и выберите имя учетной записи Batch.

2. На странице учетной записи Batch выберите Ключи в области навигации слева.

3. На странице "Ключи" скопируйте следующие значения:

   • Учетная запись пакетной обработки
   • Конечная точка учетной записи
   • Первичный ключ доступа
   • Имя учетной записи хранения
   • Key1

Создание и запуск конвейера

1. Если Azure Data Factory Studio еще не запущена, выберите Запустить студию на странице фабрики данных в портале Azure.

2. В Студии фабрики данных щелкните значок карандаша "Автор " в области навигации слева.

3. В разделе Ресурсы фабрики щелкните значок +, а затем выберите Конвейер.

4. В области свойств справа измените имя конвейера на Запуск Python.

   

5. В панели действий разверните Пакетную службу и перетащите пользовательскую деятельность в область конструктора конвейеров.

6. Под холстом конструктора на вкладке "Общие " введите testPipeline в разделе "Имя".

   

7. Выберите вкладку Azure Batch и затем выберите Создать.

8. Заполните форму новой связанной службы следующим образом:

   • Имя: введите имя связанной службы, например AzureBatch1.
   • Ключ доступа: введите основной ключ доступа, скопированный из вашей учетной записи службы Batch.
   • Имя учетной записи: введите имя учетной записи Batch.
   • URL пакетной службы: введите адрес конечной точки, скопированный из вашей учетной записи пакетной службы, например https://batchdotnet.eastus.batch.azure.com.
   • Имя пула: введите custom-activity-pool, пул, который вы создали в Пакетном обозревателе.
   • Имя службы, связанной с учетной записью хранения: выберите Создать. На следующем экране введите имя связанной службы хранилища, например AzureBlobStorage1, выберите подписку Azure и связанную учетную запись хранения, а затем нажмите кнопку "Создать".

9. В нижней части экрана Создание новой связанной службы, выберите Проверить подключение. Когда подключение выполнено успешно, нажмите кнопку "Создать".

   

10. Выберите вкладку "Параметры" и введите или выберите следующие параметры:

    • Команда: Нажмите ВВОД cmd /C python main.py.
    • Связанная служба ресурсов: выберите созданную службу связанного хранилища, например AzureBlobStorage1, и проверьте подключение, чтобы убедиться, что оно выполнено успешно.
    • Путь к папке: щелкните значок папки, а затем выберите входной контейнер и нажмите кнопку "ОК". Файлы из этой папки скачиваются из контейнера на узлы пула перед запуском скрипта Python.

    

11. Выберите «Проверить» на панели инструментов конвейера, чтобы выполнить валидацию конвейера.

12. Выберите "Отладка", чтобы проверить конвейер и убедиться, что он работает правильно.

13. Выберите "Опубликовать все ", чтобы опубликовать конвейер.

14. Выберите " Добавить триггер", а затем нажмите кнопку "Триггер" , чтобы запустить конвейер или создать или изменить , чтобы запланировать его.

    

Просмотр файлов журналов с помощью Batch Explorer

Если при запуске конвейера возникают предупреждения или ошибки, можно использовать Batch Explorer для просмотра stdout.txt и stderr.txt выходных файлов для получения дополнительных сведений.

1. В пакетном обозревателе выберите задания на левой боковой панели.
2. Выберите задание adfv2-custom-activity-pool.
3. Выберите задачу, у которой был код выхода из строя.
4. Просмотрите stdout.txt и stderr.txt файлы для изучения и диагностики проблемы.

Очистка ресурсов

Учетные записи пакетов, задания и задачи бесплатны, но вычислительные узлы несут расходы, даже когда они не работают. Рекомендуется выделить пулы узлов только по мере необходимости и удалить пулы, когда вы закончите работу с ними. Удаление пулов приводит к удалению всех выходных данных задач на узлах, а также самих узлов.

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

Следующие шаги

В этом руководстве вы узнали, как использовать скрипт Python с Batch Explorer, Storage Explorer и Data Factory для запуска рабочей нагрузки в пакетной обработке. Дополнительные сведения о фабрике данных см. в статье "Что такое Фабрика данных Azure?"