Definisci Amazon SageMaker Pipelines

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

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

3. Scegli Create (Crea) .

4. Scegli Vuoto.

5. Nella barra laterale sinistra, scegli Elabora dati e trascinali nell'area di disegno.

6. Nell'area di disegno, scegli la fase Elabora dati che hai aggiunto.

7. Per aggiungere un set di dati di input, scegli Aggiungi sotto 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 sotto 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 Train model e trascinalo sull'area di disegno.

2. Nell'area di disegno, scegli la fase del modello Train che hai aggiunto.

3. Per aggiungere un set di dati di input, scegli Aggiungi sotto 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 Amazon URI S3 nel campo Posizione (URIS3) o scegli Browse S3 per accedere 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. Fate clic e trascinate il cursore dalla fase Process data che avete aggiunto nella sezione precedente alla fase Train model per creare uno spigolo che colleghi i due passaggi.

1. Nella barra laterale sinistra, scegli Registra modello e trascinalo nell'area di disegno.

2. Nell'area di disegno, scegli il passaggio Registra modello che hai aggiunto.

3. Per selezionare un modello da registrare, scegliete Aggiungi sotto Modello (input).

4. Scegliete 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. Fate clic e trascinate il cursore dalla fase del modello Train che avete aggiunto nella sezione precedente alla fase Registra modello per creare uno spigolo che colleghi i due gradini.

1. Nella barra laterale sinistra, scegli Deploy model (endpoint) e trascinalo nell'area di disegno.

2. Nell'area di disegno, scegli il passaggio Deploy model (endpoint) che hai aggiunto.

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

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

5. Inserisci un nome e una descrizione per il tuo endpoint.

6. Fate clic e trascinate il cursore dal passaggio Registra modello aggiunto nella sezione precedente al passo Deploy model (endpoint) per creare uno spigolo che colleghi i due passaggi.

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 preelaborazione viene passato alla fase di elaborazione per essere eseguito sui dati di input. La fase di addestramento utilizza quindi le funzionalità e le etichette di addestramento preelaborate per addestrare un modello. La fase di valutazione utilizza il modello addestrato e le funzionalità e le etichette dei test preelaborati 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. Questo è molto simile al run metodo di un'istanza del processore in SageMaker PythonSDK. 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. Fasi Properties come queste possono essere utilizzate nelle fasi successive e risolte nei rispettivi valori di runtime in fase di esecuzione.

   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 stimatore per l'XGBoostalgoritmo e il set di dati di input. Il tipo di istanza di addestramento viene passato allo strumento di valutazione. Un tipico script di allenamento:

   • carica i dati dai canali di input

   • configura l'allenamento con iperparametri

   • addestra un modello

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

   SageMaker carica il modello su Amazon S3 sotto forma di model.tar.gz un processo di formazione alla fine del processo di formazione.

   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 un TrainingStep utilizzando l'istanza estimator e le proprietà di. ProcessingStep Passa il S3Uri canale "train" e "validation" di uscita alTrainingStep. 

   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 JSON file 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 un file ProcessingStep utilizzando l'istanza del processore, i canali di input e output e lo evaluation.py script. Passa:

   • la S3ModelArtifacts proprietà fin dalla fase step_train di formazione

   • il canale S3Uri di "test" uscita della fase di step_process lavorazione

   Questo è molto simile al run metodo di un'istanza del processore in SageMaker PythonSDK. 

   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 una trasformazione CreateModelStep in batch

1. Crea un SageMaker modello. 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. Definite l'input del modello per il vostro SageMaker modello.

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

3. Crea il tuo CreateModelStep utilizzando l'istanza CreateModelInput e il SageMaker modello che hai definito.

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

Definire una TransformStep trasformazione in batch

1. Crea un'istanza di trasformazione con il tipo di istanza di calcolo, il numero di istanze e il bucket Amazon S3 di output desiderati. URI 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 RegisterModel fase per la creazione di un pacchetto modello

• 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 lo 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 risultato utilizzando il file delle proprietà che hai indicizzato nella fase di elaborazione e il rispettivo valore dell'errore JSONPath 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 JSON pipeline per assicurarti che sia ben formata.

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