Definición de una canalización

1. Abra la consola de Studio siguiendo las instrucciones de Lanza Amazon SageMaker Studio.

2. En el panel de navegación izquierdo, seleccione Canalizaciones.

3. Seleccione Crear.

4. Seleccione En blanco.

5. En la barra lateral izquierda, elija Procesar datos y arrástrelos al lienzo.

6. En el lienzo, elija el paso Procesar datos que ha añadido.

7. Para añadir un conjunto de datos de entrada, seleccione Agregar en Datos (entrada) en la barra lateral derecha y seleccione un conjunto de datos.

8. Para añadir una ubicación para guardar los conjuntos de datos de salida, seleccione Agregar en Datos (salida) en la barra lateral derecha y vaya hasta el destino.

9. Complete el resto de los campos en la barra lateral derecha. Para obtener información sobre los campos de estas pestañas, consulte sagemaker.workflow.steps. ProcessingStep.

1. En la barra lateral izquierda, seleccione Entrenar modelo y arrástrelo al lienzo.

2. En el lienzo, elija el paso Entrenar modelo que ha añadido.

3. Para añadir un conjunto de datos de entrada, seleccione Agregar en Datos (entrada) en la barra lateral derecha y seleccione un conjunto de datos.

4. Para elegir una ubicación para guardar los artefactos del modelo, introduzca un URI de Amazon S3 en el campo Ubicación (URI de S3) o elija Examinar S3 para ir a la ubicación de destino.

5. Complete el resto de los campos en la barra lateral derecha. Para obtener información sobre los campos de estas pestañas, consulte sagemaker.workflow.steps. TrainingStep.

6. Haga clic y arrastre el cursor desde el paso Procesar datos que añadió en la sección anterior hasta el paso Entrenar modelo para crear un borde que conecte los dos pasos.

1. En la barra lateral izquierda, seleccione Registrar modelo y arrástrelo al lienzo.

2. En el lienzo, elija el paso Registrar modelo que ha añadido.

3. Para seleccionar un modelo al que anular el registro, elija Agregar en Modelo (entrada).

4. Seleccione Crear un grupo de modelos para añadir el modelo a un nuevo grupo de modelos.

5. Complete el resto de los campos en la barra lateral derecha. Para obtener información sobre los campos de estas pestañas, consulte sagemaker.workflow.step_collections. RegisterModel.

6. Haga clic y arrastre el cursor desde el paso Entrenar modelo que añadió en la sección anterior al paso Registrar modelo para crear un borde que conecte los dos pasos.

1. En la barra lateral izquierda, seleccione Implementar modelo (punto de conexión) y arrástrelo al lienzo.

2. En el lienzo, elija el paso Implementar modelo (punto de conexión) que ha añadido.

3. Para elegir un modelo para implementar, elija Agregar en Modelo (entrada).

4. Pulse el botón de opción Crear punto de conexión para crear un nuevo punto de conexión.

5. Escriba un Nombre y Descripción para el punto de conexión.

6. Haga clic y arrastre el cursor desde el paso Registrar modelo que añadió en la sección anterior al paso Implementar modelo (punto de conexión) para crear un borde que conecte los dos pasos.

7. Complete el resto de los campos en la barra lateral derecha.

Para descargar los conjuntos de datos

1. Descargue el conjunto de datos en el bucket de Amazon S3 predeterminado de su cuenta.

   !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. Descargue un segundo conjunto de datos para la transformación por lotes una vez creado el modelo.

   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)

Para crear un paso de procesamiento

1. Cree un directorio para el script de procesamiento.

   !mkdir -p abalone

2. En el directorio /abalone, cree un archivo denominado preprocessing.py con el contenido siguiente. Este script de procesamiento previo se transfiere al paso de procesamiento para ejecutarse en los datos de entrada. A continuación, el paso de entrenamiento usa las características y etiquetas de entrenamiento preprocesadas para entrenar un modelo. En el paso de evaluación, se usa el modelo entrenado y las características y etiquetas de prueba preprocesadas para evaluar el modelo. El script usa scikit-learn para hacer lo siguiente:

   • Rellenar los datos categóricos sex ausentes y codificarlos para que sean adecuados para el entrenamiento.

   • Escalar y normalizar todos los campos numéricos excepto rings y sex.

   • Dividir los datos en conjuntos de entrenamiento, validación y prueba.

   %%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. Cree una instancia de un SKLearnProcessor para transferirla al paso de procesamiento.

   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. Cree un paso de procesamiento. Este paso incluye SKLearnProcessor, los canales de entrada y salida y el script preprocessing.py que ha creado. Es muy similar al run método de una instancia de procesador en el SDK de Python para SageMaker IA. El parámetro input_data que se pasa a ProcessingStep son los datos de entrada del propio paso. La instancia del procesador utiliza estos datos de entrada cuando se ejecuta.

   Observe los canales denominados "train, "validation y "test" especificados en la configuración de salida para el trabajo de procesamiento. Las Properties del paso como estas pueden utilizarse en pasos posteriores y se resuelven a sus valores de tiempo de ejecución en el tiempo de ejecución.

   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
   )

Para definir un paso de entrenamiento

1. Especifique la ruta del modelo en la que desea guardar los modelos del entrenamiento.

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

2. Configura un estimador para el XGBoost algoritmo y el conjunto de datos de entrada. El tipo de instancia de entrenamiento se transfiere al estimador. Script de entrenamiento típico:

   • carga datos de los canales de entrada

   • configura el entrenamiento con hiperparámetros

   • entrena un modelo

   • guarda un modelo en model_dir poder alojarlo más adelante

   SageMaker La IA carga el modelo a Amazon S3 en forma de a model.tar.gz al final del trabajo de formación.

   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. Cree un TrainingStep mediante la instancia del estimador y las propiedades del ProcessingStep. Pase el S3Uri de los canales de salida "train" y "validation" al 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
   )

Para definir un paso de procesamiento para la evaluación del modelo

1. En el directorio /abalone cree un archivo denominado evaluation.py. Este script se utiliza en un paso de procesamiento para realizar la evaluación del modelo. Toma un modelo entrenado y el conjunto de datos de prueba como entrada y, a continuación, produce un archivo JSON que contiene las métricas de evaluación de la clasificación.

   %%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. Cree una instancia de un ScriptProcessor que se utilice para crear unProcessingStep.

   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. Cree un ProcessingStep utilizando la instancia del procesador, los canales de entrada y salida y el script evaluation.py. Pase lo siguiente:

   • la propiedad S3ModelArtifacts desde el paso de entrenamiento step_train

   • el S3Uri del canal de salida "test" del paso de procesamiento step_process

   Es muy similar al run método de una instancia de procesador en el SDK de Python para SageMaker IA. 

   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],
   )

Para definir una transformación CreateModelStep por lotes

1. Cree un modelo de SageMaker IA. Transfiera la propiedad S3ModelArtifacts desde el paso de entrenamiento 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. Defina la entrada del modelo para su modelo de SageMaker IA.

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

3. Cree la suya CreateModelStep con la CreateModelInput instancia del modelo de SageMaker IA que definió.

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

Para definir una transformación por lotes TransformStep para realizar

1. Cree una instancia de transformador con el tipo de instancia de cómputo, el recuento de instancias y el URI de bucket de Amazon S3 de salida adecuados. Transfiera la propiedad ModelName desde el paso CreateModel de step_create_model.

   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. Cree un TransformStep con la instancia de transformador que definió y el parámetro de canalización 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)
   )

Para definir un RegisterModel paso para crear un paquete modelo

• Construya un paso RegisterModel con la instancia del estimador que utilizó para el paso de entrenamiento. Transfiera la propiedad S3ModelArtifacts desde el paso de entrenamiento step_train y especifique un ModelPackageGroup. Canalizaciones crea este ModelPackageGroup por usted.

  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
  )

Para definir un paso de condición para verificar la precisión del modelo

1. Defina una condición ConditionLessThanOrEqualTo con el valor de precisión que se encuentra en al salida del paso de procesamiento de la evaluación del modelo, step_eval. Obtenga este resultado utilizando el archivo de propiedades que indexó en el paso de procesamiento y el valor JSONPath de error cuadrático medio, respectivamente. "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. Construya un ConditionStep. Pase la condición ConditionEquals y, a continuación, establezca los pasos de registro del paquete de modelos y transformación por lotes como los siguientes pasos si se cumple la condición.

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

Creación de una canalización

1. Defina lo siguiente para su canalización: name, parameters y steps. Los nombres deben ser únicos en un par (account, region).

   nota

   Un paso solo puede aparecer una vez en la lista de pasos de la canalización o en las listas de pasos if/else del paso de condición. No puede aparecer en ambas.

   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. De forma opcional, examine la definición de canalización de JSON para asegurarse de que esté bien formada.

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