Definizione di una pipeline

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

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. Crea una directory per lo script di elaborazione.

   !mkdir -p abalone

2. Nella directory /abalone, crea un file denominato preprocessing.py con il contenuto seguente. Questo script di pre-elaborazione viene passato alla fase di elaborazione per l’esecuzione sui dati di input. La fase di addestramento utilizza quindi le funzionalità e le etichette di addestramento pre-elaborate per addestrare un modello. La fase di valutazione utilizza il modello addestrato e le etichette di test pre-elaborate per valutare il modello. Lo script utilizza scikit-learn per effettuare quanto segue:

   • Inserire i dati categorici sex mancanti e codificarli in modo che siano adatti all’addestramento.

   • Ridimensionare e normalizzare tutti i campi numerici ad eccezione di rings e sex.

   • Suddividere i dati in set di dati di addestramento, test e convalida.

   %%writefile abalone/preprocessing.py
   import argparse
   import os
   import requests
   import tempfile
   import numpy as np
   import pandas as pd
   
   
   from sklearn.compose import ColumnTransformer
   from sklearn.impute import SimpleImputer
   from sklearn.pipeline import Pipeline
   from sklearn.preprocessing import StandardScaler, OneHotEncoder
   
   
   # Because this is a headerless CSV file, specify the column names here.
   feature_columns_names = [
       "sex",
       "length",
       "diameter",
       "height",
       "whole_weight",
       "shucked_weight",
       "viscera_weight",
       "shell_weight",
   ]
   label_column = "rings"
   
   feature_columns_dtype = {
       "sex": str,
       "length": np.float64,
       "diameter": np.float64,
       "height": np.float64,
       "whole_weight": np.float64,
       "shucked_weight": np.float64,
       "viscera_weight": np.float64,
       "shell_weight": np.float64
   }
   label_column_dtype = {"rings": np.float64}
   
   
   def merge_two_dicts(x, y):
       z = x.copy()
       z.update(y)
       return z
   
   
   if __name__ == "__main__":
       base_dir = "/opt/ml/processing"
   
       df = pd.read_csv(
           f"{base_dir}/input/abalone-dataset.csv",
           header=None, 
           names=feature_columns_names + [label_column],
           dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype)
       )
       numeric_features = list(feature_columns_names)
       numeric_features.remove("sex")
       numeric_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="median")),
               ("scaler", StandardScaler())
           ]
       )
   
       categorical_features = ["sex"]
       categorical_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="constant", fill_value="missing")),
               ("onehot", OneHotEncoder(handle_unknown="ignore"))
           ]
       )
   
       preprocess = ColumnTransformer(
           transformers=[
               ("num", numeric_transformer, numeric_features),
               ("cat", categorical_transformer, categorical_features)
           ]
       )
       
       y = df.pop("rings")
       X_pre = preprocess.fit_transform(df)
       y_pre = y.to_numpy().reshape(len(y), 1)
       
       X = np.concatenate((y_pre, X_pre), axis=1)
       
       np.random.shuffle(X)
       train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))])
   
       
       pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False)
       pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False)
       pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)

3. Creare un'istanza di un SKLearnProcessor da passare alla fase di elaborazione.

   from sagemaker.sklearn.processing import SKLearnProcessor
   
   
   framework_version = "0.23-1"
   
   sklearn_processor = SKLearnProcessor(
       framework_version=framework_version,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

4. Creare una fase di elaborazione. Questa fase comprende  SKLearnProcessor, i canali di input e output e lo script preprocessing.py che hai creato. È molto simile al metodo run dell’istanza di un processore in SageMaker AI Python SDK. Il parametro input_data passato a ProcessingStep è costituito dai dati di input della fase stessa. Questi dati di input vengono utilizzati dall'istanza del processore durante l'esecuzione.

   Notate "train i "validation canali con "test" nome e specificati nella configurazione di output per il processo di elaborazione. Le fasi Properties come queste possono essere utilizzate nelle fasi successive e si risolvono nei rispettivi valori di runtime al runtime.

   from sagemaker.processing import ProcessingInput, ProcessingOutput
   from sagemaker.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 strumento di valutazione per l'algoritmo XGBoost e il set di dati di input. Il tipo di istanza di addestramento viene passato allo strumento di valutazione. Uno script di addestramento tipico:

   • carica i dati dai canali di input

   • configura l’addestramento con iperparametri

   • addestra un modello

   • salva un modello in model_dir così che possa essere ospitato in un secondo momento

   SageMaker AI carica il modello su Amazon S3 sotto forma di model.tar.gz alla fine del job di addestramento.

   from sagemaker.estimator import Estimator
   
   
   image_uri = sagemaker.image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = Estimator(
       image_uri=image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=model_path,
       sagemaker_session=pipeline_session,
       role=role,
   )
   xgb_train.set_hyperparameters(
       objective="reg:linear",
       num_round=50,
       max_depth=5,
       eta=0.2,
       gamma=4,
       min_child_weight=6,
       subsample=0.7,
       silent=0
   )

3. Crea TrainingStep utilizzando l’istanza dello strumento di stima e le proprietà di ProcessingStep. In particolare, passa S3Uri di "train" e il canale di uscita "validation" a TrainingStep. 

   from sagemaker.inputs import TrainingInput
   from sagemaker.workflow.steps import TrainingStep
   
   
   train_args = xgb_train.fit(
       inputs={
           "train": TrainingInput(
               s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                   "train"
               ].S3Output.S3Uri,
               content_type="text/csv"
           ),
           "validation": TrainingInput(
               s3_data=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. Nella directory /abalone, crea un file denominato evaluation.py. Questo script viene usato in una fase di elaborazione per eseguire la valutazione del modello. Prende come input un modello addestrato e il set di dati di test, quindi produce un file JSON contenente le metriche di valutazione della classificazione.

   %%writefile abalone/evaluation.py
   import json
   import pathlib
   import pickle
   import tarfile
   import joblib
   import numpy as np
   import pandas as pd
   import xgboost
   
   
   from sklearn.metrics import mean_squared_error
   
   
   if __name__ == "__main__":
       model_path = f"/opt/ml/processing/model/model.tar.gz"
       with tarfile.open(model_path) as tar:
           tar.extractall(path=".")
       
       model = pickle.load(open("xgboost-model", "rb"))
   
       test_path = "/opt/ml/processing/test/test.csv"
       df = pd.read_csv(test_path, header=None)
       
       y_test = df.iloc[:, 0].to_numpy()
       df.drop(df.columns[0], axis=1, inplace=True)
       
       X_test = xgboost.DMatrix(df.values)
       
       predictions = model.predict(X_test)
   
       mse = mean_squared_error(y_test, predictions)
       std = np.std(y_test - predictions)
       report_dict = {
           "regression_metrics": {
               "mse": {
                   "value": mse,
                   "standard_deviation": std
               },
           },
       }
   
       output_dir = "/opt/ml/processing/evaluation"
       pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
       
       evaluation_path = f"{output_dir}/evaluation.json"
       with open(evaluation_path, "w") as f:
           f.write(json.dumps(report_dict))

2. Crea un'istanza di un ScriptProcessor che viene utilizzata per creare un ProcessingStep.

   from sagemaker.processing import ScriptProcessor
   
   
   script_eval = ScriptProcessor(
       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,
   )

3. Crea ProcessingStep utilizzando l’istanza del processore, i canali di input e output e lo script evaluation.py. Passa:

   • la proprietà S3ModelArtifacts dalla fase di addestramento step_train

   • S3Uri del canale di output "test" della fase di elaborazione step_process

   È molto simile al metodo run dell’istanza di un processore in SageMaker AI Python SDK. 

   from sagemaker.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 un CreateModelStep per la trasformazione in batch

1. Crea un modello SageMaker AI. Passa alla proprietà S3ModelArtifacts sin dalla fase di addestramento step_train.

   from sagemaker.model import Model
   
   
   model = Model(
       image_uri=image_uri,
       model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Definisci l’input del modello per il modello SageMaker AI.

   from sagemaker.inputs import CreateModelInput
   
   
   inputs = CreateModelInput(
       instance_type="ml.m5.large",
       accelerator_type="ml.eia1.medium",
   )

3. Crea CreateModelStep utilizzando CreateModelInput e l’istanza del modello SageMaker AI che hai definito.

   from sagemaker.workflow.steps import CreateModelStep
   
   
   step_create_model = CreateModelStep(
       name="AbaloneCreateModel",
       model=model,
       inputs=inputs,
   )

Per definire un TransformStep per eseguire la trasformazione in batch

1. Crea un'istanza Transformer con il tipo di istanza di calcolo appropriato, il numero di istanze e l'URI del bucket Amazon S3 di output desiderati. Passa alla proprietà ModelName sin dalla fase step_create_model CreateModel.

   from sagemaker.transformer import Transformer
   
   
   transformer = Transformer(
       model_name=step_create_model.properties.ModelName,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=f"s3://{default_bucket}/AbaloneTransform"
   )

2. Crea un TransformStep utilizzando l'istanza del transformer che hai definito e il parametro pipeline batch_data.

   from sagemaker.inputs import TransformInput
   from sagemaker.workflow.steps import TransformStep
   
   
   step_transform = TransformStep(
       name="AbaloneTransform",
       transformer=transformer,
       inputs=TransformInput(data=batch_data)
   )

Per definire una fase RegisterModel per creare un pacchetto di modelli

• Costruisci una fase RegisterModel utilizzando l'istanza dello strumento di valutazione che hai usato per la fase di addestramento. Passa alla proprietà S3ModelArtifacts sin dalla fase di addestramento step_train e specifica un ModelPackageGroup. Pipelines crea ModelPackageGroup per te.

  from sagemaker.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.workflow.step_collections import RegisterModel
  
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  step_register = RegisterModel(
      name="AbaloneRegisterModel",
      estimator=xgb_train,
      model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      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
  )

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

1. Definisci una condizione ConditionLessThanOrEqualTo utilizzando il valore di accuratezza trovato nell'output della fase di elaborazione della valutazione del modello, step_eval. Ottieni questo output utilizzando il file delle proprietà che hai indicizzato nella fase di elaborazione e il rispettivo JSONPath del valore di errore quadratico medio, "mse".

   from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
   from sagemaker.workflow.condition_step import ConditionStep
   from sagemaker.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
   )

2. Costruisci un ConditionStep. Passa la condizione ConditionEquals, quindi imposta le fasi di registrazione del pacchetto di modelli e di trasformazione in batch come fasi successive, se la condizione viene soddisfatta.

   step_cond = ConditionStep(
       name="AbaloneMSECond",
       conditions=[cond_lte],
       if_steps=[step_register, step_create_model, step_transform],
       else_steps=[], 
   )

Per creare una pipeline

1. Definisci quanto segue per la tua pipeline: name, parameters e steps. I nomi devono essere univoci all'interno di una coppia (account, region).

   Nota

   Una fase può apparire solo una volta nell'elenco delle fasi della pipeline o negli elenchi delle fasi if/else della fase condizionale. Non può apparire in entrambi.

   from sagemaker.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())