Definir um pipeline

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

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

3. Escolha Criar.

4. Escolha em branco.

5. Na barra lateral esquerda, escolha Processar dados e arraste-os para a tela.

6. Na tela, 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 Train model e arraste-o até a tela.

2. Na tela, escolha a etapa do modelo de trem 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 Amazon URI S3 no campo Localização (URIS3) 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 Dados do processo que você adicionou na seção anterior à etapa do modelo Train para criar uma borda conectando as duas etapas.

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

2. Na tela, 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 do modelo Train que você adicionou na seção anterior à etapa do modelo Registrar para criar uma borda conectando as duas etapas.

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

2. Na tela, escolha a etapa de implantação do 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 ponto final para criar um novo ponto final.

5. Insira um nome e uma descrição para seu 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. Esse script de pré-processamento é passado para a etapa de processamento para execução nos dados de entrada. A etapa de treinamento então usa os recursos e rótulos de treinamento pré-processados para treinar um modelo. A etapa de avaliação usa o modelo treinado e os recursos 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 SageMaker PythonSDK. 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. Etapas Properties como essas podem ser usadas em etapas subsequentes e resolvidas para seus valores de tempo de execução em tempo de execução.

   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 para model_dir que ele possa ser hospedado posteriormente

   SageMaker 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 do. ProcessingStep Passe o canal S3Uri do "train" e "validation" de saída para TrainingStep o. 

   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 JSON arquivo 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 evaluation.py script. Passe em:

   • a S3ModelArtifacts propriedade da etapa step_train de treinamento

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

   Isso é muito semelhante ao run método de uma instância de processador no 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 uma CreateModelStep para transformação em lote

1. Crie um SageMaker modelo. 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 SageMaker modelo.

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

3. Crie seu CreateModelStep usando a instância CreateModelInput e SageMaker modelo 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, a contagem de instâncias e o bucket de saída desejado do Amazon S3. URI 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 isso 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 JSON pipeline para garantir que ele esteja bem formado.

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