Adicionar uma etapa

Na interface de usuário de arrastar e soltar do Pipelines, você pode usar uma etapa Executar código para executar seu próprio código como uma etapa do pipeline. Você pode fazer upload de uma função Python, script ou caderno para ser executado como parte do seu pipeline. Você deve usar essa etapa se quiser orquestrar um trabalho de ML personalizado que aproveite os atributos avançados do SageMaker ou outros serviços AWS.

A etapa Executar código carrega arquivos em seu bucket padrão do Amazon S3 para o Amazon SageMaker. Esse bucket pode não ter as permissões de compartilhamento de recursos entre origens (CORS) definidas. Para saber mais sobre como configurar as permissões CORS, consulte Requisito do CORS para dados de imagem de entrada.

A etapa Executar código usa um trabalho de treinamento do Amazon SageMaker para executar seu código. Certifique-se de que seu perfil do IAM tenha as permissões da API sagemaker:DescribeTrainingJob e sagemaker:CreateTrainingJob. Para saber mais sobre todas as permissões necessárias para o Amazon SageMaker e como configurá-las, consulte Permissões de API do Amazon SageMaker: referência a ações, permissões e recursos.

Para adicionar uma etapa de execução de código a um pipeline usando o Pipeline Designer, faça o seguinte:

Use uma etapa de processamento para criar um trabalho de processamento para processamento de dados. Para obter mais informações sobre trabalhos de processamento, consulte Processar dados e avaliar modelos.

Você usa uma etapa de treinamento para criar um trabalho de treinamento para treinar um modelo. Para obter mais informações sobre trabalhos de treinamento, consulte Treinar um modelo com o Amazon SageMaker.

Uma etapa de treinamento requer um estimador, bem como entradas de dados de treinamento e validação.

Você usa uma etapa de ajuste para criar um trabalho de ajuste de hiperparâmetros, também conhecido como otimização de hiperparâmetros (HPO). Um trabalho de ajuste de hiperparâmetros executa vários trabalhos de treinamento, cada um produzindo uma versão do modelo. Para obter mais informações sobre ajuste de hiperparâmetros, consulte Ajuste automático de modelos com o SageMaker.

O trabalho de ajuste está associado ao experimento do SageMaker para o pipeline, com os trabalhos de treinamento criados como testes. Para obter mais informações, consulte Integração de experimentos.

Uma etapa de ajuste requer um HyperParameterTuner e entradas de treinamento. Você pode retreinar trabalhos de ajuste anteriores especificando o parâmetro warm_start_config do HyperparameterTuner. Para obter mais informações sobre ajuste de hiperparâmetros e inicialização a quente, consulte Executar um trabalho de ajuste de hiperparâmetros de inicialização a quente.

Você usa o método get_top_model_s3_uri da classe SageMaker.workflow.steps.TuningStep para obter o artefato do modelo de uma das versões do modelo de melhor desempenho. Para um caderno que mostra como usar uma etapa de ajuste em um pipeline do SageMaker, consulte sagemaker-pipelines-tuning-step.ipynb.

O exemplo a seguir mostra como criar uma definição de TuningStep.

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep

tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession())
    
step_tuning = TuningStep(
    name = "HPTuning",
    step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://amzn-s3-demo-bucket/my-data"))
)

Obtenha a melhor versão do modelo

O exemplo a seguir mostra como obter a melhor versão do modelo do trabalho de ajuste usando o método get_top_model_s3_uri. No máximo, as 50 versões com melhor desempenho estão disponíveis, classificadas de acordo com HyperParameterTuningJobObjective. O top_k argumento é um índice das versões, onde top_k=0 está a versão com melhor desempenho e top_k=49 a versão com pior desempenho.

best_model = Model(
    image_uri=image_uri,
    model_data=step_tuning.get_top_model_s3_uri(
        top_k=0,
        s3_bucket=sagemaker_session.default_bucket()
    ),
    ...
)

Para obter mais informações sobre os requisitos das etapas de ajuste, consulte a documentação sagemaker.workflow.steps.TuningStep.

O ajuste fino treina um modelo de base pré-treinado do Amazon SageMaker JumpStart em um novo conjunto de dados. Esse processo, também conhecido como aprendizado por transferência, pode produzir modelos precisos com conjuntos de dados menores e menos tempo de treinamento. Ao ajustar um modelo, você pode usar o conjunto de dados padrão ou escolher seus próprios dados. Para saber mais sobre como ajustar um modelo de base do JumpStart, consulte Ajuste um modelo.

A etapa de ajuste fino usa um trabalho de treinamento do Amazon SageMaker para personalizar seu modelo. Certifique-se de que seu perfil do IAM tenha as permissões de API sagemaker:DescribeTrainingJob e sagemaker:CreateTrainingJob para executar o trabalho de ajuste fino em seu pipeline. Para saber mais sobre as permissões necessárias para o Amazon SageMaker e como configurá-las, consulte Permissões de API do Amazon SageMaker: referência a ações, permissões e recursos.

Para adicionar a etapa Ajustar modelo ao seu pipeline usando o editor de arrastar e soltar, siga estas etapas:

Use a API AutoML para criar um trabalho do AutoML para treinar automaticamente um modelo. Para obter mais informações sobre o AutoML, consulte Automatizar o desenvolvimento de modelos com o Amazon SageMaker Autopilot.

Os exemplos a seguir mostram como criar uma definição usando AutoMLStep.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.workflow.automl_step import AutoMLStep

pipeline_session = PipelineSession()

auto_ml = AutoML(...,
    role="<role>",
    target_attribute_name="my_target_attribute_name",
    mode="ENSEMBLING",
    sagemaker_session=pipeline_session) 

input_training = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-validation-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="validation",
)

step_args = auto_ml.fit(
    inputs=[input_training, input_validation]
)

step_automl = AutoMLStep(
    name="AutoMLStep",
    step_args=step_args,
)     

Obtenha a melhor versão do modelo

A etapa do AutoML treina automaticamente vários candidatos a modelos. Obtenha o modelo com a melhor métrica objetiva do trabalho do AutoML usando o método get_best_auto_ml_model da seguinte forma: Você também deve usar um role do IAM para acessar os artefatos do modelo.

best_model = step_automl.get_best_auto_ml_model(role=<role>)

Para obter mais informações, consulte a etapa do AutoML no SageMaker Python SDK.

Use um ModelStep para criar ou registrar um modelo do SageMaker. Para obter mais informações sobre requisitos do ModelStep, consulte a documentação do SageMaker.Workflow.model_step.ModelStep.

Criar um modelo

Você pode usar um ModelStep para criar um modelo do SageMaker. A ModelStep requer artefatos do modelo e informações sobre o tipo de instância do SageMaker que você precisa usar para criar o modelo. Para obter mais informações sobre modelos do SageMaker, consulte Treinar um modelo com o Amazon SageMaker.

O exemplo a seguir mostra como criar uma definição de ModelStep.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.model import Model
from sagemaker.workflow.model_step import ModelStep

step_train = TrainingStep(...)
model = Model(
    image_uri=pytorch_estimator.training_image_uri(),
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=PipelineSession(),
    role=role,
)

step_model_create = ModelStep(
   name="MyModelCreationStep",
   step_args=model.create(instance_type="ml.m5.xlarge"),
)

Registrar um modelo

Você pode usar ModelStep a para registrar um sagemaker.model.Model ou sagemaker.pipeline.PipelineModel com o Registro de Modelos do Amazon SageMaker. Um PipelineModel representa um pipeline de inferência, que é um modelo composto por uma sequência linear de contêineres que processam solicitações de inferência. Para obter mais informações sobre como registrar um modelo, consulte Implantação do registro do modelo com o Model Registry.

O exemplo a seguir mostra como criar um ModelStep que registra a PipelineModel.

import time

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

pipeline_session = PipelineSession()

code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix)

sklearn_model = SKLearnModel(
   model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri,
   entry_point='inference.py',
   source_dir='sklearn_source_dir/',
   code_location=code_location,
   framework_version='1.0-1',
   role=role,
   sagemaker_session=pipeline_session,
   py_version='py3'
)

xgboost_model = XGBoostModel(
   model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
   entry_point='inference.py',
   source_dir='xgboost_source_dir/',
   code_location=code_location,
   framework_version='0.90-2',
   py_version='py3',
   sagemaker_session=pipeline_session,
   role=role
)

from sagemaker.workflow.model_step import ModelStep
from sagemaker import PipelineModel

pipeline_model = PipelineModel(
   models=[sklearn_model, xgboost_model],
   role=role,sagemaker_session=pipeline_session,
)

register_model_step_args = pipeline_model.register(
    content_types=["application/json"],
   response_types=["application/json"],
   inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
   transform_instances=["ml.m5.xlarge"],
   model_package_group_name='sipgroup',
)

step_model_registration = ModelStep(
   name="AbaloneRegisterModel",
   step_args=register_model_step_args,
)

Você usa uma etapa de criar modelo para criar um modelo do SageMaker. Para mais informações sobre modelos do SageMaker, consulte Treinar um modelo com o Amazon SageMaker..

A etapa de criação do modelo requer artefatos do modelo e informações sobre o tipo de instância do SageMaker que você precisa usar para criar o modelo. O exemplo a seguir mostra como criar uma definição de etapa de criar modelo. Para obter mais informações sobre os requisitos da etapa de criar modelo, consulte a documentação sagemaker.workflow.steps.CreateModelStep.

A etapa de registro de modelo registra um modelo no registro de modelo do SageMaker.

No Pipeline Designer, use a etapa de implantar modelo (endpoint) para implantar o modelo em um endpoint. Você pode criar um novo endpoint ou usar um existente. A inferência em tempo real é ideal para workloads de inferência em que você tem requisitos em tempo real, interativos e de baixa latência. Você pode implantar seu modelo nos serviços de hospedagem do SageMaker e obter um endpoint em tempo real que pode ser usado para inferência. Esses endpoints são totalmente gerenciados e oferecem apoio ao ajuste de escala automático. Para saber mais sobre inferência em tempo real no SageMaker, consulte Inferência em tempo real.

Antes de adicionar uma etapa de implantar modelo ao pipeline, certifique-se de que seu perfil do IAM tenha as seguintes permissões:

Para saber mais sobre todas as permissões necessárias para o SageMaker e como configurá-las, consulte Permissões de API do Amazon SageMaker: referência a ações, permissões e recursos.

Para adicionar uma etapa de implantação de modelo ao seu Pipeline no editor de arrastar e soltar, conclua as seguintes etapas:

Você usa uma etapa de transformação para transformação em lote para executar inferência em um conjunto de dados inteiro. Para obter mais informações sobre a transformação em lotes, consulte Transformações em lotes com pipelines de inferência.

Uma etapa de transformação requer um transformador e os dados nos quais executar a transformação em lote. O exemplo a seguir mostra como criar uma definição de etapa de transformação. Para obter mais informações sobre os requisitos da etapa de transformação, consulte a documentação sagemaker.workflow.steps.TransformStep.

Você usa uma etapa de condição para avaliar a condição das propriedades da etapa para avaliar qual ação deve ser tomada em seguida no pipeline.

Uma etapa de condição exige:

Você usa uma etapa Callback para incorporar processos e serviços da AWS adicionais ao seu fluxo de trabalho que não são fornecidos diretamente pelo Amazon SageMaker Pipelines. Quando uma etapa Callback é executada, ocorre o seguinte procedimento:

Para obter mais informações sobre os requisitos da etapa Callback, consulte a documentação do SageMaker.workflow.callback_step.CallbackStep. Para obter uma solução completa, consulte Estender os pipelines do SageMaker para incluir etapas personalizadas usando etapas de retorno de chamada.

O exemplo a seguir demonstra uma implementação do procedimento anterior.

from sagemaker.workflow.callback_step import CallbackStep

step_callback = CallbackStep(
    name="MyCallbackStep",
    sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue",
    inputs={...},
    outputs=[...]
)

callback_handler_code = '
    import boto3
    import json

    def handler(event, context):
        sagemaker_client=boto3.client("sagemaker")

        for record in event["Records"]:
            payload=json.loads(record["body"])
            token=payload["token"]

            # Custom processing

            # Call SageMaker to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

Parando o comportamento

Um processo de pipeline não para durante a execução de uma etapa Callback.

Quando você chama StopPipelineExecution em um processo de pipeline com uma etapa Callback em execução, o Pipelines envia uma mensagem do Amazon SQS para a fila do SQS. O corpo da mensagem SQS contém um campo Status, que está definido como Stopping. O exemplo a seguir mostra um exemplo de corpo de mensagem SQS.

{
  "token": "26vcYbeWsZ",
  "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a",
  "arguments": {
    "number": 5,
    "stringArg": "some-arg",
    "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv"
  },
  "status": "Stopping"
}

Você deve adicionar lógica ao seu consumidor de mensagens do Amazon SQS para realizar qualquer ação necessária (por exemplo, limpeza de recursos) após o recebimento da mensagem. Em seguida, adicione uma chamada para SendPipelineExecutionStepSuccess ou SendPipelineExecutionStepFailure.

O Pipelines interrompe o processo de pipeline somente quando recebe uma dessas chamadas.

Você usa uma etapa do Lambda para executar uma função AWS Lambda. Você pode executar uma função do Lambda existente ou o SageMaker pode criar e executar uma nova função do Lambda. Para um caderno que mostra como usar uma etapa do Lambda em um pipeline do SageMaker, consulte sagemaker-pipelines-lambda-step.ipynb.

O SageMaker fornece a classe SageMaker:lambda_helper:Lambda para criar, atualizar, invocar e excluir funções do Lambda. Lambda tem a seguinte assinatura:

Lambda(
    function_arn,       # Only required argument to invoke an existing Lambda function

    # The following arguments are required to create a Lambda function:
    function_name,
    execution_role_arn,
    zipped_code_dir,    # Specify either zipped_code_dir and s3_bucket, OR script
    s3_bucket,          # S3 bucket where zipped_code_dir is uploaded
    script,             # Path of Lambda function script
    handler,            # Lambda handler specified as "lambda_script.lambda_handler"
    timeout,            # Maximum time the Lambda function can run before the lambda step fails
    ...
)

A classe SageMaker.Workflow.Lambda_step.LambdaStep tem um argumento lambda_func do tipo Lambda. Para invocar uma função do Lambda existente, o único requisito é fornecer o nome do recurso da Amazon (ARN) da função para function_arn. Se você não fornecer um valor para function_arn, deverá especificar handler uma das seguintes opções:

O exemplo a seguir mostra como criar uma definição de etapa Lambda que invoca uma função do Lambda existente.

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
        function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda"
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

O exemplo a seguir mostra como criar uma definição de etapa Lambda que cria e invoca uma função do Lambda usando um script de função do Lambda.

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
      function_name="split-dataset-lambda",
      execution_role_arn=execution_role_arn,
      script="lambda_script.py",
      handler="lambda_script.lambda_handler",
      ...
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

Entradas e saídas

Se sua função Lambda tiver entradas ou saídas, elas também devem ser definidas em sua etapa Lambda.

Ao definir a etapa Lambda, inputs deve ser um dicionário de pares de valores-chave. Cada valor do inputs dicionário deve ser de um tipo primitivo (string, número inteiro ou flutuante). Não há compatibilidade com objetos aninhados. Se não for definido, o inputs valor padrão será None.

O outputs valor deve ser uma lista de chaves. Essas chaves se referem a um dicionário definido na saída da Lambda função. Por exemplo inputs, essas chaves devem ser de tipos primitivos e não há compatibilidade com objetos aninhados.

Tempo limite e comportamento de parada

A Lambda classe tem um timeout argumento que especifica o tempo máximo que a função do Lambda pode ser executada. O valor padrão é de 120 segundos, com máximo de 10 minutos. Se a função do Lambda estiver em execução quando o tempo limite for atingido, a etapa Lambda falhará; no entanto, a função do Lambda continuará em execução.

Um processo de pipeline não pode ser interrompido enquanto uma etapa do Lambda está em execução porque a função do Lambda invocada pela etapa do Lambda não pode ser interrompida. Se você tentar interromper o processo enquanto a função do Lambda estiver em execução, o pipeline aguardará a conclusão da função ou até que o tempo limite seja atingido. Isso depende do que ocorrer primeiro. O processo, então, é interrompido. Se a função do Lambda terminar, o status do processo do pipeline será Stopped. Se o tempo limite for atingido, o status do processo do pipeline será Failed.

Você pode usar a etapa ClarifyCheck para realizar verificações de oscilação da linha de base em relação às linhas de base anteriores para análise de desvio e explicabilidade do modelo. Em seguida, você pode gerar e registrar suas linhas de base com o método model.register() e passar a saída desse método para Etapa do modelo usando step_args. Essas linhas de base para verificação de oscilação podem ser usadas pelo Amazon SageMaker Model Monitor para seus endpoints de modelo. Como resultado, você não precisa fazer uma sugestão de linha de base separadamente.

A etapa ClarifyCheck também pode extrair linhas de base para verificação de oscilação do registro do modelo. A etapa ClarifyCheck usa o contêiner predefinido do SageMaker Clarify. O contêiner fornece uma variedade de recursos de monitoramento de modelos, incluindo sugestão de restrições e validação de restrições em relação a uma determinada linha de base. Para obter mais informações, consulte Contêineres predefinidos do SageMaker Clarify.

Configurando a etapa ClarifyCheck

Você pode configurar a etapa ClarifyCheck para realizar somente um dos seguintes tipos de verificação sempre que ela for usada em um pipeline:

Para fazer isso, defina o parâmetro clarify_check_config com um dos seguintes valores de tipo de verificação:

A etapa ClarifyCheck inicia um trabalho de processamento que executa o contêiner predefinido do SageMaker Clarify e exige configurações dedicadas para o trabalho de verificação e processamento. ClarifyCheckConfig e CheckJobConfig são funções auxiliares para essas configurações. Essas funções auxiliares estão alinhadas com a forma como a tarefa de processamento do SageMaker Clarify calcula para verificar o desvio do modelo, o desvio de dados ou a explicabilidade do modelo. Para obter mais informações, consulte Execute trabalhos de processamento do SageMaker Clarify para análise de preconceito e explicabilidade.

Controlar os comportamentos das etapas para verificação de oscilação

A etapa ClarifyCheck requer os dois sinalizadores booleanos a seguir para controlar seu comportamento:

Para obter mais informações, consulte Cálculo da linha de base, detecção de oscilações e ciclo de vida com as etapas do ClarifyCheck e do QualityCheck no Amazon SageMaker Pipelines.

Trabalhar com linhas de base

Opcionalmente, você pode especificar o model_package_group_name para localizar a linha de base existente. Em seguida, a etapa ClarifyCheck extrai o DriftCheckBaselines do pacote de modelo aprovado mais recente no grupo de pacotes de modelo.

Ou você pode fornecer uma linha de base anterior por meio do parâmetro supplied_baseline_constraints. Se você especificar o model_package_group_name e o supplied_baseline_constraints, a etapa ClarifyCheck usará a linha de base especificada pelo parâmetro supplied_baseline_constraints.

Para obter mais informações sobre como usar os requisitos das etapas ClarifyCheck, consulte o SageMaker.workflow.steps.clarifyCheckStep no SDK do Amazon SageMaker para Python. Para um caderno do Amazon SageMaker Studio Classic que mostra como usar a etapa ClarifyCheck no Pipelines do SageMaker, consulte sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_bias_data_config = DataConfig(
    s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']),
    label=0,
    dataset_type="text/csv",
    s3_analysis_config_output_path=data_bias_analysis_cfg_output_path,
)

data_bias_config = BiasConfig(
    label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]]  
)

data_bias_check_config = DataBiasCheckConfig(
    data_config=data_bias_data_config,
    data_bias_config=data_bias_config,
)h

data_bias_check_step = ClarifyCheckStep(
    name="DataBiasCheckStep",
    clarify_check_config=data_bias_check_config,
    check_job_config=check_job_config,
    skip_check=False,
    register_new_baseline=False
   supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json",
    model_package_group_name="MyModelPackageGroup"
)

Você pode usar a etapa QualityCheck para realizar sugestões de linha de base e verificar oscilações em relação a uma linha de base anterior para verificar a qualidade dos dados ou a qualidade do modelo em um pipeline. Em seguida, você pode gerar e registrar suas linhas de base com o método model.register() e passar a saída desse método para Etapa do modelo usando step_args. ]

O Model Monitor pode usar essas linhas de base para verificar a oscilação dos endpoints do modelo, para que você não precise fazer uma sugestão de linha de base separadamente. A etapa QualityCheck também pode extrair linhas de base para verificação de oscilação do registro do modelo. A etapa QualityCheck aproveita o contêiner predefinido do Amazon SageMaker Model Monitor. O contêiner tem uma variedade de recursos de monitoramento de modelos, incluindo sugestão de restrições, geração de estatísticas e validação de restrições em relação a uma linha de base. Para obter mais informações, consulte Contêiner pré-criado do Amazon SageMaker Model Monitor.

Configurando a etapa QualityCheck

Você pode configurar a etapa QualityCheck para realizar somente um dos seguintes tipos de verificação sempre que ela for usada em um pipeline:

Para fazer isso, defina o parâmetro quality_check_config com um dos seguintes valores de tipo de verificação:

A etapa QualityCheck inicia um trabalho de processamento que executa o contêiner pré-construído do Model Monitor e requer configurações dedicadas para a verificação e o trabalho de processamento. QualityCheckConfig e CheckJobConfig são funções auxiliares para essas configurações. Essas funções auxiliares estão alinhadas com a forma como o Model Monitor cria uma linha de base para a qualidade do modelo ou o monitoramento da qualidade dos dados. Para obter mais informações sobre as sugestões de linha de base do Model Monitor, consulte Criar uma linha de base e Criar uma linha de base da qualidade de modelo.

Controlar os comportamentos das etapas para verificação de oscilação

A etapa QualityCheck requer os dois sinalizadores booleanos a seguir para controlar seu comportamento:

Para obter mais informações, consulte Cálculo da linha de base, detecção de oscilações e ciclo de vida com as etapas do ClarifyCheck e do QualityCheck no Amazon SageMaker Pipelines.

Trabalhar com linhas de base

Você pode especificar uma linha de base anterior diretamente por meio dos parâmetros supplied_baseline_statistics e supplied_baseline_constraints. Você também pode especificar o model_package_group_name, e a etapa QualityCheck extrai DriftCheckBaselines do pacote de modelo aprovado mais recente no grupo de pacotes de modelo.

Quando você especifica o seguinte, a etapa QualityCheck usa a linha de base especificada por supplied_baseline_constraints e supplied_baseline_statistics no tipo de verificação da etapa QualityCheck:

Para obter mais informações sobre como usar os requisitos das etapas QualityCheck, consulte o SageMaker.workflow.steps.qualityCheckStep no SDK do Amazon SageMaker para Python. Para um caderno do Amazon SageMaker Studio Classic que mostra como usar a etapa QualityCheck no Pipelines do SageMaker, consulte sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_quality_check_config = DataQualityCheckConfig(
    baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"),
    output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep'])
)

data_quality_check_step = QualityCheckStep(
    name="DataQualityCheckStep",
    skip_check=False,
    register_new_baseline=False,
    quality_check_config=data_quality_check_config,
    check_job_config=check_job_config,
    supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json",
    supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json",
    model_package_group_name="MyModelPackageGroup"
)

Use a etapa do EMR do Amazon SageMaker Pipelines para:

Para obter mais informações sobre o Amazon EMR, consulte o Conceitos básicos do Amazon EMR.

A etapa do EMR exige que EMRStepConfig inclua a localização do arquivo JAR usado pelo cluster do Amazon EMR e os argumentos a serem passados. Você também fornece o ID do cluster do Amazon EMR se quiser executar a etapa em um cluster do EMR em execução. Você também pode passar a configuração do cluster para executar a etapa do EMR em um cluster que ela cria, gerencia e encerra para você. As seções a seguir incluem exemplos e links para exemplos de cadernos que demonstram os dois métodos.

Inicie um novo trabalho em um cluster do Amazon EMR em execução

Par iniciar um novo trabalho em um cluster do Amazon EMR em execução, transmita o ID do cluster como uma string para o argumento cluster_id de EMRStep. O exemplo a seguir demonstra esse procedimento.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

step_emr = EMRStep (
    name="EMRSampleStep", # required
    cluster_id="j-1ABCDEFG2HIJK", # include cluster_id to use a running cluster
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

Para ver uma amostra de caderno que orienta você com um exemplo completo, consulte Etapa de EMR do Pipelines com cluster do EMR em execução.

Inicie um novo trabalho em um cluster do Amazon EMR em execução

Para iniciar um novo trabalho em um novo cluster que o EMRStep criou para você, forneça a configuração de cluster como um dicionário. O dicionário deve ter a mesma estrutura de uma solicitação de RunJobFlow. No entanto, não inclua os seguintes campos na configuração do cluster:

Todos os outros argumentos RunJobFlow estão disponíveis para uso na configuração do cluster. Para obter detalhes sobre a sintaxe da solicitação, consulte RunJobFlow.

O exemplo a seguir passa uma configuração de cluster para uma definição de etapa do EMR. Isso faz com que a etapa inicie um novo trabalho em um novo cluster do EMR. A configuração do cluster EMR neste exemplo inclui especificações para os nós primários e principais do cluster EMR. Para obter mais informações sobre os tipos de nós do Amazon EMR, consulte Compreender os tipos de nós: nós primários, principais e de tarefas.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_step_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

# include your cluster configuration as a dictionary
emr_cluster_config = {
    "Applications": [
        {
            "Name": "Spark", 
        }
    ],
    "Instances":{
        "InstanceGroups":[
            {
                "InstanceRole": "MASTER",
                "InstanceCount": 1,
                "InstanceType": "m5.2xlarge"
            },
            {
                "InstanceRole": "CORE",
                "InstanceCount": 2,
                "InstanceType": "m5.2xlarge"
            }
        ]
    },
    "BootstrapActions":[],
    "ReleaseLabel": "emr-6.6.0",
    "JobFlowRole": "job-flow-role",
    "ServiceRole": "service-role"
}

emr_step = EMRStep(
    name="emr-step",
    cluster_id=None,
    display_name="emr_step",
    description="MyEMRStepDescription",
    step_config=emr_step_config,
    cluster_config=emr_cluster_config
)

Para ver uma amostra de caderno que orienta você com um exemplo completo, consulte Etapa de EMR do Pipelines com gerenciamento de ciclo de vida do cluster.

Use NotebookJobStep para executar seu trabalho de caderno do SageMaker de forma não interativa como uma etapa de pipeline. Se você criar seu pipeline na interface de arrastar e soltar do Pipelines, use Executar a etapa de código para executar seu caderno. Para obter mais informações sobre SageMaker Notebook Jobs, consulte Cadernos de trabalhos do SageMaker.

O NotebookJobStep requer no mínimo um caderno de entrada, URI de imagem e nome do kernel. Para obter mais informações sobre os requisitos da etapa do trabalho de caderno e outros parâmetros que você pode definir para personalizar sua etapa, consulte sagemaker.workflow.steps.NotebookJobStep.

O exemplo a seguir usa argumentos mínimos para definir um NotebookJobStep.

from sagemaker.workflow.notebook_job_step import NotebookJobStep


notebook_job_step = NotebookJobStep(
    input_notebook=input_notebook,
    image_uri=image_uri,
    kernel_name=kernel_name
)

Sua etapa de pipeline NotebookJobStep é tratada como uma tarefa do caderno do SageMaker. Como resultado, acompanhe o status de execução no painel de trabalhos do caderno na interface do Studio Classic incluindo tags específicas com o argumento tags. Para mais detalhes sobre tags a serem incluídas, consulte Visualizar seus trabalhos de caderno no painel da interface do Studio.

Além disso, se você agendar seu trabalho do caderno usando o SageMaker Python SDK, só poderá especificar determinadas imagens para executar o trabalho do caderno. Para obter mais informações, consulte Restrições de imagem para trabalhos do caderno do SageMaker Python SDK.

Use uma etapa de falha para interromper uma execução do Amazon SageMaker Pipelines quando uma condição ou estado desejado não for atingido. A etapa de falha também permite que você insira uma mensagem de erro personalizada, indicando a causa da falha na execução do pipeline.

Limitações do uso da etapa de falha

Você pode criar a mensagem de erro da etapa de falha na forma de uma sequência de texto estática. Como alternativa, você também pode usar os parâmetros de pipeline, uma operação de junção ou outras propriedades de etapa para criar uma mensagem de erro mais informativa, se usar o SDK.