Poradnik: Uruchom zadanie Batch za pomocą Data Factory, Batch Explorer, Eksplorator Storage i Python

Ten samouczek pokaże Ci, jak stworzyć i uruchomić potok usługi Azure Data Factory, który realizuje zadanie obliczeniowe w ramach usługi Azure Batch. Skrypt języka Python jest uruchamiany w węzłach usługi Batch w celu pobrania danych wejściowych wartości rozdzielanych przecinkami (CSV) z kontenera usługi Azure Blob Storage, manipulowania danymi i zapisywania danych wyjściowych w innym kontenerze magazynu. Za pomocą programu Batch Explorer można utworzyć pulę i węzły usługi Batch, a za pomocą Eksploratora usługi Azure Storage pracować z kontenerami i plikami magazynu.

Z tego samouczka dowiesz się, jak wykonywać następujące czynności:

• Użyj Batch Explorer do utworzenia puli i węzłów Batch.
• Użyj Eksploratora Storage do tworzenia kontenerów i przesyłania plików wejściowych.
• Tworzenie skryptu języka Python w celu manipulowania danymi wejściowymi i generowania danych wyjściowych.
• Utwórz potok usługi Data Factory, który uruchamia obciążenie usługi Batch.
• Użyj programu Batch Explorer, aby przyjrzeć się plikom dziennika danych wyjściowych.

Wymagania wstępne

• Konto platformy Azure z aktywną subskrypcją. Jeśli jej nie masz, utwórz bezpłatne konto.
• Konto usługi Batch z połączonym kontem usługi Azure Storage. Konta można utworzyć przy użyciu dowolnej z następujących metod: portal Azure | interfejs wiersza polecenia Azure | Bicep | szablon ARM | Terraform.
• Instancja Data Factory. Aby utworzyć fabrykę danych, postępuj zgodnie z instrukcjami w temacie Tworzenie fabryki danych.
• Program Batch Explorer został pobrany i zainstalowany.
• Storage Explorer pobrany i zainstalowany.
• Środowisko Python w wersji 3.8 lub nowszej z zainstalowanym pakietem azure-storage-blob przy użyciu polecenia pip.
• Wejściowy zestaw danych iris.csv pobrany z usługi GitHub.

Tworzenie puli i węzłów usługi Batch za pomocą programu Batch Explorer

Użyj programu Batch Explorer, aby utworzyć pulę węzłów obliczeniowych w celu uruchomienia obciążenia.

1. Zaloguj się do usługi Batch Explorer przy użyciu poświadczeń platformy Azure.

2. Wybierz konto usługi Batch.

3. Wybierz pozycję Pule na lewym pasku bocznym, a następnie wybierz ikonę + , aby dodać pulę.

   

4. Wypełnij formularz Dodawanie puli do konta w następujący sposób:

   • Pod ID, wprowadź custom-activity-pool.
   • W obszarze Węzły dedykowane wprowadź wartość 2.
   • Dla opcji Wybierz konfigurację systemu operacyjnego wybierz kartę Data Science, a następnie wybierz Dsvm Win 2019.
   • W obszarze Wybierz rozmiar maszyny wirtualnej wybierz pozycję Standard_F2s_v2.
   • Dla Rozpocznij zadanie, wybierz Dodaj zadanie początkowe. Na ekranie zadania uruchamiania w obszarze Wiersz polecenia wprowadź cmd /c "pip install azure-storage-blob pandas", a następnie wybierz pozycję Wybierz. To polecenie instaluje pakiet azure-storage-blob na każdym z węzłów w momencie uruchamiania.

5. Wybierz pozycję Zapisz i zamknij.

Tworzenie kontenerów obiektów blob za pomocą Eksplorator usługi Storage

Użyj Eksploratora magazynu, aby utworzyć kontenery obiektów blob do przechowywania plików wejściowych i wyjściowych, a następnie prześlij pliki wejściowe.

1. Zaloguj się do Eksploratora magazynu, używając poświadczeń Azure.
2. Na pasku bocznym po lewej stronie znajdź i rozwiń konto magazynowe połączone z kontem Batch.
3. Kliknij prawym przyciskiem myszy Kontenery obiektów blob i wybierz Utwórz kontener obiektów blob, lub wybierz Utwórz kontener obiektów blob z Akcji w dolnej części paska bocznego.
4. Wprowadź dane wejściowe w polu wprowadzania.
5. Utwórz kolejny kontener blobów o nazwie output.
6. Wybierz kontener wejściowy, a następnie wybierz pozycję Przekaż>pliki w okienku po prawej stronie.
7. Na ekranie Przekazywanie plików w obszarze Wybrane pliki wybierz wielokropek ... obok pola wprowadzania.
8. Przejdź do lokalizacji pobranego pliku iris.csv , wybierz pozycję Otwórz, a następnie wybierz pozycję Przekaż.

Tworzenie skryptu języka Python

Poniższy skrypt Pythona ładuje plik zestawu danych iris.csv z kontenera input w Eksploratorze pamięci masowej, manipuluje danymi i zapisuje wyniki do kontenera output.

Skrypt musi używać ciągu połączenia dla konta usługi Azure Storage połączonego z kontem usługi Batch. Aby uzyskać parametry połączenia:

1. W witrynie Azure Portal wyszukaj i wybierz nazwę konta magazynu połączonego z kontem usługi Batch.
2. Na stronie konta magazynowego wybierz pozycję Klucze dostępu z lewego panelu nawigacji w sekcji Zabezpieczenia + sieć.
3. W obszarze key1 wybierz pozycję Pokaż obok pozycji Parametry połączenia, a następnie wybierz ikonę Kopiuj, aby skopiować parametry połączenia.

Wklej parametry połączenia do następującego skryptu, zastępując symbol zastępczy <storage-account-connection-string>. Zapisz skrypt jako plik o nazwie main.py.

Ważne

Udostępnianie kluczy kont w źródle aplikacji nie jest zalecane w przypadku użycia produkcyjnego. Należy ograniczyć dostęp do poświadczeń i odwołać się do nich w kodzie przy użyciu zmiennych lub pliku konfiguracji. Najlepiej przechowywać klucze kont magazynu i usługi Batch w usłudze Azure Key Vault.

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

Aby uzyskać więcej informacji na temat pracy z usługą Azure Blob Storage, zapoznaj się z dokumentacją usługi Azure Blob Storage.

Uruchom skrypt lokalnie, aby przetestować i zweryfikować funkcjonalność.

python main.py

Skrypt powinien utworzyć plik wyjściowy o nazwie iris_setosa.csv zawierający tylko rekordy danych, które mają gatunek = setosa. Po zweryfikowaniu, że działa prawidłowo, prześlij plik skryptu main.py do kontenera input w Eksploratorze usługi Storage.

Skonfiguruj potok Data Factory

Utwórz i zweryfikuj potok usługi Data Factory, który używa skryptu języka Python.

Pobieranie informacji o koncie

Potok w usłudze Data Factory wykorzystuje nazwy kont Batch i kont magazynu, wartości kluczy kont oraz punkt końcowy konta Batch. Aby uzyskać te informacje z witryny Azure Portal:

1. Na pasku usługi Azure Search wyszukaj i wybierz nazwę konta usługi Batch.

2. Na stronie swojego konta Batch wybierz Klucze z lewego paska nawigacji.

3. Na stronie Klucze skopiuj następujące wartości:

   • Konto usługi Batch
   • Punkt końcowy konta
   • Podstawowy klucz dostępu
   • Nazwa konta magazynu
   • Klucz1

Utwórz i uruchom potok

1. Jeśli program Azure Data Factory Studio nie jest jeszcze uruchomiony, wybierz pozycję Uruchom studio na stronie usługi Data Factory w witrynie Azure Portal.

2. W narzędziu Data Factory Studio wybierz ikonę ołówka Author w lewym obszarze nawigacji.

3. W obszarze Zasoby fabryki wybierz ikonę + , a następnie wybierz pozycję Potok.

4. W okienku Właściwości po prawej stronie zmień nazwę pipeline'u na Run Python.

   

5. W okienku Działania rozwiń Usługę wsadową i przeciągnij Działanie Niestandardowe na powierzchnię projektanta potoku.

6. Poniżej kanwy projektanta na karcie Ogólne wprowadź testPipeline w polu Nazwa.

   

7. Wybierz kartę Azure Batch , a następnie wybierz pozycję Nowy.

8. Wypełnij formularz Nowa połączona usługa w następujący sposób:

   • Nazwa: wprowadź nazwę połączonej usługi, na przykład AzureBatch1.
   • Klucz dostępu: wprowadź podstawowy klucz dostępu skopiowany z konta usługi Batch.
   • Nazwa konta: wprowadź nazwę konta usługi Batch.
   • Adres URL usługi Batch: wprowadź punkt końcowy konta skopiowany z konta usługi Batch, na przykład https://batchdotnet.eastus.batch.azure.com.
   • Nazwa puli: Wprowadź custom-activity-pool jako nazwę puli, którą utworzyłeś w programie Batch Explorer.
   • Nazwa połączonego konta magazynowego: Wybierz Nowe. Na następnym ekranie wprowadź Nazwę połączonej usługi magazynu, taką jak AzureBlobStorage1, wybierz subskrypcję Azure i połączone konto magazynu, a następnie wybierz pozycję Utwórz.

9. Na dole ekranu Nowa połączona usługa Batch wybierz Testuj połączenie. Po pomyślnym nawiązaniu połączenia wybierz pozycję Utwórz.

   

10. Wybierz kartę Ustawienia , a następnie wprowadź lub wybierz następujące ustawienia:

    • Polecenie: wprowadź cmd /C python main.py.
    • Usługa połączenia zasobów: wybierz utworzoną połączoną usługę magazynu, taką jak AzureBlobStorage1, i przetestuj połączenie, aby upewnić się, że działa pomyślnie.
    • Ścieżka folderu: wybierz ikonę folderu, a następnie wybierz kontener wejściowy i wybierz przycisk OK. Pliki z tego folderu są pobierane z kontenera do węzłów puli przed uruchomieniem skryptu języka Python.

    

11. Wybierz Weryfikuj na pasku narzędzi rurociągu, aby zweryfikować rurociąg.

12. Wybierz Debuguj, aby przetestować pipeline i upewnić się, że działa prawidłowo.

13. Wybierz Opublikuj wszystko, aby opublikować potok.

14. Wybierz Dodaj wyzwalacz, a następnie wybierz Wyzwól teraz, aby uruchomić potok, lub Nowy/Edytuj aby go zaplanować.

    

Wyświetlanie plików dziennika za pomocą programu Batch Explorer

Jeśli uruchomienie potoku generuje ostrzeżenia lub błędy, możesz użyć programu Batch Explorer, aby zapoznać się z plikami wyjściowymi stdout.txt i stderr.txt, aby uzyskać więcej informacji.

1. W programie Batch Explorer wybierz pozycję Zadania na lewym pasku bocznym.
2. Wybierz zadanie adfv2-custom-activity-pool.
3. Wybierz zadanie, które miało kod błędu zakończenia.
4. Wyświetl pliki stdout.txt i stderr.txt, aby zbadać i zdiagnozować problem.

Czyszczenie zasobów

Konta Batch, zadania i operacje są bezpłatne, ale węzły obliczeniowe generują opłaty nawet wtedy, gdy nie są uruchamiane zadania. Najlepiej przydzielać pule węzłów tylko w razie potrzeby i usuwać je po skończeniu pracy. Usunięcie pul powoduje usunięcie wszystkich danych wyjściowych zadań na węzłach oraz usunięcie samych węzłów.

Pliki wejściowe i wyjściowe pozostają na koncie magazynu i mogą generować koszty. Gdy pliki nie są już potrzebne, możesz usunąć pliki lub kontenery. Jeśli nie potrzebujesz już konta usługi Batch ani połączonego konta magazynu, możesz je usunąć.

Następne kroki

W tym samouczku nauczyłeś się, jak używać skryptu języka Python z Batch Explorer, Storage Explorer i Data Factory do uruchamiania zadania Batch. Aby uzyskać więcej informacji na temat usługi Data Factory, zobacz Co to jest usługa Azure Data Factory?