Aggiornare il passaggio di esecuzione parallela all'SDK v2

In SDK v2, "Passaggio di esecuzione parallela" viene consolidato nel concetto di processo come parallel job. Il processo parallelo mantiene la stessa destinazione per consentire agli utenti di accelerare l'esecuzione del processo distribuendo attività ripetute in cluster di calcolo multi-nodi potenti. Oltre al passaggio 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.
• Le funzionalità più potenti sono sviluppate solo in processi paralleli v2. Ad esempio, riprendere il processo parallelo non riuscito/annullato per continuare a elaborare i mini batch non elaborati o non elaborati riutilizzando il risultato riuscito per risparmiare 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. Per informazioni dettagliate sull'aggiornamento della pipeline, vedere come aggiornare la pipeline dalla versione 1 alla versione 2 .

Nota: lo script di immissione utente è compatibile tra il passaggio di esecuzione parallela v1 e il processo parallelo v2. È quindi possibile continuare a usare lo stesso entry_script.py quando si aggiorna il processo di esecuzione parallela.

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 una pipeline di Azure Machine Learning 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

  # 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-0.24-ubuntu18.04-py37-cpu:1",
      ),
  )
  

Usare il passaggio parallelo 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

Mapping di impostazioni e configurazioni dei processi paralleli

Passaggi successivi

Per altre informazioni, vedere la documentazione qui:

• Esempi di SDK per l'esecuzione parallela v1
• Esempi di SDK per processi paralleli v2