Aggiornare il passaggio di esecuzione parallela a SDK v2

In SDK v2, "Passaggio di esecuzione parallela" viene consolidato nel concetto di processo come parallel job. Il job parallelo mantiene lo stesso obiettivo per permettere agli utenti di accelerare l'esecuzione del loro lavoro distribuendo le attività ripetute su potenti cluster di calcolo multi-nodo. Oltre alla fase di esecuzione parallela, il processo parallelo v2 offre vantaggi aggiuntivi.

• Interfaccia flessibile, che consente all'utente di definire più input e output personalizzati per il processo parallelo. È possibile connetterli con altri passaggi per utilizzare o gestire il contenuto nello script di immissione
• Semplificare lo schema di input, che sostituisce Dataset come input usando il concetto v2 data asset . È possibile usare facilmente i file locali o l'URI della directory BLOB come input per il processo parallelo.
• Funzionalità più avanzate disponibili solo nel processo parallelo v2. Ad esempio, riprendere un processo parallelo non riuscito o annullato per continuare l'elaborazione di mini batch non elaborati o non elaborati riutilizzando risultati riusciti, risparmiando il lavoro duplicato.

Per aggiornare il passaggio di esecuzione parallela dell'SDK v1 corrente alla versione 2, è necessario

• Usare parallel_run_function per creare un processo parallelo sostituendo ParallelRunConfig e ParallelRunStep in v1.
• Aggiornare la pipeline v1 alla versione 2. Richiamare quindi il processo parallelo v2 come passaggio nella pipeline v2. Vedere come aggiornare la pipeline dalla versione 1 alla versione 2 per ulteriori dettagli sull'aggiornamento della pipeline.

Questo articolo offre un confronto tra scenari in SDK v1 e SDK v2. Negli esempi seguenti verrà creato un processo parallelo per stimare i dati di input in un processo di pipeline. Si vedrà come creare un processo parallelo e come usarlo in un processo della pipeline per SDK v1 e SDK v2.

Prerequisiti

• Preparare l'ambiente SDK v2: Installare Azure Machine Learning SDK v2 per Python
• Informazioni sulla base della pipeline SDK v2: Come creare Azure Machine Learning pipeline con Python SDK v2

Creare un passaggio parallelo

• SDK v1

  # Create the configuration to wrap the inference script
  from azureml.pipeline.steps import ParallelRunStep, ParallelRunConfig
  
  parallel_run_config = ParallelRunConfig(
      source_directory=scripts_folder,
      entry_script=script_file,
      mini_batch_size=PipelineParameter(name="batch_size_param", default_value="5"),
      error_threshold=10,
      output_action="append_row",
      append_row_file_name="mnist_outputs.txt",
      environment=batch_env,
      compute_target=compute_target,
      process_count_per_node=PipelineParameter(name="process_count_param", default_value=2),
      node_count=2
  )
  
  # Create the Parallel run step
  parallelrun_step = ParallelRunStep(
      name="predict-digits-mnist",
      parallel_run_config=parallel_run_config,
      inputs=[ input_mnist_ds_consumption ],
      output=output_dir,
      allow_reuse=False
  )
  

• SDK v2

  # import required libraries
  from azure.ai.ml import Input, Output
  from azure.ai.ml.constants import AssetTypes, InputOutputModes
  from azure.ai.ml.parallel import parallel_run_function, RunFunction
  
  # parallel job 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,
      mini_batch_size="1",
      mini_batch_error_threshold=1,
      max_concurrency_per_instance=1,
      task=RunFunction(
          code="./src",
          entry_script="file_batch_inference.py",
          program_arguments="--job_output_path ${{outputs.job_output_path}}",
          environment="azureml:AzureML-sklearn-1.5-ubuntu22.04-py310-cpu@latest",
      ),
  )
  

Utilizzare la fase parallela nella pipeline

• SDK v1

  # Run pipeline with parallel run step
  from azureml.core import Experiment
  
  pipeline = Pipeline(workspace=ws, steps=[parallelrun_step])
  experiment = Experiment(ws, 'digit_identification')
  pipeline_run = experiment.submit(pipeline)
  pipeline_run.wait_for_completion(show_output=True)
  

• SDK v2

  @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"
  
  # run pipeline job
  pipeline_job = ml_client.jobs.create_or_update(
      pipeline_job, experiment_name="pipeline_samples"
  )
  

Mapping delle funzionalità chiave in SDK v1 e SDK v2

Mappatura delle configurazioni e impostazioni dei processi paralleli

Passaggi successivi

Per altre informazioni, vedere la documentazione qui:

• Esempi di "Parallel run step SDK v1"
• Esempi Parallel job SDK v2