Definir um pipeline

1. Abra o console do Studio seguindo as instruções em Inicie o Amazon SageMaker Studio.

2. No painel de navegação à esquerda, selecione Pipelines.

3. Escolha Criar.

4. Escolha Em branco.

5. Na barra lateral esquerda, escolha Processar dados e arraste-o para o canvas.

6. No canvas, escolha a etapa Processar dados que você adicionou.

7. Para adicionar um conjunto de dados de entrada, escolha Adicionar em Dados (entrada) na barra lateral direita e selecione um conjunto de dados.

8. Para adicionar um local para salvar conjuntos de dados de saída, escolha Adicionar em Dados (saída) na barra lateral direita e navegue até o destino.

9. Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte sagemaker.workflow.steps. ProcessingStep.

1. Na barra lateral esquerda, escolha Treinar modelo e arraste-o até o canvas.

2. No canvas, escolha a etapa Treinar modelo que você adicionou.

3. Para adicionar um conjunto de dados de entrada, escolha Adicionar em Dados (entrada) na barra lateral direita e selecione um conjunto de dados.

4. Para escolher um local para salvar seus artefatos de modelo, insira um URI do Amazon S3 no campo Localização (URI do S3) ou escolha Procurar no S3 para navegar até o local de destino.

5. Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte sagemaker.workflow.steps. TrainingStep.

6. Clique e arraste o cursor da etapa Processar dados que você adicionou na seção anterior à etapa Treinar modelo para criar uma borda conectando as duas etapas.

1. Na barra lateral esquerda, escolha Registrar modelo e arraste-o para o canvas.

2. No canvas, escolha a etapa Registrar modelo que você adicionou.

3. Para selecionar um modelo para registrar, escolha Adicionar em Modelo (entrada).

4. Escolha Criar um grupo de modelos para adicionar seu modelo a um novo grupo de modelos.

5. Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte sagemaker.workflow.step_collections. RegisterModel.

6. Clique e arraste o cursor da etapa Treinar modelo que você adicionou na seção anterior à etapa Registrar modelo para criar uma borda conectando as duas etapas.

1. Na barra lateral esquerda, escolha Implantar modelo (endpoint) e arraste-o até o canvas.

2. No canvas, escolha a etapa Implantar modelo (endpoint) que você adicionou.

3. Para escolher um modelo para implantar, escolha Adicionar em Modelo (entrada).

4. Escolha o botão de opção Criar endpoint para criar um novo endpoint.

5. Insira um Nome e uma Descrição para o endpoint.

6. Clique e arraste o cursor da etapa Registrar modelo que você adicionou na seção anterior à etapa Implantar modelo (endpoint) para criar uma borda conectando as duas etapas.

7. Preencha os campos restantes na barra lateral direita.

Para fazer download do conjunto de dados

1. Faça o download do conjunto de dados no bucket do Amazon S3 padrão da sua conta.

   !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. Faça o download de um segundo conjunto de dados para transformação em lote após a criação do 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 criar uma etapa de processamento

1. Crie um diretório para o script de processamento.

   !mkdir -p abalone

2. No diretório /abalone, crie um arquivo denominado preprocessing.py com o conteúdo a seguir. O script de pré-processamento é transmitido para a etapa de processamento para execução nos dados de entrada. Depois, a etapa de treinamento usa os rótulos e atributos de treinamento pré-processados para treinar um modelo. A etapa de avaliação usa o modelo treinado e os atributos e rótulos de teste pré-processados para avaliar o modelo. O script usa scikit-learn para fazer o seguinte:

   • Preencha os dados categóricos sex ausentes e codifique-os para que sejam adequados para treinamento.

   • Dimensione e normalize todos os campos numéricos, exceto rings e sex.

   • Divida os dados em conjuntos de dados de treinamento, teste e validação.

   %%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. Crie uma instância de um SKLearnProcessor para transmitir para a etapa de processamento.

   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. Crie uma etapa de processamento. Essa etapa inclui o SKLearnProcessor, os canais de entrada e saída e o script preprocessing.py que você criou. Isso é muito semelhante ao run método de uma instância de processador no SDK SageMaker AI Python. O parâmetro input_data transmitido para ProcessingStep são os dados de entrada da própria etapa. Esses dados de entrada são usados pela instância do processador quando ela é executada.

   Observe os canais denominados "train, "validation e "test" especificados na configuração de saída do trabalho de processamento. A etapa Properties, como essas, pode ser usada em etapas subsequentes e resolução para seus valores de runtime no 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
   )

Para definir uma etapa de treinamento

1. Especifique o caminho do modelo em que você deseja salvar os modelos do treinamento.

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

2. Configure um estimador para o XGBoost algoritmo e o conjunto de dados de entrada. O tipo de instância de treinamento é transmitido para o estimador. Um roteiro de treinamento típico:

   • carrega dados dos canais de entrada

   • configura o treinamento com hiperparâmetros

   • treina um modelo

   • salva um modelo em model_dir para que ele possa ser hospedado posteriormente

   SageMaker A IA carrega o modelo para o Amazon S3 na forma de model.tar.gz um no final do trabalho de treinamento.

   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. Crie um TrainingStep usando a instância do estimador e as propriedades de ProcessingStep. Passe S3Uri do "train" e o canal de saída "validation" para 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 uma etapa de processamento para avaliação do modelo

1. Crie um arquivo denominado evaluation.py no diretório /abalone. Esse script é usado em uma etapa de processamento para realizar a avaliação do modelo. Ele usa um modelo treinado e o conjunto de dados de teste como entrada e, em seguida, produz um arquivo JSON contendo métricas de avaliação de classificação.

   %%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. Crie uma instância de um ScriptProcessor que seja usada para criar uma 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. Crie um ProcessingStep usando a instância do processador, os canais de entrada e saída e o script evaluation.py. Transmita:

   • a propriedade S3ModelArtifacts da etapa de treinamento step_train

   • o S3Uri do canal de saída "test" da etapa de processamento step_process

   Isso é muito semelhante ao run método de uma instância de processador no SDK SageMaker AI Python. 

   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 uma CreateModelStep para transformação em lote

1. Crie um modelo de SageMaker IA. Transmita a propriedade S3ModelArtifacts a partir da etapa de treinamento 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 a entrada do modelo para seu modelo de SageMaker IA.

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

3. Crie sua CreateModelStep usando a CreateModelInput instância do modelo de SageMaker IA que você definiu.

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

Para definir uma TransformStep para realizar a transformação em lote

1. Crie uma instância transformadora com o tipo de instância de computação apropriado, a contagem de instâncias e o URI de saída desejado do bucket Amazon S3. Transmita a propriedade ModelName a partir das etapas step_create_model e 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. Crie uma TransformStep usando a instância do transformador que você definiu e o parâmetro do 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)
   )

Para definir uma RegisterModel etapa para criar um pacote de modelo

• Construa uma etapa RegisterModel usando a instância do estimador que você usou para a etapa de treinamento. Transmita a propriedade S3ModelArtifacts a partir da etapa de treinamento step_train e especifique um ModelPackageGroup. O Pipelines cria o ModelPackageGroup para você.

  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 uma etapa de condição para verificar a precisão do modelo

1. Defina uma condição ConditionLessThanOrEqualTo usando o valor de precisão encontrado na saída da etapa de processamento da avaliação do modelo, step_eval. Obtenha essa saída usando o arquivo de propriedades que você indexou na etapa de processamento e o respectivo valor médio JSONPath do erro quadrático,. "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. Construa uma ConditionStep. Transmita a condição ConditionEquals e, em seguida, defina as etapas de registro do pacote modelo e de transformação em lote como as próximas etapas, caso a condição seja aprovada.

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

Para criar um pipeline

1. Defina o seguinte para seu pipeline: name, parameters e steps. Os nomes devem ser exclusivos dentro de um par (account, region).

   nota

   Uma etapa só pode aparecer uma vez na lista de etapas do pipeline ou nas listas de etapas hipotéticas da etapa de condição. Ela não pode aparecer em 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 a definição do pipeline JSON para garantir que ela esteja bem formada.

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