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 uma instância de um processador para passar para a etapa de processamento.

   from sagemaker.core.processing import Processor
   from sagemaker.core import image_uris
   
   framework_version = "0.23-1"
   
   sklearn_image_uri = image_uris.retrieve(
       framework="sklearn",
       region=region,
       version=framework_version,
       instance_type="ml.m5.xlarge"
   )
   
   sklearn_processor = Processor(
       image_uri=sklearn_image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Crie uma etapa de processamento.

   from sagemaker.core.processing import ProcessingInput, ProcessingOutput
   from sagemaker.mlops.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 de treinamento para o algoritmo XGBoost e o conjunto de dados de entrada.

   from sagemaker.train import ModelTrainer
   from sagemaker.train.configs import Compute, OutputDataConfig
   from sagemaker.core import image_uris
   
   image_uri = image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = ModelTrainer(
       training_image=image_uri,
       compute=Compute(instance_type="ml.m5.xlarge", instance_count=1),
       output_data_config=OutputDataConfig(s3_output_path=model_path),
       hyperparameters={
           "objective": "reg:linear",
           "num_round": "50",
           "max_depth": "5",
           "eta": "0.2",
           "gamma": "4",
           "min_child_weight": "6",
           "subsample": "0.7",
           "silent": "0"
       },
       sagemaker_session=pipeline_session,
       role=role,
   )

3. Crie um TrainingStep usando a instância de treinamento e as propriedades doProcessingStep.

   from sagemaker.train.configs import InputData
   from sagemaker.mlops.workflow.steps import TrainingStep
   
   train_args = xgb_train.train(
       input_data_config=[
           InputData(
               channel_name="train",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "train"
               ].S3Output.S3Uri,
               content_type="text/csv"
           ),
           InputData(
               channel_name="validation",
               data_source=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 uma instância de a Processor para avaliação.

   from sagemaker.core.processing import Processor
   
   script_eval = Processor(
       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,
   )

2. Crie um ProcessingStep usando a instância do processador, os canais de entrada e saída e o evaluation.py script.

   from sagemaker.core.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 transformação em lote

• Crie um modelo de SageMaker IA e uma etapa do modelo.

  from sagemaker.serve import ModelBuilder
  from sagemaker.mlops.workflow.model_step import ModelStep
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  step_create_model = ModelStep(
      name="AbaloneCreateModel",
      step_args=model_builder.build(),
  )

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

• Crie uma instância de transformador e umaTransformStep.

  from sagemaker.core.transformer import Transformer
  from sagemaker.mlops.workflow.steps import TransformStep
  
  transformer = Transformer(
      model_name=step_create_model.properties.ModelName,
      instance_type="ml.m5.xlarge",
      instance_count=1,
      output_path=f"s3://{default_bucket}/AbaloneTransform",
      sagemaker_session=pipeline_session,
  )
  
  transform_args = transformer.transform(data=batch_data)
  
  step_transform = TransformStep(
      name="AbaloneTransform",
      step_args=transform_args,
  )

Para definir uma etapa de registro de modelo para criar um pacote de modelo

• Construa uma etapa de registro do modelo.

  from sagemaker.core.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.mlops.workflow.model_step import ModelStep
  from sagemaker.serve import ModelBuilder
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  register_args = model_builder.register(
      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
  )
  
  step_register = ModelStep(
      name="AbaloneRegisterModel",
      step_args=register_args,
  )

Para definir uma etapa de condição para verificar a precisão do modelo

• Defina uma condição e construa umaConditionStep.

  from sagemaker.core.workflow.conditions import ConditionLessThanOrEqualTo
  from sagemaker.mlops.workflow.condition_step import ConditionStep
  from sagemaker.core.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
  )
  
  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.

   from sagemaker.mlops.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())