Skicka Spark-jobb i Azure Machine Learning

  • Artikel

GÄLLER FÖR:Azure CLI ml extension v2 (aktuell)Python SDK azure-ai-ml v2 (aktuell)

Azure Machine Learning stöder inlämning av fristående maskininlärningsjobb och skapande av maskininlärningspipelines som omfattar flera steg i arbetsflödet för maskininlärning. Azure Machine Learning hanterar både skapande av fristående Spark-jobb och skapandet av återanvändbara Spark-komponenter som Azure Machine Learning-pipelines kan använda. I den här artikeln får du lära dig hur du skickar Spark-jobb med hjälp av:

  • Azure Machine Learning-studio användargränssnitt
  • Azure Machine Learning CLI
  • Azure Machine Learning SDK

Mer information om Apache Spark i Azure Machine Learning-begrepp finns i den här resursen.

Förutsättningar

GÄLLER FÖR:Azure CLI ml extension v2 (aktuell)

Anteckning

Mer information om resursåtkomst när du använder serverlös Spark-beräkning i Azure Machine Learning och ansluten Synapse Spark-pool finns i Säkerställa resursåtkomst för Spark-jobb.

Koppla användartilldelad hanterad identitet med CLI v2

  1. Skapa en YAML-fil som definierar den användartilldelade hanterade identiteten som ska kopplas till arbetsytan:
    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. Med parametern --file använder du YAML-filen i az ml workspace update kommandot för att koppla den användartilldelade hanterade identiteten:
    az ml workspace update --subscription <SUBSCRIPTION_ID> --resource-group <RESOURCE_GROUP> --name <AML_WORKSPACE_NAME> --file <YAML_FILE_NAME>.yaml

Koppla användartilldelad hanterad identitet med hjälp av ARMClient

  1. Installera ARMClient, ett enkelt kommandoradsverktyg som anropar Azure Resource Manager API.
  2. Skapa en JSON-fil som definierar den användartilldelade hanterade identiteten som ska kopplas till arbetsytan:
    {
    "properties":{
    },
    "location": "<AZURE_REGION>",
    "identity":{
        "type":"SystemAssigned,UserAssigned",
        "userAssignedIdentities":{
            "/subscriptions/<SUBSCRIPTION_ID/resourceGroups/<RESOURCE_GROUP>/providers/Microsoft.ManagedIdentity/userAssignedIdentities/<AML_USER_MANAGED_ID>": { }
        }
    }
}
  3. Kör följande kommando i PowerShell-prompten eller kommandotolken för att koppla den användartilldelade hanterade identiteten till arbetsytan. 
    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'

Anteckning

Skicka ett fristående Spark-jobb

Ett Python-skript som utvecklats av interaktiv dataomvandling kan användas för att skicka ett batchjobb för att bearbeta en större mängd data efter att ha gjort nödvändiga ändringar för Python-skriptparameterisering. Ett enkelt batchjobb för dataomvandling kan skickas som ett fristående Spark-jobb.

Ett Spark-jobb kräver ett Python-skript som tar argument, som kan utvecklas med ändring av Python-koden som utvecklats från interaktiv dataomvandling. Ett Python-exempelskript visas här.

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

Anteckning

Det här Python-kodexemplet använder pyspark.pandas, som endast stöds av Spark-körningsversion 3.2.

Skriptet ovan tar två argument --titanic_data och --wrangled_data, som skickar sökvägen för indata respektive utdatamapp.

GÄLLER FÖR:Azure CLI ml extension v2 (aktuell)

För att skapa ett jobb kan ett fristående Spark-jobb definieras som en YAML-specifikationsfil som kan användas i az ml job create kommandot med parametern --file . Definiera dessa egenskaper i YAML-filen på följande sätt:

YAML-egenskaper i Spark-jobbspecifikationen

  • type - inställd på spark.

  • code – definierar platsen för mappen som innehåller källkod och skript för det här jobbet.

  • entry – definierar startpunkten för jobbet. Den bör omfatta någon av följande egenskaper:

    • file – definierar namnet på Python-skriptet som fungerar som en startpunkt för jobbet.
    • class_name – definierar namnet på den klass som fungerar som en startpunkt för jobbet.

  • py_files – definierar en lista över .zip, .eggeller .py filer som ska placeras i PYTHONPATH, för att jobbet ska kunna köras. Den här egenskapen är valfri.

  • jars – definierar en lista över .jar filer som ska ingå i Spark-drivrutinen och kören CLASSPATHför att jobbet ska kunna köras. Den här egenskapen är valfri.

  • files – definierar en lista över filer som ska kopieras till arbetskatalogen för varje utförare för lyckad jobbkörning. Den här egenskapen är valfri.

  • archives – definierar en lista över arkiv som ska extraheras till arbetskatalogen för varje utförare för lyckad jobbkörning. Den här egenskapen är valfri.

  • conf – definierar dessa Egenskaper för Spark-drivrutin och -köre:

    • spark.driver.cores: antalet kärnor för Spark-drivrutinen.
    • spark.driver.memory: allokerat minne för Spark-drivrutinen i gigabyte (GB).
    • spark.executor.cores: antalet kärnor för Spark-kören.
    • spark.executor.memory: minnesallokeringen för Spark-kören i gigabyte (GB).
    • spark.dynamicAllocation.enabled – om exekutorer ska allokeras dynamiskt, som ett True eller False -värde.
    • Om dynamisk allokering av utförare är aktiverad definierar du följande egenskaper:
      • spark.dynamicAllocation.minExecutors – det minsta antalet Spark-körinstanser för dynamisk allokering.
      • spark.dynamicAllocation.maxExecutors – det maximala antalet Spark-körinstanser för dynamisk allokering.
    • Om dynamisk allokering av utförare är inaktiverad definierar du den här egenskapen:
      • spark.executor.instances – antalet Spark-körinstanser.

  • environment – en Azure Machine Learning-miljö för att köra jobbet.

  • args – de kommandoradsargument som ska skickas till jobbets startpunkt Python-skript eller -klass. Se YAML-specifikationsfilen som finns här för ett exempel.

  • resources – Den här egenskapen definierar de resurser som ska användas av en serverlös Spark-beräkning i Azure Machine Learning. Den använder följande egenskaper:

    • instance_type – den beräkningsinstanstyp som ska användas för Spark-poolen. Följande instanstyper stöds för närvarande:
      • standard_e4s_v3
      • standard_e8s_v3
      • standard_e16s_v3
      • standard_e32s_v3
      • standard_e64s_v3
    • runtime_version – definierar Spark-körningsversionen. Följande Spark-körningsversioner stöds för närvarande:
      • 3.1
      • 3.2
      • 3.3

        Viktigt

        Azure Synapse Runtime för Apache Spark: Meddelanden

        • Azure Synapse Runtime för Apache Spark 3.1:
          • Slutdatum (EOLA) Meddelandedatum: 26 januari 2023
          • Slutdatum för support: 31 juli 2023. Efter det här datumet inaktiveras körningen.
        • Azure Synapse Runtime för Apache Spark 3.2:
          • EOLA-meddelandedatum: 8 juli 2023
          • Slutdatum för support: 8 juli 2024. Efter det här datumet inaktiveras körningen.
        • För fortsatt support och optimala prestanda rekommenderar vi migrering till Apache Spark 3.3.

    Ett exempel på detta visas här:

    resources:
  instance_type: standard_e8s_v3
  runtime_version: "3.3"

  • compute – Den här egenskapen definierar namnet på en ansluten Synapse Spark-pool, som du ser i det här exemplet:

    compute: mysparkpool

  • inputs – den här egenskapen definierar indata för Spark-jobbet. Indata för ett Spark-jobb kan vara antingen ett literalvärde eller data som lagras i en fil eller mapp.

    • Ett literalvärde kan vara ett tal, ett booleskt värde eller en sträng. Några exempel visas här:
      inputs:
  sampling_rate: 0.02 # a number
  hello_number: 42 # an integer
  hello_string: "Hello world" # a string
  hello_boolean: True # a boolean value
    • Data som lagras i en fil eller mapp bör definieras med hjälp av följande egenskaper:
      • type – ange den här egenskapen till uri_file, eller uri_folder, för indata som finns i en fil eller en mapp.
      • path – URI för indata, till exempel azureml://, abfss://eller wasbs://.
      • mode – ange den här egenskapen till direct. Det här exemplet visar definitionen av ett jobbindata, som kan kallas $${inputs.titanic_data}}:
        inputs:
  titanic_data:
    type: uri_file
    path: azureml://datastores/workspaceblobstore/paths/data/titanic.csv
    mode: direct

  • outputs – den här egenskapen definierar Spark-jobbutdata. Utdata för ett Spark-jobb kan skrivas till antingen en fil eller en mappplats, som definieras med hjälp av följande tre egenskaper:

    • type– den här egenskapen kan anges till eller uri_folder för att uri_file skriva utdata till en fil eller en mapp.
    • path – den här egenskapen definierar utdataplats-URI:n, till exempel azureml://, abfss://eller wasbs://.
    • mode – ange den här egenskapen till direct. Det här exemplet visar definitionen av ett jobbutdata, som kan kallas ${{outputs.wrangled_data}}:
      outputs:
  wrangled_data:
    type: uri_folder
    path: azureml://datastores/workspaceblobstore/paths/data/wrangled/
    mode: direct

  • identity – den här valfria egenskapen definierar den identitet som används för att skicka jobbet. Det kan ha user_identity och managed värden. Om YAML-specifikationen inte definierar en identitet använder Spark-jobbet standardidentiteten.

Fristående Spark-jobb

Den här YAML-exempelspecifikationen visar ett fristående Spark-jobb. Den använder en serverlös Spark-beräkning i 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"

Anteckning

Om du vill använda en ansluten Synapse Spark-pool definierar du compute egenskapen i yaml-exempelspecifikationsfilen som visas ovan i stället för egenskapen resources .

YAML-filerna som visas ovan kan användas i az ml job create kommandot med parametern --file för att skapa ett fristående Spark-jobb enligt följande:

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

Du kan köra kommandot ovan från:

Spark-komponent i ett pipelinejobb

En Spark-komponent ger flexibiliteten att använda samma komponent i flera Azure Machine Learning-pipelines som ett pipelinesteg.

GÄLLER FÖR:Azure CLI ml extension v2 (aktuell)

YAML-syntaxen för en Spark-komponent liknar YAML-syntaxen för Spark-jobbspecifikationen på de flesta sätt. Dessa egenskaper definieras på olika sätt i Spark-komponentens YAML-specifikation:

  • name – Namnet på Spark-komponenten.

  • version – versionen av Spark-komponenten.

  • display_name – namnet på Spark-komponenten som ska visas i användargränssnittet och någon annanstans.

  • description – beskrivningen av Spark-komponenten.

  • inputs – den här egenskapen liknar inputs egenskapen som beskrivs i YAML-syntaxen för Spark-jobbspecifikationen, förutom att den path inte definierar egenskapen. Det här kodfragmentet visar ett exempel på egenskapen Spark-komponent inputs :

    inputs:
  titanic_data:
    type: uri_file
    mode: direct

  • outputs – den här egenskapen liknar egenskapen outputs som beskrivs i YAML-syntaxen för Spark-jobbspecifikationen, förutom att den path inte definierar egenskapen. Det här kodfragmentet visar ett exempel på egenskapen Spark-komponent outputs :

    outputs:
  wrangled_data:
    type: uri_folder
    mode: direct

Anteckning

En Spark-komponent definierar identityinte , compute eller resources egenskaper. Yaml-specifikationsfilen för pipeline definierar dessa egenskaper.

Den här YAML-specifikationsfilen innehåller ett exempel på en Spark-komponent:

$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

Spark-komponenten som definierats i YAML-specifikationsfilen ovan kan användas i ett Azure Machine Learning-pipelinejobb. Mer information om YAML-syntaxen som definierar ett pipelinejobb finns i YAML-schema för pipelinejobb . Det här exemplet visar en YAML-specifikationsfil för ett pipelinejobb, med en Spark-komponent och en serverlös Spark-beräkning i 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.2"

Anteckning

Om du vill använda en ansluten Synapse Spark-pool definierar du compute egenskapen i yaml-exempelspecifikationsfilen som visas ovan i stället för resources egenskapen.

Ovanstående YAML-specifikationsfil kan användas i az ml job create kommandot med hjälp av parametern --file för att skapa ett pipelinejobb enligt följande:

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

Du kan köra kommandot ovan från:

Felsöka Spark-jobb

Om du vill felsöka ett Spark-jobb kan du komma åt loggarna som genererats för jobbet i Azure Machine Learning-studio. Så här visar du loggarna för ett Spark-jobb:

  1. Gå till Jobb från den vänstra panelen i Azure Machine Learning-studio-användargränssnittet
  2. Välj fliken Alla jobb
  3. Välj värdet Visningsnamn för jobbet
  4. På sidan jobbinformation väljer du fliken Utdata + loggar
  5. Expandera loggmappen i utforskaren och expandera sedan mappen azureml
  6. Komma åt Spark-jobbloggarna i mapparna för drivrutins- och bibliotekshanteraren

Anteckning

Om du vill felsöka Spark-jobb som skapats under interaktiva dataomvandlingar i en notebook-session väljer du Jobbinformation i det övre högra hörnet i notebook-användargränssnittet. Ett Spark-jobb från en interaktiv notebook-session skapas under experimentnamnet notebook-runs.

Nästa steg