Comment utiliser un travail parallèle dans le pipeline (V2)

Article
04/04/2023

S’APPLIQUE À :Extension Azure CLI v2 (actuelle)Kit de développement logiciel (SDK) Python azure-ai-ml v2 (version actuelle)

Le travail parallèle permet aux utilisateurs d’accélérer l’exécution de leur travail en distribuant des tâches répétées sur de puissants clusters de calcul à plusieurs nœuds. Par exemple, prenez le scénario dans lequel vous exécutez un modèle de détection d’objet sur un grand ensemble d’images. Avec le travail parallèle Azure Machine Learning, vous pouvez facilement distribuer vos images pour exécuter du code personnalisé en parallèle sur un cluster de calcul spécifique. La parallélisation peut réduire considérablement le coût dans la durée. En outre, en utilisant un travail parallèle Azure Machine Learning, vous pouvez simplifier et automatiser votre processus pour le rendre plus efficace.

Prérequis

Le travail parallèle Azure Machine Learning ne peut être utilisé qu’en tant qu’une des étapes d’un travail de pipeline. Il est donc important d’être familiarisé avec l’utilisation de pipelines. Pour en savoir plus sur Azure Machine Learning, consultez les articles suivants.

Comprendre ce qu’est un pipeline Azure Machine Learning
Découvrez comment utiliser le pipeline Azure Machine Learning avec CLI v2 et SDK v2.

Pourquoi les travaux parallèles sont-ils nécessaires ?

Dans le monde réel, les ingénieurs ML ont toujours des exigences de mise à l’échelle sur leurs tâches d’entraînement ou d’inférence. Par exemple, lorsqu’un scientifique des données fournit un script unique pour entraîner un modèle de prédiction des ventes, les ingénieurs ML doivent appliquer cette tâche de formation à chaque magasin individuel. Pendant ce processus de scale-out, certains défis sont les suivants :

Pression différée causée par un temps d’exécution long.
Intervention manuelle pour gérer les problèmes inattendus afin de maintenir la tâche en cours.

La valeur principale du travail parallèle Azure Machine Learning consiste à fractionner une seule tâche série en mini-lots et à distribuer ces mini-lots à plusieurs calculs pour qu’ils soient exécutés en parallèle. En utilisant des travaux parallèles, nous pouvons :

Réduire considérablement le temps d’exécution de bout en bout.
Utiliser les paramètres de gestion automatique des erreurs du travail parallèle Azure Machine Learnin.

Vous devez envisager d’utiliser le travail parallèle Azure Machine Learning si :

Vous envisagez d’entraîner de nombreux modèles en plus de vos données partitionnés.
Vous souhaitez accélérer votre tâche d’inférence de lot à grande échelle.

Préparer un travail parallèle

Contrairement à d’autres types de travaux, un travail parallèle nécessite une préparation. Suivez les sections suivantes pour préparer la création de votre travail parallèle.

Déclarer les entrées à distribuer et le paramétrage de la division des données

Le travail parallèle nécessite qu’une seule donnée d’entrée majeure soit fractionnée et traitée avec parallèle. Les données d’entrée principales peuvent être des données tabulaires ou un ensemble de fichiers. Différents types d’entrée peuvent avoir une méthode de division des données

Le tableau suivant illustre la relation entre les données d’entrée et la méthode de division des données :

Format de données	Type d’entrée Azure Machine Learning	Mode d’entrée Azure Machine Learning	Méthode de division des données
Liste de fichiers	`mltable` ou `uri_folder`	ro_mount ou télécharger	Par taille (nombre de fichiers) Par partition
Données tabulaires	`mltable`	directes	Par taille (taille physique estimée) Par partition

Vous pouvez déclarer vos données d’entrée principales avec l’attribut input_data dans le Kit de développement logiciel (SDK) YAML ou Python du travail parallèle. Et vous pouvez le lier à l’une de vos inputs définies de votre travail parallèle en utilisant ${{inputs.<input name>}}. Ensuite, vous devez définir la méthode de division des données pour votre entrée principale en renseignant un attribut différent :

Méthode de division des données	Nom de l’attribut	Type d'attribut	Exemple de travail
Par taille	mini_batch_size	string	Prédiction par lots Iris
Par partition	partition_keys	Liste de chaînes	Prédiction des ventes de jus d’orange

Azure CLI
Python

S’APPLIQUE À :Extension Azure ML CLI v2 (actuelle)

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

S’APPLIQUE À : Kit de développement logiciel (SDK) Python azure-ai-mlv2 (préversion)

Déclarez job_data_path comme l’une des entrées. Liez-le à l’attribut 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:azureml-sklearn-1.0:23",
    ),
)

Une fois que vous avez défini le paramètre de division des données, vous pouvez configurer le nombre de ressources pour votre parallélisation en indiquant les deux attributs ci-dessous :

Nom de l’attribut	Type	Description	Valeur par défaut
`instance_count`	entier	Nombre de nœuds à dédier au travail.	1
`max_concurrency_per_instance`	entier	Nombre de processeurs sur chaque nœud.	Pour un calcul GPU, la valeur par défaut est 1. Pour un calcul CPU, la valeur par défaut est le nombre de cœurs.

Ces deux attributs fonctionnent avec votre cluster de calcul spécifié.

Exemple de code pour définir deux attributs :

Azure CLI
Python

S’APPLIQUE À :Extension Azure ML CLI v2 (actuelle)

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

S’APPLIQUE À : Kit de développement logiciel (SDK) Python azure-ai-mlv2 (préversion)

# 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:azureml-sklearn-1.0:23",
    ),
)

Notes

Si vous utilisez tabulaire mltable comme données d’entrée principales, vous devez disposer du fichier de spécification MLTABLE avec la section transformations - read_delimited remplie sous votre chemin d’accès spécifique. Pour plus d’exemples, consultez Créer une ressource de données mltable

Implémenter des fonctions prédéfinies dans le script d’entrée

Le script d’entrée est un fichier Python unique dans lequel l’utilisateur doit implémenter trois fonctions prédéfinies avec du code personnalisé. Le travail parallèle Azure Machine Learning suit le diagramme ci-dessous pour les exécuter dans chaque processeur.

Nom de la fonction	Obligatoire	Description	Entrée	Renvoie
Init()	O	Utilisez cette fonction pour la préparation courante avant de commencer à exécuter des mini-lots. Par exemple, utilisez-la pour charger le modèle dans un objet global.	--	--
Run(mini_batch)	O	Implémentez la logique d’exécution principale pour mini_batches.	mini_batch : Trame de données Pandas si les données d’entrée sont des données tabulaires. Liste du chemin d’accès au fichier si les données d’entrée sont un répertoire.	Trame de données, liste ou tuple.
Shutdown()	N	Fonction facultative pour effectuer un nettoyage personnalisé avant de retourner le calcul au pool.	--	--

Consultez les exemples de script d’entrée suivants pour obtenir plus de détails :

Une fois que le script d’entrée est prêt, vous pouvez définir les deux attributs suivants pour l’utiliser dans votre travail parallèle :

Nom de l’attribut	Type	Description	Valeur par défaut
`code`	string	Chemin d’accès local au répertoire du code source à télécharger et à utiliser pour le travail.
`entry_script`	string	Fichier Python qui contient l’implémentation de fonctions parallèles prédéfinies.

Exemple de code pour définir deux attributs :

Azure CLI
Python

S’APPLIQUE À :Extension Azure ML CLI v2 (actuelle)

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}}

S’APPLIQUE À : Kit de développement logiciel (SDK) Python azure-ai-mlv2 (préversion)

# 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:azureml-sklearn-1.0:23",
    ),
)

Important

La fonction Run(mini_batch) nécessite un retour d’un élément de trame de données, de liste ou de tuple. Le travail parallèle utilise le nombre de ce retour pour mesurer les éléments de réussite sous ce mini-lot. Idéalement, le nombre de mini-lots doit être égal au nombre de listes de retour si tous les éléments ont été bien traités dans ce mini-lot.

Important

Si vous souhaitez analyser des arguments dans Init() ou Run(mini_batch), utilisez « parse_known_args » au lieu de « parse_args » pour éviter les exceptions. Consultez l’exemple iris_score pour le script d’entrée avec l’analyseur d’arguments.

Important

Si vous utilisez mltable comme données d’entrée principales, vous devez installer la bibliothèque « mltable » dans votre environnement. Consultez la ligne 9 de cet exemple de fichier conda.

Prendre en compte les paramètres d’automatisation

Le travail parallèle Azure Machine Learning expose de nombreux paramètres pour contrôler automatiquement le travail sans intervention manuelle. Pour plus de détails, consultez le tableau suivant.

Clé	Type	Description	Valeurs autorisées	Valeur par défaut	Définir dans l’attribut	Définir dans les arguments du programme
seuil d’erreur du mini-lot	entier	Définissez le nombre de mini-lots ayant échoué qui peuvent être ignorés dans ce travail parallèle. Si le nombre de mini-lots ayant échoué est supérieur à ce seuil, le travail parallèle est marqué comme ayant échoué. Le mini-lot est marqué comme ayant échoué si : - le nombre de retours de run() est inférieur au nombre d’entrées par mini-lot. - des exceptions ont été interceptées dans le code run() personnalisé. « -1 » est le nombre par défaut, ce qui signifie ignorer tous les mini-lots ayant échoué pendant le travail parallèle.	[-1, int.max]	-1	mini_batch_error_threshold	N/A
nouvelles tentatives max. de mini-lots	entier	Définissez le nombre de nouvelles tentatives lors de l’échec ou du délai d’expiration du mini-lot. Si toutes les nouvelles tentatives ont échoué, le mini-lot est marqué comme ayant échoué par calcul `mini_batch_error_threshold`.	[0, int.max]	2	retry_settings.max_retries	N/A
délai d’expiration du mini-lot	entier	Définissez le délai d’expiration en secondes pour l’exécution de la fonction run() personnalisée. Si le temps d’exécution est supérieur à ce seuil, le mini-lot est abandonné et marqué comme un mini-lot ayant échoué pour déclencher une nouvelle tentative.	(0, 259200]	60	retry_settings.timeout	N/A
seuil d'erreur de l’élément	entier	Le seuil des éléments ayant échoué. Les éléments ayant échoué sont comptabilisés par l’écart de nombre entre les entrées et les retours de chaque mini-lot. Si le nombre de mini-lots ayant échoué est supérieur à ce seuil, le travail parallèle est marqué comme ayant échoué. Remarque : « -1 » est le nombre par défaut, ce qui signifie ignorer tous les mini-lots ayant échoué pendant le travail parallèle.	[-1, int.max]	-1	N/A	--error_threshold
pourcentage d’échec autorisé	entier	Similaire à `mini_batch_error_threshold` mais utilise le pourcentage de mini-lots ayant échoué au lieu du nombre.	[0, 100]	100	N/A	--allowed_failed_percent
expiration du délai d’attente	entier	Délai d’expiration en seconde pour l’initialisation de chaque mini-lot. Par exemple, chargez des données mini-batch et transmettez-la à la fonction run().	(0, 259200]	600	N/A	--task_overhead_timeout
délai d’expiration de la mise à jour de progression	entier	Délai d’expiration en seconde pour surveiller la progression de l’exécution de mini-lots. Si aucune mise à jour de progression n’est reçue dans ce paramètre de délai d’expiration, le travail parallèle est marqué comme ayant échoué.	(0, 259200]	Calculé dynamiquement par d’autres paramètres.	N/A	--progress_update_timeout
dépassement du délai de création de la première tâche	entier	Le délai en secondes pour contrôler le temps entre le début du travail et l'exécution du premier mini-lot.	(0, 259200]	600	N/A	--first_task_creation_timeout
niveau de journalisation	string	Définissez quel niveau de journaux sera sauvegardé vers les fichiers journaux utilisateur.	INFORMATIONS, AVERTISSEMENT ou DÉBOGAGE	INFO	Niveau de journalisation	N/A
ajouter une ligne à	string	Agrégez tous les retours de chaque exécution de mini-lot et sortez-le dans ce fichier. Peut faire référence à l’une des sorties du travail parallèle à l’aide de l’expression ${{outputs.<>output_name}}			task.append_row_to	N/A
copier des journaux d’activité vers le parent	string	Option booléenne permettant de copier la progression, la vue d’ensemble et les journaux du travail vers le pipeline parent.	True ou False	Faux	N/A	--copy_logs_to_parent
intervalle du moniteur de ressources	entier	Intervalle de temps en secondes pour sauvegarder l’utilisation des ressources de nœud (par exemple, processeur, mémoire) dans le dossier journal sous le chemin « logs/sys/perf ». Remarque : les journaux de ressources de sauvegarde fréquents ralentiront légèrement la vitesse d’exécution de votre mini-lot. Définissez cette valeur sur « 0 » pour arrêter la sauvegarde de l’utilisation des ressources.	[0, int.max]	600	N/A	--resource_monitor_interval

Exemple de code pour mettre à jour ces paramètres :

Azure CLI
Python

S’APPLIQUE À :Extension Azure ML CLI v2 (actuelle)

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}}

S’APPLIQUE À : Kit de développement logiciel (SDK) Python azure-ai-mlv2 (préversion)

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

Créer un travail parallèle dans le pipeline

Azure CLI
Python

S’APPLIQUE À :Extension Azure ML CLI v2 (actuelle)

Vous pouvez créer votre travail parallèle en ligne avec votre travail de 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}}

S’APPLIQUE À : Kit de développement logiciel (SDK) Python azure-ai-ml v2 (actuelle)

Tout d’abord, vous devez importer les bibliothèques requises, lancer votre ml_client avec les informations d’identification appropriées et créer/récupérer vos calculs :

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

Ensuite, implémentez votre travail parallèle en remplissant 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}}",
    ),
)

Enfin, utilisez votre travail parallèle comme étape dans votre pipeline et liez ses entrées/sorties avec d’autres étapes :

@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"

Envoyer un travail de pipeline et vérifier l’étape parallèle dans l’interface utilisateur de Studio

Azure CLI
Python

S’APPLIQUE À :Extension Azure ML CLI v2 (actuelle)

Vous pouvez soumettre votre travail de pipeline à l’aide de la commande CLI :

az ml job create --file pipeline.yml

S’APPLIQUE À : Kit de développement logiciel (SDK) Python azure-ai-ml v2 (actuelle)

Vous pouvez soumettre votre travail de pipeline à l’aide de l’étape parallèle en utilisant la fonction jobs.create_or_update de ml_client :

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

Une fois que vous avez envoyé votre travail de pipeline, le widget SDK ou CLI vous donnera un lien URL web vers l’interface utilisateur de Studio. Le lien vous guidera vers l’affichage graphique de pipeline par défaut. Double-sélectionnez l’étape parallèle pour ouvrir le volet droit de votre travail parallèle.

Pour vérifier les paramètres de votre travail parallèle, accédez à l’onglet Paramètres, développez Paramètres d’exécution et vérifiez la section Parallèle :

Pour déboguer l’échec de votre travail parallèle, accédez à l’onglet Sorties + Journaux, développez le dossier des journaux à partir des répertoires de sortie à gauche et vérifiez job_result.txt pour comprendre pourquoi le travail parallèle a échoué. Pour plus d’informations sur la structure de journalisation du travail parallèle, consultez le readme.txt sous le même dossier.

Travail parallèle dans des exemples de pipeline

Étapes suivantes

Pour obtenir le schéma yaml détaillé du travail parallèle, consultez la référence YAML pour le travail parallèle.
Pour savoir comment intégrer vos données à MLTABLE, consultez Créer une ressource de données mltable.
Pour savoir comment déclencher régulièrement votre pipeline, consultez Comment planifier le pipeline.

Partager via

Comment utiliser un travail parallèle dans le pipeline (V2)

Prérequis

Pourquoi les travaux parallèles sont-ils nécessaires ?

Préparer un travail parallèle

Déclarer les entrées à distribuer et le paramétrage de la division des données

Implémenter des fonctions prédéfinies dans le script d’entrée

Prendre en compte les paramètres d’automatisation

Créer un travail parallèle dans le pipeline

Envoyer un travail de pipeline et vérifier l’étape parallèle dans l’interface utilisateur de Studio

Travail parallèle dans des exemples de pipeline

Étapes suivantes

Ressources supplémentaires