Condividi tramite

Inviare processi Spark in Azure Machine Learning

  • Articolo

SI APPLICA A:Estensione ML dell'interfaccia della riga di comando di Azure v2 (corrente)Python SDK azure-ai-ml v2 (corrente)

Azure Machine Learning supporta l'invio di processi di apprendimento automatico autonomi e la creazione di pipeline di apprendimento automatico che coinvolgono più passaggi del flusso di lavoro di apprendimento automatico. Azure Machine Learning gestisce sia la creazione di processi Spark autonomi che la creazione di componenti Spark riutilizzabili che possono essere usati dalle pipeline di Azure Machine Learning. Questo articolo mostra come inviare processi Spark usando:

  • Interfaccia utente di Studio di Azure Machine Learning
  • Interfaccia della riga di comando di Azure Machine Learning
  • SDK di Azure Machine Learning

Per altre informazioni su Apache Spark in Azure Machine Learning, vedere questa risorsa.

Prerequisiti

SI APPLICA A:Estensione ML dell'interfaccia della riga di comando di Azure v2 (corrente)

Nota

  • Per altre informazioni sull'accesso alle risorse durante l'uso dell’ambiente di calcolo Spark serverless di Azure Machine Learning e del pool di Spark per Synapse collegato, vedere Garantire l'accesso alle risorse ai processi Spark.
  • Azure Machine Learning offre un pool di quote condivise da cui tutti gli utenti possono accedere alla quota di calcolo per eseguire i test per un periodo di tempo limitato. Quando si usa l’ambiente di calcolo Spark serverless, Azure Machine Learning consente di accedere a questa quota condivisa per un breve periodo di tempo.

Collegare un'identità gestita assegnata dall'utente tramite l'interfaccia della riga di comando v2

  1. Creare un file YAML che definisce l'identità gestita assegnata dall'utente che deve essere collegata all'area di lavoro: 
    identity:
  type: system_assigned,user_assigned
  tenant_id: <TENANT_ID>
  user_assigned_identities:
    '/subscriptions/<SUBSCRIPTION_ID/resourceGroups/<RESOURCE_GROUP>/providers/Microsoft.ManagedIdentity/userAssignedIdentities/<AML_USER_MANAGED_ID>':
      {}
  2. Con il parametro --file usare il file YAML nel comandoaz ml workspace update per collegare l'identità gestita assegnata dall'utente: 
    az ml workspace update --subscription <SUBSCRIPTION_ID> --resource-group <RESOURCE_GROUP> --name <AML_WORKSPACE_NAME> --file <YAML_FILE_NAME>.yaml

Collegare un'identità gestita assegnata dall'utente usando ARMClient

  1. Installare ARMClient, un semplice strumento da riga di comando che richiama l'API di Azure Resource Manager.
  2. Creare un file JSON che definisce l'identità gestita assegnata dall'utente che deve essere collegata all'area di lavoro: 
    {
    "properties":{
    },
    "location": "<AZURE_REGION>",
    "identity":{
        "type":"SystemAssigned,UserAssigned",
        "userAssignedIdentities":{
            "/subscriptions/<SUBSCRIPTION_ID/resourceGroups/<RESOURCE_GROUP>/providers/Microsoft.ManagedIdentity/userAssignedIdentities/<AML_USER_MANAGED_ID>": { }
        }
    }
}
  3. Per collegare l'identità gestita assegnata dall'utente all'area di lavoro, eseguire il comando seguente nel prompt di PowerShell o nel prompt dei comandi. 
    armclient PATCH https://management.azure.com/subscriptions/<SUBSCRIPTION_ID>/resourceGroups/<RESOURCE_GROUP>/providers/Microsoft.MachineLearningServices/workspaces/<AML_WORKSPACE_NAME>?api-version=2022-05-01 '@<JSON_FILE_NAME>.json'

Nota

Inviare un processo Spark autonomo

Dopo aver apportato le modifiche necessarie per la parametrizzazione degli script Python, è possibile usare uno script Python sviluppato dal data wrangling interattivo per inviare un processo batch per elaborare un volume di dati più grande. Un processo batch di data wrangling semplice può essere inviato come processo Spark autonomo.

Un processo Spark richiede uno script Python che accetta argomenti che è possibile sviluppare modificando il codice Python sviluppato per il data wrangling interattivo. Di seguito è riportato uno script Python di esempio.

# titanic.py
import argparse
from operator import add
import pyspark.pandas as pd
from pyspark.ml.feature import Imputer

parser = argparse.ArgumentParser()
parser.add_argument("--titanic_data")
parser.add_argument("--wrangled_data")

args = parser.parse_args()
print(args.wrangled_data)
print(args.titanic_data)

df = pd.read_csv(args.titanic_data, index_col="PassengerId")
imputer = Imputer(inputCols=["Age"], outputCol="Age").setStrategy(
    "mean"
)  # Replace missing values in Age column with the mean value
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(args.wrangled_data, index_col="PassengerId")

Nota

Questo esempio di codice Python usa pyspark.pandas. Ciò è supportato solo dal runtime Spark versione 3.2 o successiva.

Lo script precedente accetta due argomenti --titanic_data e --wrangled_data, che passano rispettivamente il percorso dei dati di input e della cartella di output.

SI APPLICA A:Estensione ML dell'interfaccia della riga di comando di Azure v2 (corrente)

Per creare un processo è possibile definire un processo Spark autonomo come file di specifica YAML, che può essere usato nel comando az ml job create con il parametro--file. Definire queste proprietà nel file YAML:

Proprietà YAML nella specifica del processo Spark

  • type : impostare su spark.

  • code : definisce il percorso della cartella che contiene il codice sorgente e gli script per questo processo.

  • entry : definisce il punto di ingresso per il processo. Deve coprire una di queste proprietà:

    • file : definisce il nome dello script Python che serve da punto di ingresso per il processo.

  • py_files : definisce un elenco di file .zip, .egg o .py da includere in PYTHONPATH per l'esecuzione corretta del processo. Questa proprietà è facoltativa.

  • jars : definisce un elenco di file .jar da includere nel driver Spark e nell'executor CLASSPATH per l'esecuzione corretta del processo. Questa proprietà è facoltativa.

  • files : definisce un elenco di file che devono essere copiati nella directory di lavoro di ogni executor per l'esecuzione corretta del processo. Questa proprietà è facoltativa.

  • archives : definisce un elenco di archivi che devono essere estratti nella directory di lavoro di ogni executor per l'esecuzione corretta del processo. Questa proprietà è facoltativa.

  • conf : definisce queste proprietà del driver Spark e dell'executor:

    • spark.driver.cores: numero di core per il driver Spark.
    • spark.driver.memory: memoria allocata per il driver Spark in gigabyte (GB).
    • spark.executor.cores: numero di core per l'executor Spark.
    • spark.executor.memory: memoria allocata per l'executor Spark in gigabyte (GB).
    • spark.dynamicAllocation.enabled - indica se gli executor devono essere allocati in modo dinamico, come valore di True o di False.
    • Se l'allocazione dinamica degli executor è abilitata, definire queste proprietà:
      • spark.dynamicAllocation.minExecutors : numero minimo di istanze degli executor Spark per l'allocazione dinamica.
      • spark.dynamicAllocation.maxExecutors : numero massimo di istanze degli executor Spark per l'allocazione dinamica.
    • Se l'allocazione dinamica degli executor è disabilitata, definire questa proprietà:
      • spark.executor.instances : numero di istanze dell'executor Spark.

  • environment : un ambiente di Azure Machine Learning per eseguire il processo.

  • args : gli argomenti della riga di comando che devono essere passati allo script Python del punto di ingresso del processo. Per un esempio, vedere il file di specifica YAML fornito qui.

  • resources : questa proprietà definisce le risorse da usare in un ambiente di calcolo Spark serverless di Azure Machine Learning. Usa le proprietà seguenti:

    • instance_type : tipo di istanza di calcolo da usare per il pool di Spark. Attualmente sono supportati i tipi di istanza seguenti:
      • standard_e4s_v3
      • standard_e8s_v3
      • standard_e16s_v3
      • standard_e32s_v3
      • standard_e64s_v3
    • runtime_version : definisce la versione del runtime di Spark. Attualmente sono supportate le versioni di runtime di Spark seguenti:
      • 3.2
      • 3.3

        Importante

        Runtime di Azure Synapse per Apache Spark: annunci

        • Runtime di Azure Synapse per Apache Spark 3.2:
          • Data annuncio EOLA: 8 luglio 2023
          • Data di fine supporto: 8 luglio 2024. Dopo questa data il runtime sarà disabilitato.
        • Per un supporto continuo e prestazioni ottimali si consiglia di eseguire la migrazione ad Apache Spark 3.3.

    Questo è un esempio:

    resources:
  instance_type: standard_e8s_v3
  runtime_version: "3.3"

  • compute : questa proprietà definisce il nome di un pool di Spark per Synapse collegato, come illustrato in questo esempio:

    compute: mysparkpool

  • inputs : questa proprietà definisce gli input per il processo Spark. Gli input per un processo Spark possono essere un valore letterale o dati archiviati in un file o in una cartella.

    • Un valore letterale può essere un numero, un valore booleano o una stringa. Di seguito sono riportati alcuni esempi: 
      inputs:
  sampling_rate: 0.02 # a number
  hello_number: 42 # an integer
  hello_string: "Hello world" # a string
  hello_boolean: True # a boolean value
    • I dati archiviati in un file o in una cartella devono essere definiti usando queste proprietà:
      • type : impostare questa proprietà suuri_file o uri_folder per i dati di input contenuti rispettivamente in un file o in una cartella.
      • path - URI dei dati di input, ad esempio azureml://, abfss:// o wasbs://.
      • mode: impostare questa proprietà su direct. Questo esempio mostra la definizione di un input del processo, a cui è possibile fare riferimento come $${inputs.titanic_data}}: 
        inputs:
  titanic_data:
    type: uri_file
    path: azureml://datastores/workspaceblobstore/paths/data/titanic.csv
    mode: direct

  • outputs : questa proprietà definisce gli output del processo Spark. Gli output di un processo Spark possono essere scritti in un file o in un percorso di cartella, definiti usando le tre proprietà seguenti:

    • type : questa proprietà può essere impostata su uri_file o uri_folder per la scrittura di dati di output rispettivamente in un file o in una cartella.
    • path : questa proprietà definisce l'URI del percorso di output, ad esempio azureml://, abfss:// o wasbs://.
    • mode: impostare questa proprietà su direct. Questo esempio mostra la definizione di un output del processo, a cui è possibile fare riferimento come: ${{outputs.wrangled_data}} 
      outputs:
  wrangled_data:
    type: uri_folder
    path: azureml://datastores/workspaceblobstore/paths/data/wrangled/
    mode: direct

  • identity : questa proprietà facoltativa definisce l'identità usata per inviare questo processo. Può avere i valori user_identity e managed. Se la specifica YAML non definisce un'identità, il processo Spark usa l'identità predefinita.

Processo Spark autonomo

Questa specifica YAML di esempio mostra un processo Spark autonomo. Usa un ambiente di calcolo Spark serverless di Azure Machine Learning:

$schema: http://azureml/sdk-2-0/SparkJob.json
type: spark

code: ./ 
entry:
  file: titanic.py

conf:
  spark.driver.cores: 1
  spark.driver.memory: 2g
  spark.executor.cores: 2
  spark.executor.memory: 2g
  spark.executor.instances: 2

inputs:
  titanic_data:
    type: uri_file
    path: azureml://datastores/workspaceblobstore/paths/data/titanic.csv
    mode: direct

outputs:
  wrangled_data:
    type: uri_folder
    path: azureml://datastores/workspaceblobstore/paths/data/wrangled/
    mode: direct

args: >-
  --titanic_data ${{inputs.titanic_data}}
  --wrangled_data ${{outputs.wrangled_data}}

identity:
  type: user_identity

resources:
  instance_type: standard_e4s_v3
  runtime_version: "3.3"

Nota

Per usare un pool di Spark per Synapse collegato definire la proprietà compute nel file di specifica YAML di esempio mostrato in precedenza, anziché nella proprietà resources.

I file YAML mostrati in precedenza possono essere usati nel comando az ml job create con il parametro --file per creare un processo Spark autonomo, come mostrato di seguito:

az ml job create --file <YAML_SPECIFICATION_FILE_NAME>.yaml --subscription <SUBSCRIPTION_ID> --resource-group <RESOURCE_GROUP> --workspace-name <AML_WORKSPACE_NAME>

È possibile eseguire il comando precedente da:

Componente Spark in un processo della pipeline

Un componente Spark offre la flessibilità di usare lo stesso componente in più Pipeline di Azure Machine Learning come passaggio della pipeline.

SI APPLICA A:Estensione ML dell'interfaccia della riga di comando di Azure v2 (corrente)

La sintassi YAML per un componente Spark è simile alla sintassi YAML per la specifica del processo Spark da più punti di vista. Queste proprietà sono definite in modo diverso nella specifica YAML del componente Spark:

  • name : nome del componente Spark.

  • version : versione del componente Spark.

  • display_name : nome del componente Spark da visualizzare nell'interfaccia utente e altrove.

  • description : la descrizione del componente Spark.

  • inputs : questa proprietà è simile alla proprietà inputs descritta nella sintassi YAML per la specifica del processo Spark, ad eccezione del fatto che non definisce la proprietà path. Questo frammento di codice mostra un esempio della proprietà del componente Sparkinputs:

    inputs:
  titanic_data:
    type: uri_file
    mode: direct

  • outputs : questa proprietà è simile alla proprietà outputs descritta nella sintassi YAML per la specifica del processo Spark, ad eccezione del fatto che non definisce la proprietàpath. Questo frammento di codice mostra un esempio della proprietà del componente Sparkoutputs:

    outputs:
  wrangled_data:
    type: uri_folder
    mode: direct

Nota

Un componente Spark non definisce le proprietà identity, compute o resources. Il file di specifica YAML della pipeline definisce queste proprietà.

Questo file di specifica YAML fornisce un esempio di un componente Spark:

$schema: http://azureml/sdk-2-0/SparkComponent.json
name: titanic_spark_component
type: spark
version: 1
display_name: Titanic-Spark-Component
description: Spark component for Titanic data

code: ./src
entry:
  file: titanic.py

inputs:
  titanic_data:
    type: uri_file
    mode: direct

outputs:
  wrangled_data:
    type: uri_folder
    mode: direct

args: >-
  --titanic_data ${{inputs.titanic_data}}
  --wrangled_data ${{outputs.wrangled_data}}

conf:
  spark.driver.cores: 1
  spark.driver.memory: 2g
  spark.executor.cores: 2
  spark.executor.memory: 2g
  spark.dynamicAllocation.enabled: True
  spark.dynamicAllocation.minExecutors: 1
  spark.dynamicAllocation.maxExecutors: 4

Il componente Spark definito nel file di specifica YAML precedente può essere usato in un processo della pipeline di Azure Machine Learning. Per altre informazioni sulla sintassi YAML che definisce un processo della pipeline, vedere Schema YAML del processo della pipeline. Questo esempio mostra un file di specifica YAML per un processo della pipeline con un componente Spark e un ambiente di calcolo Spark serverless di Azure Machine Learning:

$schema: http://azureml/sdk-2-0/PipelineJob.json
type: pipeline
display_name: Titanic-Spark-CLI-Pipeline
description: Spark component for Titanic data in Pipeline

jobs:
  spark_job:
    type: spark
    component: ./spark-job-component.yaml
    inputs:
      titanic_data: 
        type: uri_file
        path: azureml://datastores/workspaceblobstore/paths/data/titanic.csv
        mode: direct

    outputs:
      wrangled_data:
        type: uri_folder
        path: azureml://datastores/workspaceblobstore/paths/data/wrangled/
        mode: direct

    identity:
      type: managed

    resources:
      instance_type: standard_e8s_v3
      runtime_version: "3.3"

Nota

Per usare un pool di Spark per Synapse collegato definire la proprietà compute nel file di specifica YAML di esempio mostrato in precedenza, anziché la proprietà resources.

Il file di specifica YAML precedente può essere usato nel comando az ml job create usando il parametro --file per creare un processo della pipeline come mostrato di seguito:

az ml job create --file <YAML_SPECIFICATION_FILE_NAME>.yaml --subscription <SUBSCRIPTION_ID> --resource-group <RESOURCE_GROUP> --workspace-name <AML_WORKSPACE_NAME>

È possibile eseguire il comando precedente da:

Risoluzione dei problemi relativi ai processi Spark

Per risolvere i problemi relativi a un processo Spark è possibile accedere ai log generati per tale processo in Studio di Azure Machine Learning. Per visualizzare i log per un processo Spark:

  1. Passare a Processi dal pannello sinistro nell'interfaccia utente di Studio di Azure Machine Learning
  2. Selezionare la scheda Tutti i processi
  3. Selezionare il valore Nome visualizzato per il processo
  4. Nella pagina dei dettagli del processo selezionare la scheda Output e log
  5. In Esplora file espandere la cartella log e quindi espandere la cartella azureml
  6. Accedere ai log dei processi Spark all'interno delle cartelle del driver e di Gestione librerie

Nota

Per risolvere i problemi relativi ai processi Spark creati durante il data wrangling interattivo in una sessione del notebook selezionare Dettagli del processo nell'angolo superiore destro dell'interfaccia utente del notebook. I processi Spark di una sessione interattiva del notebook vengono creati con il nome dell'esperimento esecuzioni del notebook.

Passaggi successivi