Come usare il processo in parallelo nella pipeline (V2)

Articolo
09/03/2024

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

Il processo in parallelo consente agli utenti di accelerare l'esecuzione del processo distribuendo attività ripetute in cluster di calcolo multinodo potenti. Si consideri ad esempio lo scenario in cui si esegue un modello di rilevamento oggetti in un set di immagini di grandi dimensioni. Con il processo in parallelo di Azure Machine Learning, è possibile distribuire facilmente le immagini per eseguire codice personalizzato in parallelo in un cluster di calcolo specifico. La parallelizzazione potrebbe ridurre significativamente il costo orario. L'uso del processo in parallelo di Azure Machine Learning consente inoltre di semplificare e automatizzare il processo per renderlo più efficiente.

Prerequisito

Il processo in parallelo di Azure Machine Learning può essere usato solo come uno dei passaggi in un processo della pipeline. È quindi importante avere familiarità con l'uso delle pipeline. Per altre informazioni sulle pipeline di Azure Machine Learning, vedere gli articoli seguenti.

Comprensione del concetto di pipeline di Azure Machine Learning
Familiarità con l'uso della pipeline di Azure Machine Learning con l'interfaccia della riga di comando v2 e SDK v2.

Perché sono necessari processi in parallelo?

Nel mondo reale i tecnici che si occupano di apprendimento automatico hanno sempre requisiti di scalabilità per le attività di training o inferenza. Ad esempio, quando un data scientist fornisce un singolo script per eseguire il training di un modello di stima delle vendite, i tecnici che si occupano di apprendimento automatico devono applicare questa attività di training a ogni singolo archivio. Durante questo processo di aumento del numero di istanze si verificano alcune problematiche, tra cui:

Pressione dovuta al ritardo causata da tempi di esecuzione lunghi.
Intervento manuale per gestire problemi imprevisti per continuare l'attività.

Il valore principale del processo in parallelo di Azure Machine Learning consiste nel suddividere una singola attività seriale in mini batch e inviare tali mini batch a più ambienti di calcolo per l'esecuzione in parallelo. Usando processi in parallelo, è possibile:

Ridurre significativamente il tempo di esecuzione end-to-end.
Usare le impostazioni di gestione automatica degli errori del processo in parallelo di Azure Machine Learning.

È consigliabile usare il processo in parallelo di Azure Machine Learning se:

Si prevede di eseguire il training di molti modelli sui dati partizionati.
Si vuole accelerare l'attività di inferenza batch su larga scala.

Preparare il processo in parallelo

A differenza di altri tipi di processi, un processo in parallelo richiede preparazione. Seguire le sezioni successive per preparare la creazione del processo in parallelo.

Dichiarare gli input da distribuire e l'impostazione della divisione dati

Il processo in parallelo richiede una sola origine di dati di input principali da suddividere ed elaborare in parallelo. I dati di input principali possono essere dati tabulari o un set di file. Tipi di input diversi possono avere un metodo di divisione dati diverso.

Nella tabella seguente viene illustrata la relazione tra i dati di input e il metodo di divisione dati:

Formato dati	Tipo di input di Azure Machine Learning	Modalità di input di Azure Machine Learning	Metodo di divisione dati
Elenco di file	`mltable` oppure `uri_folder`	ro_mount o download	Per dimensione (numero di file) Per partizioni
Dati tabulari	`mltable`	dirette	Per dimensione (dimensioni fisiche stimate) Per partizioni

È possibile dichiarare i dati di input principali con l'attributo input_data nel codice YAML o Python SDK per il processo in parallelo. È anche possibile associarli a uno dei valori inputs definiti del processo in parallelo usando ${{inputs.<input name>}}. È quindi necessario definire il metodo di divisione dati per l'input principale compilando un attributo diverso:

Metodo di divisione dati	Attribute name	Tipo di attributo	Esempio di processo
Per dimensioni	mini_batch_size	string	Stima batch Iris
Per partizioni	partition_keys	Elenco di stringhe	Stima delle vendite di succo d'arancia

Interfaccia della riga di comando di Azure
Python

SI APPLICA A: estensione ml dell’interfaccia della riga di comando di Azure v2 (corrente)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

SI APPLICA A: Python SDK azure-ai-ml v2 (corrente)

Dichiarare job_data_path come uno degli input. Associarlo all'attributo input_data.

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.5/labels/latest",
    ),
)

Dopo aver definito l'impostazione della divisione dei dati, è possibile configurare il numero di risorse per la parallelizzazione compilando i due attributi seguenti:

Attribute name	Type	Descrizione	Default value
`instance_count`	integer	Numero di nodi da usare per il processo.	1
`max_concurrency_per_instance`	integer	Numero di processori in ogni nodo.	Per una risorsa di calcolo di GPU, il valore predefinito è 1. Per una risorsa di calcolo della CPU, il valore predefinito è il numero di core.

Questi due attributi interagiscono con il cluster di calcolo specificato.

Codice di esempio per impostare due attributi:

Interfaccia della riga di comando di Azure
Python

SI APPLICA A: estensione ml dell’interfaccia della riga di comando di Azure v2 (corrente)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

SI APPLICA A: Python SDK azure-ai-ml v2 (corrente)

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.5/labels/latest",
    ),
)

Nota

Se si usano mltable tabulari come dati di input principali, è necessario avere il file di specifica MLTABLE con la sezione transformations - read_delimited compilata nel percorso specifico. Per altri esempi, vedere Creare un asset di dati mltable

Implementare funzioni predefinite nello script di immissione

Lo script di immissione è un singolo file Python in cui l'utente deve implementare tre funzioni predefinite con codice personalizzato. Il processo in parallelo di Azure Machine Learning segue il diagramma seguente per eseguirli in ogni processore.

Nome della funzione	Richiesto	Descrzione	Input	Restituzione
Init()	Y	Usare questa funzione per la preparazione comune prima di iniziare a eseguire mini batch. Usarla ad esempio per caricare il modello in un oggetto globale.	--	--
Run(mini_batch)	Y	Implementa la logica di esecuzione principale per mini_batches.	mini_batch: Dataframe Pandas se i dati di input sono dati tabulari. Elenco del percorso del file se i dati di input sono una directory.	Dataframe, elenco o tupla.
Shutdown()	N	Funzione facoltativa per eseguire operazioni di pulizia personalizzate prima di restituire la risorsa di calcolo al pool.	--	--

Per ottenere altri dettagli, vedere gli esempi di script di immissione seguenti:

Quando lo script di immissione è pronto, è possibile impostare i due attributi seguenti per usarlo nel processo in parallelo:

Attribute name	Type	Descrizione	Default value
`code`	string	Percorso locale della directory del codice sorgente da caricare e usare per il processo.
`entry_script`	string	File Python che contiene l'implementazione di funzioni parallele predefinite.

Codice di esempio per impostare due attributi:

Interfaccia della riga di comando di Azure
Python

SI APPLICA A: estensione ml dell’interfaccia della riga di comando di Azure v2 (corrente)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

  task:
    type: run_function
    code: "./script"
    entry_script: iris_prediction.py
    environment:
      name: "prs-env"
      version: 1
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
      conda_file: ./environment/environment_parallel.yml
    program_arguments: >-
      --model ${{inputs.score_model}}
      --error_threshold 5
      --allowed_failed_percent 30
      --task_overhead_timeout 1200
      --progress_update_timeout 600
      --first_task_creation_timeout 600
      --copy_logs_to_parent True
      --resource_monitor_interva 20
    append_row_to: ${{outputs.job_output_file}}

SI APPLICA A: Python SDK azure-ai-ml v2 (corrente)

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.5/labels/latest",
    ),
)

Importante

La funzione Run(mini_batch) richiede la restituzione di un dataframe, un elenco o una tupla. Il processo in parallelo userà il conteggio di tali elementi restituiti per misurare gli elementi con esito positivo in tale mini batch. Idealmente, il numero di mini batch deve essere uguale al conteggio degli elenchi restituiti se tutti gli elementi sono stati elaborati correttamente in questo mini batch.

Importante

Se si vogliono analizzare gli argomenti nella funzione Init() o Run(mini_batch), usare "parse_known_args" anziché "parse_args" per evitare eccezioni. Vedere l'esempio iris_score per lo script di immissione con parser di argomenti.

Importante

Se si usa mltable come dati di input principali, è necessario installare la libreria "mltable" nell'ambiente. Vedere la riga 9 di questo file conda di esempio.

Prendere in considerazione le impostazioni di automazione

Il processo in parallelo di Azure Machine Learning espone numerose impostazioni per controllare automaticamente il processo senza intervento manuale. Per informazioni dettagliate, vedere la tabella seguente.

Chiave	Type	Descrizione	Valori consentiti	Valore predefinito	Impostato nell'attributo	Impostato negli argomenti del programma
mini batch error threshold	integer	Definisce il numero di mini batch non riusciti che potrebbero essere ignorati in questo processo in parallelo. Se il numero di mini batch non riusciti è superiore a questa soglia, il processo in parallelo verrà contrassegnato come non riuscito. Il mini batch viene contrassegnato come non riuscito se: - Il numero di elementi restituiti da run() è inferiore al numero di input di mini batch. - Vengono intercettate eccezioni nel codice run() personalizzato. "-1" è il numero predefinito e indica che tutti i mini batch non riusciti durante il processo in parallelo verranno ignorati.	[-1, int.max]	-1	mini_batch_error_threshold	N/D
mini batch max retries	integer	Definisce il numero di tentativi quando il mini batch ha esito negativo o si verifica il timeout. Se tutti i tentativi non sono riusciti, il mini batch verrà contrassegnato come non riuscito per il calcolo `mini_batch_error_threshold`.	[0, int.max]	2	retry_settings.max_retries	N/D
mini batch timeout	integer	Definisce il timeout in secondi per l'esecuzione di una funzione run() personalizzata. Se il tempo di esecuzione è superiore a questa soglia, il mini batch verrà interrotto e contrassegnato come mini batch non riuscito per attivare nuovi tentativi.	(0, 259200]	60	retry_settings.timeout	N/D
item error threshold	integer	Soglia di elementi non riusciti. Gli elementi non riusciti vengono conteggiati in base al divario numerico tra gli input e gli elementi restituiti da ogni mini batch. Se la somma degli elementi non riusciti è superiore a questa soglia, il processo in parallelo verrà contrassegnato come non riuscito. Nota: "-1" è il numero predefinito e indica che tutti gli errori durante il processo in parallelo verranno ignorati.	[-1, int.max]	-1	N/D	--error_threshold
allowed failed percent	integer	Simile a `mini_batch_error_threshold`, ma usa la percentuale di mini batch non riusciti anziché il conteggio.	[0, 100]	100	N/D	--allowed_failed_percent
overhead timeout	integer	Timeout in secondi per l'inizializzazione di ogni mini batch. Ad esempio, caricaremento dei dati del mini batch e passaggio alla funzione run().	(0, 259200]	600	N/D	--task_overhead_timeout
progress update timeout	integer	Timeout in secondi per il monitoraggio dello stato di avanzamento dell'esecuzione di mini batch. Se non vengono ricevuti aggiornamenti dello stato entro questa impostazione di timeout, il processo in parallelo verrà contrassegnato come non riuscito.	(0, 259200]	Calcolato dinamicamente da altre impostazioni.	N/D	--progress_update_timeout
first task creation timeout	integer	Timeout in secondi per il monitoraggio del tempo tra l'avvio del processo e l'esecuzione del primo mini batch.	(0, 259200]	600	N/D	--first_task_creation_timeout
logging level	string	Definisce il livello di log di cui verrà eseguito il dump nei file di log utente.	INFO, AVVISO o DEBUG	INFO	logging_level	N/D
append row to	string	Aggrega tutti i valori restituiti da ogni esecuzione di mini batch e li restituisce come output in questo file. Può fare riferimento a uno degli output del processo in parallelo usando l'espressione ${{outputs.<output_name>}}			task.append_row_to	N/D
copy logs to parent	string	Opzione booleana per specificare se copiare lo stato del processo, la panoramica e i log nel processo della pipeline padre.	True o False	Falso	N/D	--copy_logs_to_parent
resource monitor interval	integer	Intervallo di tempo in secondi per eseguire il dump dell'utilizzo delle risorse del nodo (ad esempio CPU, memoria) nella cartella di log nel percorso "logs/sys/perf". Nota: i log delle risorse di dump frequenti rallentano leggermente la velocità di esecuzione del mini batch. Impostare questo valore su "0" per interrompere il dump dell'utilizzo delle risorse.	[0, int.max]	600	N/D	--resource_monitor_interval

Codice di esempio per aggiornare queste impostazioni:

Interfaccia della riga di comando di Azure
Python

SI APPLICA A: estensione ml dell’interfaccia della riga di comando di Azure v2 (corrente)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

  task:
    type: run_function
    code: "./script"
    entry_script: iris_prediction.py
    environment:
      name: "prs-env"
      version: 1
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
      conda_file: ./environment/environment_parallel.yml
    program_arguments: >-
      --model ${{inputs.score_model}}
      --error_threshold 5
      --allowed_failed_percent 30
      --task_overhead_timeout 1200
      --progress_update_timeout 600
      --first_task_creation_timeout 600
      --copy_logs_to_parent True
      --resource_monitor_interva 20
    append_row_to: ${{outputs.job_output_file}}

SI APPLICA A: Python SDK azure-ai-ml v2 (corrente)

# parallel task to process tabular data
tabular_batch_inference = parallel_run_function(
    name="batch_score_with_tabular_input",
    display_name="Batch Score with Tabular Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        ),
        score_model=Input(
            type=AssetTypes.URI_FOLDER, description="The model for batch score."
        ),
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=2,
    mini_batch_size="100",
    mini_batch_error_threshold=5,
    logging_level="DEBUG",
    retry_settings=dict(max_retries=2, timeout=60),
    task=RunFunction(
        code="./src",
        entry_script="tabular_batch_inference.py",
        environment=Environment(
            image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
            conda_file="./src/environment_parallel.yml",
        ),
        program_arguments="--model ${{inputs.score_model}} "
        "--job_output_path ${{outputs.job_output_path}} "
        "--error_threshold 5 "
        "--allowed_failed_percent 30 "
        "--task_overhead_timeout 1200 "
        "--progress_update_timeout 600 "
        "--first_task_creation_timeout 600 "
        "--copy_logs_to_parent True "
        "--resource_monitor_interva 20 ",
        append_row_to="${{outputs.job_output_path}}",
    ),
)

Creare un processo in parallelo nella pipeline

Interfaccia della riga di comando di Azure
Python

SI APPLICA A: estensione ml dell’interfaccia della riga di comando di Azure v2 (corrente)

È possibile creare il processo in parallelo inline con il processo della pipeline:

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

display_name: iris-batch-prediction-using-parallel
description: The hello world pipeline job with inline parallel job
tags:
  tag: tagvalue
  owner: sdkteam

settings:
  default_compute: azureml:cpu-cluster

jobs:
  batch_prediction:
    type: parallel
    compute: azureml:cpu-cluster
    inputs:
      input_data: 
        type: mltable
        path: ./neural-iris-mltable
        mode: direct
      score_model: 
        type: uri_folder
        path: ./iris-model
        mode: download
    outputs:
      job_output_file:
        type: uri_file
        mode: rw_mount

    input_data: ${{inputs.input_data}}
    mini_batch_size: "10kb"
    resources:
        instance_count: 2
    max_concurrency_per_instance: 2

    logging_level: "DEBUG"
    mini_batch_error_threshold: 5
    retry_settings:
      max_retries: 2
      timeout: 60

    task:
      type: run_function
      code: "./script"
      entry_script: iris_prediction.py
      environment:
        name: "prs-env"
        version: 1
        image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
        conda_file: ./environment/environment_parallel.yml
      program_arguments: >-
        --model ${{inputs.score_model}}
        --error_threshold 5
        --allowed_failed_percent 30
        --task_overhead_timeout 1200
        --progress_update_timeout 600
        --first_task_creation_timeout 600
        --copy_logs_to_parent True
        --resource_monitor_interva 20
      append_row_to: ${{outputs.job_output_file}}

SI APPLICA A: Python SDK azure-ai-ml v2 (corrente)

Prima di tutto, è necessario importare le librerie necessarie, avviare il ml_client con le credenziali appropriate e creare/recuperare le risorse di calcolo:

# import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azure.ai.ml import MLClient, Input, Output, load_component
from azure.ai.ml.dsl import pipeline
from azure.ai.ml.entities import Environment
from azure.ai.ml.constants import AssetTypes, InputOutputModes
from azure.ai.ml.parallel import parallel_run_function, RunFunction

try:
    credential = DefaultAzureCredential()
    # Check if given credential can get token successfully.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential in case DefaultAzureCredential not work
    credential = InteractiveBrowserCredential()

# Get a handle to workspace
ml_client = MLClient.from_config(credential=credential)

# Retrieve an already attached Azure Machine Learning Compute.
cpu_compute_target = "cpu-cluster"
print(ml_client.compute.get(cpu_compute_target))
gpu_compute_target = "gpu-cluster"
print(ml_client.compute.get(gpu_compute_target))

Implementare quindi il processo in parallelo compilando parallel_run_function:

# parallel task to process tabular data
tabular_batch_inference = parallel_run_function(
    name="batch_score_with_tabular_input",
    display_name="Batch Score with Tabular Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        ),
        score_model=Input(
            type=AssetTypes.URI_FOLDER, description="The model for batch score."
        ),
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=2,
    mini_batch_size="100",
    mini_batch_error_threshold=5,
    logging_level="DEBUG",
    retry_settings=dict(max_retries=2, timeout=60),
    task=RunFunction(
        code="./src",
        entry_script="tabular_batch_inference.py",
        environment=Environment(
            image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
            conda_file="./src/environment_parallel.yml",
        ),
        program_arguments="--model ${{inputs.score_model}} "
        "--job_output_path ${{outputs.job_output_path}} "
        "--error_threshold 5 "
        "--allowed_failed_percent 30 "
        "--task_overhead_timeout 1200 "
        "--progress_update_timeout 600 "
        "--first_task_creation_timeout 600 "
        "--copy_logs_to_parent True "
        "--resource_monitor_interva 20 ",
        append_row_to="${{outputs.job_output_path}}",
    ),
)

Usare infine il processo in parallelo come passaggio nella pipeline e associarne l'input/output con altri passaggi:

@pipeline()
def parallel_in_pipeline(pipeline_job_data_path, pipeline_score_model):

    prepare_file_tabular_data = prepare_data(input_data=pipeline_job_data_path)
    # output of file & tabular data should be type MLTable
    prepare_file_tabular_data.outputs.file_output_data.type = AssetTypes.MLTABLE
    prepare_file_tabular_data.outputs.tabular_output_data.type = AssetTypes.MLTABLE

    batch_inference_with_file_data = file_batch_inference(
        job_data_path=prepare_file_tabular_data.outputs.file_output_data
    )
    # use eval_mount mode to handle file data
    batch_inference_with_file_data.inputs.job_data_path.mode = (
        InputOutputModes.EVAL_MOUNT
    )
    batch_inference_with_file_data.outputs.job_output_path.type = AssetTypes.MLTABLE

    batch_inference_with_tabular_data = tabular_batch_inference(
        job_data_path=prepare_file_tabular_data.outputs.tabular_output_data,
        score_model=pipeline_score_model,
    )
    # use direct mode to handle tabular data
    batch_inference_with_tabular_data.inputs.job_data_path.mode = (
        InputOutputModes.DIRECT
    )

    return {
        "pipeline_job_out_file": batch_inference_with_file_data.outputs.job_output_path,
        "pipeline_job_out_tabular": batch_inference_with_tabular_data.outputs.job_output_path,
    }


pipeline_job_data_path = Input(
    path="./dataset/", type=AssetTypes.MLTABLE, mode=InputOutputModes.RO_MOUNT
)
pipeline_score_model = Input(
    path="./model/", type=AssetTypes.URI_FOLDER, mode=InputOutputModes.DOWNLOAD
)
# create a pipeline
pipeline_job = parallel_in_pipeline(
    pipeline_job_data_path=pipeline_job_data_path,
    pipeline_score_model=pipeline_score_model,
)
pipeline_job.outputs.pipeline_job_out_tabular.type = AssetTypes.URI_FILE

# set pipeline level compute
pipeline_job.settings.default_compute = "cpu-cluster"

Inviare il processo della pipeline e controllare il passaggio parallelo nell'interfaccia utente dello studio

Interfaccia della riga di comando di Azure
Python

SI APPLICA A: estensione ml dell’interfaccia della riga di comando di Azure v2 (corrente)

È possibile inviare il processo della pipeline con un passaggio parallelo usando il comando dell'interfaccia della riga di comando:

az ml job create --file pipeline.yml

SI APPLICA A: Python SDK azure-ai-ml v2 (corrente)

È possibile inviare il processo della pipeline con un passaggio parallelo usando la funzione jobs.create_or_update di ml_client:

pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job, experiment_name="pipeline_samples"
)
pipeline_job

Dopo aver inviato il processo della pipeline, il widget dell'SDK o dell'interfaccia della riga di comando fornirà un collegamento URL Web all'interfaccia utente dello studio. Il collegamento porterà alla visualizzazione del grafico della pipeline per impostazione predefinita. Selezionare due volte il passaggio parallelo per aprire il pannello destro del processo in parallelo.

Per controllare le impostazioni del processo in parallelo, passare alla scheda Parametri, espandere Impostazioni di esecuzione e vedere la sezione Parallelo:

Per eseguire il debug dell'errore del processo in parallelo, passare alla scheda Output e log, espandere la cartella logs dalle directory di output a sinistra e controllare il file job_result.txt per capire perché il processo in parallelo non è riuscito. Per altre informazioni sulla struttura di registrazione del processo in parallelo, vedere il file readme.txt nella stessa cartella.

Esempi di processi in parallelo nella pipeline

Passaggi successivi

Per lo schema YAML dettagliato del processo in parallelo, vedere le Informazioni di riferimento di YAML per il processo in parallelo.
Per informazioni su come eseguire l'onboarding dei dati in MLTABLE, vedere Creare un asset di dati mltable.
Per informazioni su come attivare regolarmente la pipeline, vedere Come pianificare la pipeline.

Condividi tramite

Come usare il processo in parallelo nella pipeline (V2)

Prerequisito

Perché sono necessari processi in parallelo?

Preparare il processo in parallelo

Dichiarare gli input da distribuire e l'impostazione della divisione dati

Implementare funzioni predefinite nello script di immissione

Prendere in considerazione le impostazioni di automazione

Creare un processo in parallelo nella pipeline

Inviare il processo della pipeline e controllare il passaggio parallelo nell'interfaccia utente dello studio

Esempi di processi in parallelo nella pipeline

Passaggi successivi

Commenti e suggerimenti

Risorse aggiuntive