Definizione di una pipeline

1. Apri la console Studio seguendo le istruzioni riportate in Avvia Amazon Studio SageMaker.

2. Nel riquadro di navigazione a sinistra, seleziona Pipeline.

3. Scegli Create (Crea).

4. Scegli Vuoto.

5. Nella barra laterale sinistra, scegli Elaborazione di dati e trascinalo sul canvas.

6. Nel canvas, scegli la fase Elaborazione di dati che hai aggiunto.

7. Per aggiungere un set di dati di input, scegli Aggiungi in Dati (input) nella barra laterale destra e seleziona un set di dati.

8. Per aggiungere una posizione in cui salvare i set di dati di output, scegli Aggiungi in Dati (output) nella barra laterale destra e vai alla destinazione.

9. Completa i campi rimanenti nella barra laterale destra. Per informazioni sui campi di queste schede, consulta sagemaker.workflow.steps. ProcessingStep.

1. Nella barra laterale sinistra, scegli Addestramento di un modello e trascinalo sul canvas.

2. Nel canvas, scegli la fase Addestramento di un modello che hai aggiunto.

3. Per aggiungere un set di dati di input, scegli Aggiungi in Dati (input) nella barra laterale destra e seleziona un set di dati.

4. Per scegliere una posizione in cui salvare gli artefatti del modello, inserisci un URI Amazon S3 nel campo Posizione (URI S3) oppure scegli Sfoglia S3 per passare alla posizione di destinazione.

5. Completa i campi rimanenti nella barra laterale destra. Per informazioni sui campi di queste schede, consulta sagemaker.workflow.steps. TrainingStep.

6. Fai clic e trascina il cursore dalla fase Elaborazione di dati che hai aggiunto nella sezione precedente alla fase Addestramento di un modello per creare un arco che colleghi le due fasi.

1. Nella barra laterale sinistra, scegli Registrazione di un modello e trascinalo sul canvas.

2. Nel canvas, scegli la fase Registrazione di un modello che hai aggiunto.

3. Per selezionare un modello da registrare, scegli Aggiungi in Modello (input).

4. Scegli Crea un gruppo di modelli per aggiungere il modello a un nuovo gruppo di modelli.

5. Completa i campi rimanenti nella barra laterale destra. Per informazioni sui campi di queste schede, consulta sagemaker.workflow.step_collections. RegisterModel.

6. Fai clic e trascina il cursore dalla fase Addestramento di un modello che hai aggiunto nella sezione precedente alla fase Registrazione di un modello per creare un arco che colleghi le due fasi.

1. Nella barra laterale sinistra, scegli Implementazione di un modello (endpoint) e trascinalo sul canvas.

2. Nel canvas, scegli la fase Implementazione di un modello (endpoint) che hai aggiunto.

3. Per scegliere un modello da implementare, scegli Aggiungi in Modello (input).

4. Scegli il pulsante di opzione Crea endpoint per creare un nuovo endpoint.

5. Immetti un nome e una descrizione per l’endpoint.

6. Fai clic e trascina il cursore dalla fase Registrazione di un modello che hai aggiunto nella sezione precedente alla fase Implementazione di un modello (endpoint) per creare un arco che colleghi le due fasi.

7. Completa i campi rimanenti nella barra laterale destra.

Per scaricare i set di dati

1. Scarica il set di dati nel bucket Amazon S3 predefinito del tuo account.

   !mkdir -p data
   local_path = "data/abalone-dataset.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset.csv",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   input_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(input_data_uri)

2. Scarica un secondo set di dati per la trasformazione in batch dopo la creazione del modello.

   local_path = "data/abalone-dataset-batch.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset-batch",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   batch_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(batch_data_uri)

Per creare una fase di elaborazione

1. Create un'istanza di un processore da passare alla fase di elaborazione.

   from sagemaker.core.processing import Processor
   from sagemaker.core import image_uris
   
   framework_version = "0.23-1"
   
   sklearn_image_uri = image_uris.retrieve(
       framework="sklearn",
       region=region,
       version=framework_version,
       instance_type="ml.m5.xlarge"
   )
   
   sklearn_processor = Processor(
       image_uri=sklearn_image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Creare una fase di elaborazione.

   from sagemaker.core.processing import ProcessingInput, ProcessingOutput
   from sagemaker.mlops.workflow.steps import ProcessingStep
   
   processor_args = sklearn_processor.run(
       inputs=[
         ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),  
       ],
       outputs=[
           ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
           ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
           ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
       ],
       code="abalone/preprocessing.py",
   ) 
   
   step_process = ProcessingStep(name="AbaloneProcess", step_args=processor_args)

Per definire una fase di addestramento

1. Specifica il percorso del modello in cui desideri salvare i modelli dall'addestramento.

   model_path = f"s3://{default_bucket}/AbaloneTrain"

2. Configura uno stimatore di addestramento per l'algoritmo XGBoost e il set di dati di input.

   from sagemaker.train import ModelTrainer
   from sagemaker.train.configs import Compute, OutputDataConfig
   from sagemaker.core import image_uris
   
   image_uri = image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = ModelTrainer(
       training_image=image_uri,
       compute=Compute(instance_type="ml.m5.xlarge", instance_count=1),
       output_data_config=OutputDataConfig(s3_output_path=model_path),
       hyperparameters={
           "objective": "reg:linear",
           "num_round": "50",
           "max_depth": "5",
           "eta": "0.2",
           "gamma": "4",
           "min_child_weight": "6",
           "subsample": "0.7",
           "silent": "0"
       },
       sagemaker_session=pipeline_session,
       role=role,
   )

3. Crea un TrainingStep utilizzando l'istanza di addestramento e le proprietà di. ProcessingStep

   from sagemaker.train.configs import InputData
   from sagemaker.mlops.workflow.steps import TrainingStep
   
   train_args = xgb_train.train(
       input_data_config=[
           InputData(
               channel_name="train",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "train"
               ].S3Output.S3Uri,
               content_type="text/csv"
           ),
           InputData(
               channel_name="validation",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "validation"
               ].S3Output.S3Uri,
               content_type="text/csv"
           )
       ],
   )
   
   step_train = TrainingStep(name="AbaloneTrain", step_args=train_args)

Per definire una fase di elaborazione per la valutazione del modello

1. Crea un'istanza di a Processor per la valutazione.

   from sagemaker.core.processing import Processor
   
   script_eval = Processor(
       image_uri=image_uri,
       command=["python3"],
       instance_type="ml.m5.xlarge",
       instance_count=1,
       base_job_name="script-abalone-eval",
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Createne una ProcessingStep utilizzando l'istanza del processore, i canali di input e output e lo evaluation.py script.

   from sagemaker.core.workflow.properties import PropertyFile
   
   evaluation_report = PropertyFile(
       name="EvaluationReport",
       output_name="evaluation",
       path="evaluation.json"
   )
   
   eval_args = script_eval.run(
       inputs=[
           ProcessingInput(
               source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
               destination="/opt/ml/processing/model"
           ),
           ProcessingInput(
               source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "test"
               ].S3Output.S3Uri,
               destination="/opt/ml/processing/test"
           )
       ],
       outputs=[
           ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"),
       ],
       code="abalone/evaluation.py",
   )
   
   step_eval = ProcessingStep(
       name="AbaloneEval",
       step_args=eval_args,
       property_files=[evaluation_report],
   )

Per definire una CreateModelStep trasformazione in batch

• Crea un modello SageMaker AI e una fase del modello.

  from sagemaker.serve import ModelBuilder
  from sagemaker.mlops.workflow.model_step import ModelStep
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  step_create_model = ModelStep(
      name="AbaloneCreateModel",
      step_args=model_builder.build(),
  )

Definire una trasformazione TransformStep in batch da eseguire

• Crea un'istanza del trasformatore e unTransformStep.

  from sagemaker.core.transformer import Transformer
  from sagemaker.mlops.workflow.steps import TransformStep
  
  transformer = Transformer(
      model_name=step_create_model.properties.ModelName,
      instance_type="ml.m5.xlarge",
      instance_count=1,
      output_path=f"s3://{default_bucket}/AbaloneTransform",
      sagemaker_session=pipeline_session,
  )
  
  transform_args = transformer.transform(data=batch_data)
  
  step_transform = TransformStep(
      name="AbaloneTransform",
      step_args=transform_args,
  )

Per definire una fase di registrazione del modello per creare un pacchetto di modelli

• Costruisci una fase di registrazione del modello.

  from sagemaker.core.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.mlops.workflow.model_step import ModelStep
  from sagemaker.serve import ModelBuilder
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  register_args = model_builder.register(
      content_types=["text/csv"],
      response_types=["text/csv"],
      inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
      transform_instances=["ml.m5.xlarge"],
      model_package_group_name=model_package_group_name,
      approval_status=model_approval_status,
      model_metrics=model_metrics
  )
  
  step_register = ModelStep(
      name="AbaloneRegisterModel",
      step_args=register_args,
  )

Per definire una fase condizionale per verificare l'accuratezza del modello

• Definisci una condizione e costruisci una. ConditionStep

  from sagemaker.core.workflow.conditions import ConditionLessThanOrEqualTo
  from sagemaker.mlops.workflow.condition_step import ConditionStep
  from sagemaker.core.workflow.functions import JsonGet
  
  cond_lte = ConditionLessThanOrEqualTo(
      left=JsonGet(
          step_name=step_eval.name,
          property_file=evaluation_report,
          json_path="regression_metrics.mse.value"
      ),
      right=6.0
  )
  
  step_cond = ConditionStep(
      name="AbaloneMSECond",
      conditions=[cond_lte],
      if_steps=[step_register, step_create_model, step_transform],
      else_steps=[], 
  )

Come creare una pipeline

1. Definisci quanto segue per la tua pipeline: name, parameters e steps.

   from sagemaker.mlops.workflow.pipeline import Pipeline
   
   pipeline_name = f"AbalonePipeline"
   pipeline = Pipeline(
       name=pipeline_name,
       parameters=[
           processing_instance_count,
           model_approval_status,
           input_data,
           batch_data,
       ],
       steps=[step_process, step_train, step_eval, step_cond],
   )

2. (Facoltativo) Esamina la definizione della pipeline JSON per assicurarti che sia ben formata.

   import json
   
   json.loads(pipeline.definition())