Esercitazione: Eseguire un processo Batch tramite Data Factory con Batch Explorer, Storage Explorer e Python

Questa esercitazione illustra come creare ed eseguire una pipeline di Azure Data Factory che esegue un carico di lavoro di Azure Batch. Uno script Python viene eseguito sui nodi Batch per ottenere un input di valori separati da virgole (CSV) da un contenitore di Archiviazione BLOB di Azure, modificare i dati e scrivere l'output in un contenitore di archiviazione diverso. Si usa Batch Explorer per creare un pool e nodi di Batch e Archiviazione di Azure Explorer per usare i contenitori e i file di archiviazione.

In questa esercitazione apprenderai a:

• Usare Batch Explorer per creare un pool e nodi di Batch.
• Usare Storage Explorer per creare contenitori di archiviazione e caricare file di input.
• Sviluppare uno script Python per modificare i dati di input e produrre output.
• Creare una pipeline di Data Factory che esegue il carico di lavoro batch.
• Usare Batch Explorer per esaminare i file di log di output.

Prerequisiti

• Un account Azure con una sottoscrizione attiva. Se non se ne ha una, creare un account gratuito.
• Un account Batch con un account Archiviazione di Azure collegato. È possibile creare gli account usando uno dei metodi seguenti: portale di Azure | Portale bicep arm | di Azure | bicep. |
• Istanza di Data Factory. Per creare la data factory, seguire le istruzioni riportate in Creare una data factory.
• Batch Explorer scaricato e installato.
• Storage Explorer scaricato e installato.
• Python 3.8 o versione successiva, con il pacchetto azure-storage-blob installato usando pip.
• Set di dati di input iris.csv scaricato da GitHub.

Usare Batch Explorer per creare un pool e nodi di Batch

Usare Batch Explorer per creare un pool di nodi di calcolo per eseguire il carico di lavoro.

1. Accedere a Batch Explorer con le credenziali di Azure.

2. Selezionare l'account Batch.

3. Selezionare Pool sulla barra laterale sinistra e quindi selezionare l'icona + per aggiungere un pool.

   

4. Completare il modulo Aggiungi un pool all'account come indicato di seguito:

   • In ID immettere custom-activity-pool.
   • In Nodi dedicati immettere 2.
   • Per Selezionare una configurazione del sistema operativo, selezionare la scheda Analisi scientifica dei dati e quindi selezionare Dsvm Win 2019.
   • In Scegliere le dimensioni di una macchina virtuale selezionare Standard_F2s_v2.
   • Per Avvia attività selezionare Aggiungi un'attività di avvio. Nella schermata attività di avvio, in Riga di comando immettere cmd /c "pip install azure-storage-blob pandas"e quindi selezionare Seleziona. Questo comando installa il azure-storage-blob pacchetto in ogni nodo all'avvio.

5. Seleziona Salva e chiudi.

Usare Storage Explorer per creare contenitori BLOB

Usare Storage Explorer per creare contenitori BLOB per archiviare i file di input e di output e quindi caricare i file di input.

1. Accedere a Storage Explorer con le credenziali di Azure.
2. Nella barra laterale sinistra individuare ed espandere l'account di archiviazione collegato all'account Batch.
3. Fare clic con il pulsante destro del mouse su Contenitori BLOB e scegliere Crea contenitore BLOB oppure selezionare Crea contenitore BLOB da Azioni nella parte inferiore della barra laterale.
4. Immettere l'input nel campo di immissione.
5. Creare un altro contenitore BLOB denominato output.
6. Selezionare il contenitore di input e quindi caricare> i file nel riquadro destro.
7. Nella schermata Carica file , in File selezionati, selezionare i puntini di sospensione ... accanto al campo voce.
8. Passare al percorso del file iris.csv scaricato, selezionare Apri e quindi selezionare Carica.

Sviluppare uno script Python

Lo script Python seguente carica il file del set di dati iris.csv dal contenitore di input di Storage Explorer, modifica i dati e salva i risultati nel contenitore di output.

Lo script deve usare il stringa di connessione per l'account Archiviazione di Azure collegato all'account Batch. Per ottenere il stringa di connessione:

1. Nella portale di Azure cercare e selezionare il nome dell'account di archiviazione collegato all'account Batch.
2. Nella pagina dell'account di archiviazione selezionare Chiavi di accesso nel riquadro di spostamento a sinistra in Sicurezza e rete.
3. In key1 selezionare Mostra accanto a Stringa di connessione e quindi selezionare l'icona Copia per copiare il stringa di connessione.

Incollare il stringa di connessione nello script seguente, sostituendo il <storage-account-connection-string> segnaposto. Salvare lo script come file denominato main.py.

Importante

L'esposizione delle chiavi dell'account nell'origine dell'app non è consigliata per l'utilizzo di produzione. È consigliabile limitare l'accesso alle credenziali e farvi riferimento nel codice usando variabili o un file di configurazione. È consigliabile archiviare le chiavi dell'account batch e dell'account di archiviazione in 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)

Per altre informazioni sull'uso di Archiviazione BLOB di Azure, vedere la documentazione di Archiviazione BLOB di Azure.

Eseguire lo script in locale per testare e convalidare la funzionalità.

python main.py

Lo script deve produrre un file di output denominato iris_setosa.csv che contiene solo i record di dati con Species = setosa. Dopo aver verificato che funzioni correttamente, caricare il file di script main.py nel contenitore di input di Storage Explorer.

Configurare una pipeline di Data Factory

Creare e convalidare una pipeline di Data Factory che usa lo script Python.

Ottenere le informazioni sull'account

La pipeline di Data Factory usa i nomi degli account Batch e di archiviazione, i valori delle chiavi dell'account e l'endpoint dell'account Batch. Per ottenere queste informazioni dal portale di Azure:

1. Nella barra di Ricerca di Azure cercare e selezionare il nome dell'account Batch.

2. Nella pagina Account Batch selezionare Chiavi nel riquadro di spostamento a sinistra.

3. Nella pagina Chiavi copiare i valori seguenti:

   • Account Batch
   • Endpoint dell'account
   • Chiave di accesso primaria
   • Nome account di archiviazione
   • Chiave1

Creare ed eseguire la pipeline

1. Se Azure Data Factory Studio non è già in esecuzione, selezionare Avvia studio nella pagina Data Factory nel portale di Azure.

2. In Data Factory Studio selezionare l'icona a forma di matita Autore nel riquadro di spostamento a sinistra.

3. In Risorse factory selezionare l'icona + e quindi selezionare Pipeline.

4. Nel riquadro Proprietà a destra modificare il nome della pipeline in Esegui Python.

   

5. Nel riquadro Attività espandere Servizio Batch e trascinare l'attività Personalizzata nell'area di progettazione della pipeline.

6. Sotto l'area di disegno della finestra di progettazione, nella scheda Generale immettere testPipeline in Nome.

   

7. Selezionare la scheda Azure Batch e quindi selezionare Nuovo.

8. Completare il modulo Nuovo servizio collegato come indicato di seguito:

   • Nome: immettere un nome per il servizio collegato, ad esempio AzureBatch1.
   • Chiave di accesso: immettere la chiave di accesso primaria copiata dall'account Batch.
   • Nome account: immettere il nome dell'account Batch.
   • URL batch: immettere l'endpoint dell'account copiato dall'account Batch, ad esempio https://batchdotnet.eastus.batch.azure.com.
   • Nome pool: immettere custom-activity-pool, il pool creato in Batch Explorer.
   • Nome del servizio collegato dell'account di archiviazione: selezionare Nuovo. Nella schermata successiva immettere un nome per il servizio di archiviazione collegato, ad esempio AzureBlobStorage1, selezionare la sottoscrizione di Azure e l'account di archiviazione collegato e quindi selezionare Crea.

9. Nella parte inferiore della schermata Batch Nuovo servizio collegato selezionare Test connessione. Al termine della connessione, selezionare Crea.

   

10. Selezionare la scheda Impostazioni e immettere o selezionare le impostazioni seguenti:

    • Comando: immettere cmd /C python main.py.
    • Servizio collegato alle risorse: selezionare il servizio di archiviazione collegato creato, ad esempio AzureBlobStorage1, e testare la connessione per assicurarsi che abbia esito positivo.
    • Percorso cartella: selezionare l'icona della cartella e quindi selezionare il contenitore di input e selezionare OK. I file di questa cartella vengono scaricati dal contenitore ai nodi del pool prima dell'esecuzione dello script Python.

    

11. Selezionare Convalida sulla barra degli strumenti della pipeline per convalidare la pipeline.

12. Selezionare Debug per testare la pipeline e assicurarsi che funzioni correttamente.

13. Selezionare Pubblica tutto per pubblicare la pipeline.

14. Selezionare Aggiungi trigger e quindi Trigger now per eseguire la pipeline oppure New/Edit (Nuovo/Modifica ) per pianificarla.

    

Usare Batch Explorer per visualizzare i file di log

Se l'esecuzione della pipeline genera avvisi o errori, è possibile usare Batch Explorer per esaminare i file di output stdout.txt e stderr.txt per altre informazioni.

1. In Batch Explorer selezionare Processi dalla barra laterale sinistra.
2. Selezionare il processo adfv2-custom-activity-pool .
3. Selezionare un'attività con codice di uscita dell'errore.
4. Visualizzare i file stdout.txt e stderr.txt per analizzare e diagnosticare il problema.

Pulire le risorse

Gli account Batch, i processi e le attività sono gratuiti, ma i nodi di calcolo comportano addebiti anche quando non eseguono processi. È consigliabile allocare pool di nodi solo in base alle esigenze ed eliminare i pool al termine dell'operazione. L'eliminazione di pool elimina tutti gli output delle attività nei nodi e i nodi stessi.

I file di input e output rimangono nell'account di archiviazione e possono comportare addebiti. Quando i file non sono più necessari, è possibile eliminare i file o i contenitori. Quando non è più necessario l'account Batch o l'account di archiviazione collegato, è possibile eliminarli.

Passaggi successivi

In questa esercitazione si è appreso come usare uno script Python con Batch Explorer, Storage Explorer e Data Factory per eseguire un carico di lavoro Batch. Per altre informazioni su Data Factory, vedere Che cos'è Azure Data Factory?