Defina una canalización

1. Abra la consola de Studio siguiendo las instrucciones que se indican enLanza Amazon SageMaker Studio.

2. En el panel de navegación izquierdo, selecciona Pipelines.

3. Seleccione Crear.

4. Seleccione En blanco.

5. En la barra lateral izquierda, selecciona Procesar datos y arrástralos al lienzo.

6. En el lienzo, elige el paso Procesar datos que has añadido.

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

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

9. Complete los campos restantes 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, selecciona Modelo de tren y arrástralo al lienzo.

2. En el lienzo, elige el paso del modelo de tren que has añadido.

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

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

5. Complete los campos restantes 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 agregó en la sección anterior hasta el paso Modelo de tren para crear una arista que conecte los dos pasos.

1. En la barra lateral izquierda, selecciona Registrar modelo y arrástralo al lienzo.

2. En el lienzo, elige el paso Registrar el modelo que has añadido.

3. Para seleccionar un modelo para registrarlo, elija Añadir en Modelo (entrada).

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

5. Complete los campos restantes 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 del modelo de tren que agregó en la sección anterior hasta el paso Registrar el modelo para crear una arista que conecte los dos escalones.

1. En la barra lateral izquierda, selecciona Implementar el modelo (punto final) y arrástralo al lienzo.

2. En el lienzo, elige el paso de implementación del modelo (punto final) que agregaste.

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

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

5. Introduzca un nombre y una descripción para su punto final.

6. Haga clic y arrastre el cursor desde el paso Registrar el modelo que agregó en la sección anterior hasta el paso Implementar el modelo (punto final) para crear una arista que conecte los dos pasos.

7. Rellene los campos restantes 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 preprocesamiento se pasa al paso de procesamiento para ejecutarlo con los datos de entrada. A continuación, el paso de entrenamiento utiliza las funciones y etiquetas de entrenamiento preprocesadas para entrenar un modelo. El paso de evaluación utiliza 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. Esto es muy similar al run método de una instancia de procesador en SageMaker PythonSDK. 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. Un paso Properties como estos se puede utilizar en pasos posteriores y resolver sus valores de tiempo de ejecución en 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. Configure un estimador para el XGBoost algoritmo y el conjunto de datos de entrada. El tipo de instancia de entrenamiento se transfiere al estimador. Un guion de entrenamiento típico:

   • carga datos de los canales de entrada

   • configura el entrenamiento con hiperparámetros

   • entrena un modelo

   • guarda un modelo para model_dir poder alojarlo más tarde

   SageMaker carga el modelo en 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 una TrainingStep utilizando la instancia del estimador y las propiedades de. ProcessingStep Pase el canal S3Uri de "validation" salida "train" y 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 JSON archivo 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 una ProcessingStep con la instancia del procesador, los canales de entrada y salida y el evaluation.py script. Transfiere:

   • la S3ModelArtifacts propiedad desde el escalón step_train de formación

   • el S3Uri del canal de "test" salida de la etapa step_process de procesamiento

   Esto es muy similar al run método de una instancia de procesador en 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],
   )

Para definir una transformación CreateModelStep por lotes

1. Cree un SageMaker modelo. 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 SageMaker modelo.

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

3. Cree la suya CreateModelStep utilizando la instancia del SageMaker modelo CreateModelInput y la 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 número de instancias y el bucket de Amazon S3 de salida adecuadoURI. 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. Pipelines lo crea 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 correspondiente,. "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=[], 
   )

Para crear 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. (Opcional) Examine la definición de la JSON tubería para asegurarse de que está bien formada.

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