Tipos de etapas Propriedades da etapa Paralelismo de etapas Dependência de dados entre etapas Dependência personalizada entre etapas Use uma imagem personalizada em uma etapa

Etapas do pipeline de construção de modelos

SageMaker Os dutos são compostos por etapas. Essas etapas definem as ações que o pipeline executa e as relações entre as etapas usando propriedades.

Tópicos

Tipos de etapas
Propriedades da etapa
Paralelismo de etapas
Dependência de dados entre etapas
Dependência personalizada entre etapas
Use uma imagem personalizada em uma etapa

Tipos de etapas

O seguinte descreve os requisitos de cada tipo de etapa e fornece um exemplo de implantação da etapa. Essas não são implantações funcionais porque não fornecem os recursos e as entradas necessários. Para obter um tutorial que implementa essas etapas, consulte Crie e gerencie SageMaker pipelines.

nota

Você também pode criar uma etapa a partir do seu código de aprendizado de máquina local convertendo-a em uma etapa do SageMaker Pipelines com o @step decorador. Para ter mais informações, consulte decorador @step.

O Amazon SageMaker Model Building Pipelines oferece suporte aos seguintes tipos de etapas:

Processamento
Treinamento
Ajustar
AutoML
Model
CreateModel
RegisterModel
Transformação
Condição
Callback
Lambda
ClarifyCheck
QualityCheck
EMR
Notebook Job
Falha

decorador @step

Você pode criar uma etapa a partir do código de aprendizado de máquina local usando o @step decorador. Depois de testar seu código, você pode converter a função em uma etapa do SageMaker pipeline anotando-a com o @step decorador. SageMaker O Pipelines cria e executa um pipeline quando você passa a saída da função @step -decorada como uma etapa para o seu pipeline. Você também pode criar um pipeline DAG de várias etapas que inclua uma ou mais funções @step decoradas, bem como etapas tradicionais SageMaker do pipeline. Para obter mais detalhes sobre como criar uma etapa com o @step decorador, consulteCódigo L ift-and-shift Python com o decorador @step.

Processamento de etapa

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.

Uma etapa de processamento requer um processador, um script Python que defina o código de processamento, as saídas para processamento e os argumentos do trabalho. O exemplo a seguir mostra como criar uma definição de ProcessingStep.


from sagemaker.sklearn.processing import SKLearnProcessor

sklearn_processor = SKLearnProcessor(framework_version='1.0-1',
                                     role=<role>,
                                     instance_type='ml.m5.xlarge',
                                     instance_count=1)


from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

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")
]

step_process = ProcessingStep(
    name="AbaloneProcess",
    step_args = sklearn_processor.run(inputs=inputs, outputs=outputs,
        code="abalone/preprocessing.py")
)

Passe parâmetros de tempo de execução

O exemplo a seguir mostra como passar parâmetros de tempo de execução de um PySpark processador para umProcessingStep.


from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.spark.processing import PySparkProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

pipeline_session = PipelineSession()

pyspark_processor = PySparkProcessor(
    framework_version='2.4',
    role=<role>,
    instance_type='ml.m5.xlarge',
    instance_count=1,
    sagemaker_session=pipeline_session,
)

step_args = pyspark_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="preprocess.py",
    arguments=None,
)


step_process = ProcessingStep(
    name="AbaloneProcess",
    step_args=step_args,
)

Para obter mais informações sobre os requisitos das etapas de processamento, consulte sagemaker.workflow.steps. ProcessingStepdocumentação. Para obter um exemplo detalhado, consulte Definir uma etapa de processamento para engenharia de recursos no caderno de exemplo Orchestrate Jobs to Training and Evaluate Models with Amazon SageMaker Pipelines.

Etapa de treinamento

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 a Amazon SageMaker.

Uma etapa de treinamento requer um estimador, bem como entradas de dados de treinamento e validação. Os exemplos a seguir mostram como criar uma definição de TrainingStep. Para obter mais informações sobre os requisitos das etapas de treinamento, consulte sagemaker.workflow.steps. TrainingStepdocumentação.


from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TrainingStep

from sagemaker.xgboost.estimator import XGBoost

pipeline_session = PipelineSession()

xgb_estimator = XGBoost(..., sagemaker_session=pipeline_session)

step_args = xgb_estimator.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="TrainAbaloneModel",
    step_args=step_args,
)

Etapa de Ajuste

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 Execute o ajuste automático do modelo com SageMaker.

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

Uma etapa de ajuste requer uma HyperparameterTunere 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 do sagemaker.workflow.steps. TuningStepclasse para obter o artefato do modelo de uma das versões do modelo de melhor desempenho. Para um notebook que mostra como usar uma etapa de ajuste em um SageMaker pipeline, consulte sagemaker-pipelines-tuning-step.ipynb.

Importante

As etapas de ajuste foram introduzidas no Amazon SageMaker Python SDK v2.48.0 e no Amazon Studio Classic v3.8.0. SageMaker Você deve atualizar o Studio Classic antes de usar uma etapa de ajuste, caso contrário, o DAG do pipeline não será exibido. Para atualizar o Studio Classic, consulteDesligue e atualize o SageMaker Studio Classic.

Os exemplos a seguir mostram 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://my-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 o HyperParameterTuningJobObjetivo. 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 sagemaker.workflow.steps. TuningStepdocumentação.

Etapa do AutoML

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

nota

Atualmente, a etapa do AutoML oferece suporte somente ao modo de treinamento em conjunto.

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://my-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://my-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. Você pode obter o modelo com a melhor métrica objetiva do trabalho do AutoML usando o método get_best_auto_ml_model e uma role IAM para acessar os artefatos do modelo da seguinte forma.


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

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

Etapa do modelo

Use a ModelStep para criar ou registrar um SageMaker modelo. Para obter mais informações sobre ModelStep os requisitos, consulte o sagemaker.workflow.model_step. ModelStepdocumentação.

Criar um modelo

Você pode usar a ModelStep para criar um SageMaker modelo. A ModelStep exige artefatos do modelo e informações sobre o tipo de SageMaker instância que você precisa usar para criar o modelo. Para obter mais informações sobre SageMaker modelos, consulte Treinar um modelo com a 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 um no registro sagemaker.pipeline.PipelineModel de SageMaker modelos da Amazon. 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 Registrar e implantar modelos com o Registro do modelo.

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,
)

CreateModel Etapa

Importante

Recomendamos usar Etapa do modelo para criar modelos a partir da versão 2.90.0 do SDK do Python SageMaker . CreateModelStepcontinuará funcionando nas versões anteriores do SDK do SageMaker Python, mas não tem mais suporte ativo.

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

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


from sagemaker.workflow.steps import CreateModelStep

step_create_model = CreateModelStep(
    name="AbaloneCreateModel",
    model=best_model,
    inputs=inputs
)

RegisterModel Etapa

Importante

Recomendamos usar Etapa do modelo para registrar modelos a partir da versão 2.90.0 do SDK do Python SageMaker . RegisterModelcontinuará funcionando nas versões anteriores do SDK do SageMaker Python, mas não tem mais suporte ativo.

Você usa uma RegisterModel etapa para registrar um SageMaker.model.Model ou um sagemaker.pipeline. PipelineModelcom o registro de SageMaker modelos da Amazon. 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 Registrar e implantar modelos com o Registro do modelo. Para obter mais informações sobre os requisitos das RegisterModel etapas, consulte sagemaker.workflow.step_collections. RegisterModeldocumentação.

O exemplo a seguir mostra como criar uma etapa RegisterModel que registra um PipelineModel.


import time
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

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=sagemaker_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=sagemaker_session,
 role=role)

from sagemaker.workflow.step_collections import RegisterModel
from sagemaker import PipelineModel
pipeline_model = PipelineModel(models=[sklearn_model,xgboost_model],role=role,sagemaker_session=sagemaker_session)

step_register = RegisterModel(
 name="AbaloneRegisterModel",
 model=pipeline_model,
 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',
)

Se model não for fornecida, a etapa do modelo de registro requer um estimador, conforme mostrado no exemplo a seguir.


from sagemaker.workflow.step_collections import RegisterModel

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
)

Etapa de transformação

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 Executar 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 uma etapa Transform. Para obter mais informações sobre os requisitos das Transform etapas, consulte sagemaker.workflow.steps. TransformStepdocumentação.


from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.transformer import Transformer
from sagemaker.inputs import TransformInput
from sagemaker.workflow.steps import TransformStep

transformer = Transformer(..., sagemaker_session=PipelineSession())

step_transform = TransformStep(
    name="AbaloneTransform",
    step_args=transformer.transform(data="s3://my-bucket/my-data"),
)

Etapa de condição

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 requer uma lista de condições, uma lista de etapas a serem executadas se a condição avalia como true e uma lista de etapas a serem executadas se a condição avalia como false. O exemplo a seguir mostra como criar uma definição de ConditionStep.

Limitações

SageMaker Os pipelines não suportam o uso de etapas de condição aninhadas. Você não pode passar uma etapa de condição como entrada para outra etapa de condição.
Uma etapa de condição não pode usar etapas idênticas nas duas ramificações. Se você precisar da mesma funcionalidade de etapa em ambas as ramificações, duplique a etapa e dê a ela um nome diferente.


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
)

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

Para obter mais informações sobre ConditionStep requisitos, consulte sagemaker.workflow.condition_step. ConditionStepReferência da API. Para obter mais informações sobre as condições suportadas, consulte Amazon SageMaker Model Building Pipelines - Conditions na documentação do SageMaker Python SDK.

Etapa de retorno de chamada

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

SageMaker O Pipelines envia uma mensagem para uma fila do Amazon Simple Queue Service (Amazon SQS) especificada pelo cliente. A mensagem contém um token SageMaker gerado pelo Pipelines e uma lista de parâmetros de entrada fornecida pelo cliente. Depois de enviar a mensagem, a SageMaker Pipelines espera por uma resposta do cliente.
O cliente recupera a mensagem da fila do Amazon SQS e inicia seu processo personalizado.
Quando o processo termina, o cliente chama uma das seguintes APIs e envia o token gerado pelo Pipelines: SageMaker
- SendPipelineExecutionStepSucesso, junto com uma lista de parâmetros de saída
- SendPipelineExecutionStepFalha, junto com um motivo de falha
A chamada da API faz com que SageMaker os pipelines continuem o processo do pipeline ou falhem no processo.

Para obter mais informações sobre os requisitos das Callback etapas, consulte sagemaker.workflow.callback_step. CallbackStepdocumentação. Para obter uma solução completa, consulte Estender SageMaker pipelines para incluir etapas personalizadas usando etapas de retorno de chamada.

Importante

Callbacketapas foram introduzidas no Amazon SageMaker Python SDK v2.45.0 e no Amazon Studio Classic v3.6.2. SageMaker Você deve atualizar o Studio Classic antes de usar uma Callback etapa, caso contrário, o DAG do pipeline não será exibido. Para atualizar o Studio Classic, consulteDesligue e atualize o SageMaker Studio Classic.

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={...}
            )
'

nota

Os parâmetros de saída para CallbackStep não devem ser aninhados. Por exemplo, se você usar um dicionário aninhado como parâmetro de saída, o dicionário será tratado como uma única string (ex. {"output1": "{\"nested_output1\":\"my-output\"}"}). Se você fornecer um valor aninhado, ao tentar se referir a um determinado parâmetro de saída, será gerado um erro de cliente que não pode ser testado novamente.

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 Callback etapa em execução, o SageMaker Pipelines envia uma mensagem adicional do Amazon SQS para a fila SQS especificada. 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, seguida por uma chamada a SendPipelineExecutionStepSuccess ou SendPipelineExecutionStepFailure.

Somente quando o SageMaker Pipelines recebe uma dessas chamadas, isso interrompe o processo do pipeline.

Etapa do Lambda

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

Importante

As etapas do Lambda foram introduzidas no Amazon SageMaker Python SDK v2.51.0 e no Amazon Studio Classic v3.9.1. SageMaker Você deve atualizar o Studio Classic antes de usar uma etapa do Lambda ou o pipeline DAG não será exibido. Para atualizar o Studio Classic, consulteDesligue e atualize o SageMaker Studio Classic.

SageMaker fornece a classe SageMaker.lambda_helper.Lambda para criar, atualizar, invocar e excluir funções Lambda. Lambdatem 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
    ...
)

O sagemaker.workflow.lambda_step. LambdaStepclasse tem um lambda_func argumento do tipoLambda. 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:

zipped_code_dir: o caminho da função do Lambda compactada

s3_bucket: bucket do Amazon S3 onde zipped_code_dir deve ser carregado
script: o caminho do arquivo de script da função do Lambda

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.

nota

Os parâmetros de entrada e saída não devem ser aninhados. Por exemplo, se você usar um dicionário aninhado como parâmetro de saída, o dicionário será tratado como uma única string (ex. {"output1": "{\"nested_output1\":\"my-output\"}"}). Se você fornecer um valor aninhado e tentar referenciá-lo posteriormente, será gerado um erro de cliente que não pode ser repetido.

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á suporte para 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á suporte para 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 do Lambda ou até que o tempo limite seja atingido, o que ocorrer primeiro, e depois será 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.

ClarifyCheck Etapa

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 viés 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 deriva podem ser usadas pelo Amazon SageMaker Model Monitor para seus endpoints de modelo, para que você não precise 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 ClarifyCheck etapa aproveita o contêiner pré-construído Amazon SageMaker Clarify, que 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 ter mais informações, consulte Comece com um contêiner 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.

Verificação de viés de dados
Verificação de viés do modelo
Verificação da explicabilidade do modelo

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

DataBiasCheckConfig
ModelBiasCheckConfig
ModelExplainabilityCheckConfig

A ClarifyCheck etapa inicia uma tarefa de processamento que executa o contêiner pré-construído do SageMaker Clarify e requer configurações dedicadas para a verificação e a tarefa de processamento. ClarifyCheckConfige CheckJobConfig são funções auxiliares para essas configurações que estão alinhadas com a forma como a tarefa de processamento do SageMaker Clarify calcula para verificar o viés do modelo, o viés de dados ou a explicabilidade do modelo. Para ter mais informações, consulte Execute trabalhos de processamento do SageMaker Clarify para análise de viés 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:

skip_check: esse parâmetro indica se a verificação de oscilação em relação à linha de base anterior foi ignorada ou não. Se estiver definido como False, a linha de base anterior do tipo de verificação configurado deverá estar disponível.
register_new_baseline: esse parâmetro indica se uma linha de base recém-calculada pode ser acessada por meio da propriedade de etapa BaselineUsedForDriftCheckConstraints. Se estiver definido como False, a linha de base anterior do tipo de verificação configurado também deve estar disponível. Isso pode ser acessado através da propriedade BaselineUsedForDriftCheckConstraints.

Para ter mais informações, consulte Cálculo de linha de base, detecção de desvios, ciclo de vida e ClarifyCheck etapas QualityCheck no Amazon Model Building Pipelines SageMaker .

Trabalhar com linhas de base

Opcionalmente, você pode especificar o model_package_group_name para localizar a linha de base existente e a etapa ClarifyCheck extrai o pacote de modelo aprovado mais recente DriftCheckBaselines no grupo de pacotes de modelos. 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 ClarifyCheck etapas, consulte sagemaker.workflow.steps. ClarifyCheckEntre no Amazon SageMaker SageMaker SDK para Python. Para um notebook Amazon SageMaker Studio Classic que mostra como usar o ClarifyCheck step in SageMaker Pipelines, consulte sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

exemplo Crie uma etapa `ClarifyCheck` para verificação de viés de dados


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"
)

QualityCheck Etapa

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 QualityCheck etapa aproveita o contêiner pré-construído Amazon SageMaker Model Monitor, que 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 ter mais informações, consulte Contêiner pré-construí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.

Verificação de qualidade de dados
Verificação de qualidade do modelo

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

DataQualityCheckConfig
ModelQualityCheckConfig

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. As QualityCheckConfig e CheckJobConfig são funções auxiliares para essas configurações que 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 Crie uma linha de base de qualidade do 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:

skip_check: esse parâmetro indica se a verificação de oscilação em relação à linha de base anterior foi ignorada ou não. Se estiver definido como False, a linha de base anterior do tipo de verificação configurado deverá estar disponível.
register_new_baseline: esse parâmetro indica se uma linha de base recém-calculada pode ser acessada por meio das propriedades da etapa BaselineUsedForDriftCheckConstraints e BaselineUsedForDriftCheckStatistics. Se estiver definido como False, a linha de base anterior do tipo de verificação configurado também deve estar disponível. Eles podem ser acessados por meio das propriedades BaselineUsedForDriftCheckConstraints e BaselineUsedForDriftCheckStatistics.

Para ter mais informações, consulte Cálculo de linha de base, detecção de desvios, ciclo de vida e ClarifyCheck etapas QualityCheck no Amazon Model Building Pipelines SageMaker .

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, ou pode simplesmente especificar model_package_group_name e a etapa QualityCheck extrai DriftCheckBaselines no pacote de modelo aprovado mais recente no grupo de pacotes de modelos. Quando você especifica a model_package_group_name, a supplied_baseline_constraints e supplied_baseline_statistics, 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 que você está executando.

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

exemplo Crie uma etapa `QualityCheck` para verificação da qualidade dos dados


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"
)

Etapa do EMR

Você pode usar a etapa EMR do Amazon SageMaker Model Building Pipelines para processar etapas do Amazon EMR em um cluster do Amazon EMR em execução ou fazer com que o pipeline crie e gerencie um cluster do Amazon EMR para você. 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 a ser usado pelo cluster do Amazon EMR e quaisquer 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, ou a configuração do cluster se quiser que a etapa do EMR seja executada em um cluster que ele 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.

nota

As etapas do EMR exigem que a função passada para seu pipeline tenha permissões adicionais. Você deve anexar a política gerenciada da AWS : AmazonSageMakerPipelinesIntegrations à sua função de pipeline ou garantir que a função inclua as permissões dessa política.
A etapa do EMR não é compatível com o EMR sem servidor, nem com o Amazon EMR no EKS.
Se você processar uma etapa do EMR em um cluster em execução, só poderá usar um cluster que esteja em um dos seguintes estados: STARTING, BOOTSTRAPPING, RUNNING ou WAITING.
Se você processar etapas do EMR em um cluster em execução, poderá ter no máximo 256 etapas do EMR em um estado PENDING em um cluster do EMR. As etapas do EMR enviadas além desse limite resultam em falha na execução do pipeline. Você pode considerar usar Política de repetição para etapas do pipeline.
Você pode especificar o ID ou a configuração do cluster, mas não ambos.
A etapa do EMR depende da Amazon EventBridge para monitorar as mudanças na etapa do EMR ou no estado do cluster. Se você processa seu trabalho do Amazon EMR em um cluster em execução, a etapa do EMR usa a regra SageMakerPipelineExecutionEMRStepStatusUpdateRule para monitorar o estado da etapa do EMR. Se você processa seu trabalho em um cluster criado pela etapa do EMR para você, a etapa usa a regra SageMakerPipelineExecutionEMRClusterStatusRule para monitorar as alterações no estado do cluster. Se você encontrar alguma dessas EventBridge regras em sua AWS conta, não as exclua, caso contrário, sua etapa do EMR poderá não ser concluída.

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

Se você quiser 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 um exemplo de notebook que orienta você em um exemplo completo, consulte SageMaker Pipelines EMR Step With Running EMR Cluster.

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

Se você quiser iniciar um novo trabalho em um novo cluster EMRStep criado para você, forneça a configuração do cluster como um dicionário com a mesma estrutura de uma solicitação de RunJobfluxo. No entanto, não inclua os seguintes campos na configuração do cluster:

[Name]
[Steps]
[AutoTerminationPolicy]
[Instances][KeepJobFlowAliveWhenNoSteps]
[Instances][TerminationProtected]

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, que solicita 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ê em um exemplo completo, consulte SageMaker Pipelines EMR Step With Cluster Lifecycle Management.

Etapa de trabalho do Notebook

Use a NotebookJobStep para executar seu SageMaker Notebook Job de forma não interativa como uma etapa do pipeline. Para obter mais informações sobre trabalhos do SageMaker Notebook, consulteSageMaker Empregos em notebooks.

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

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


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 do NotebookJobStep pipeline é tratada como uma tarefa do SageMaker notebook, portanto, você pode acompanhar o status da execução no painel de tarefas do notebook Studio Classic UI se incluir tags específicas com o tags argumento. Para obter mais detalhes sobre as tags a serem incluídas, consulteVisualize suas tarefas de notebook no painel do Studio UI.

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

Etapa de falha

Você usa FailStep a para interromper a execução de um Amazon SageMaker Model Building Pipelines quando uma condição ou estado desejado não é alcançado e para marcar a execução desse pipeline como falhada. Isso FailStep também permite que você insira uma mensagem de erro personalizada, indicando a causa da falha na execução do pipeline.

nota

Quando uma FailStep e outras etapas do pipeline são executadas simultaneamente, o pipeline não termina até que todas as etapas simultâneas sejam concluídas.

Limitações de uso `FailStep`

Você não pode adicionar um FailStep à lista DependsOn de outras etapas. Para ter mais informações, consulte Dependência personalizada entre etapas.
Outras etapas não podem fazer referência ao FailStep. É sempre a última etapa na execução de um pipeline.
Você não pode repetir a execução de um pipeline que termine com a FailStep.

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

O exemplo de trecho de código a seguir usa um FailStep com um ErrorMessage configurado com parâmetros de pipeline e uma Join operação.


from sagemaker.workflow.fail_step import FailStep
from sagemaker.workflow.functions import Join
from sagemaker.workflow.parameters import ParameterInteger

mse_threshold_param = ParameterInteger(name="MseThreshold", default_value=5)
step_fail = FailStep(
    name="AbaloneMSEFail",
    error_message=Join(
        on=" ", values=["Execution failed due to MSE >", mse_threshold_param]
    ),
)

Propriedades da etapa

O properties atributo é usado para adicionar dependências de dados entre as etapas do pipeline. Essas dependências de dados são então usadas pelos SageMaker pipelines para construir o DAG a partir da definição do pipeline. Essas propriedades podem ser referenciadas como valores de espaço reservado e são resolvidas em tempo de execução.

O properties atributo de uma etapa do SageMaker Pipelines corresponde ao objeto retornado por uma Describe chamada para o tipo de SageMaker tarefa correspondente. Para cada tipo de trabalho, a chamada Describe retorna o seguinte objeto de resposta:

ProcessingStep— DescribeProcessingJob
TrainingStep— DescribeTrainingJob
TransformStep— DescribeTransformJob

Para verificar quais propriedades são referenciáveis para cada tipo de etapa durante a criação da dependência de dados, consulte Dependência de dados - Referência de propriedades no SDK do Amazon Python. SageMaker

Paralelismo de etapas

Quando uma etapa não depende de nenhuma outra etapa, ela é executada imediatamente após a execução do pipeline. No entanto, executar muitas etapas do pipeline em paralelo pode esgotar rapidamente os recursos disponíveis. Controle o número de etapas simultâneas para a execução de um pipeline com ParallelismConfiguration.

O exemplo a seguir é usado ParallelismConfiguration para definir o limite de etapas simultâneas como cinco.


pipeline.create(
    parallelism_config=ParallelismConfiguration(5),
)

Dependência de dados entre etapas

Você define a estrutura do seu DAG especificando as relações de dados entre as etapas. Para criar dependências de dados entre as etapas, transmita as propriedades de uma etapa como entrada para outra etapa no pipeline. A etapa que recebe a entrada não é iniciada até que a etapa que fornece a entrada termine de ser executada.

Uma dependência de dados usa JsonPath notação no formato a seguir. Esse formato percorre o arquivo de propriedades JSON, o que significa que você pode acrescentar <property> quantas instâncias forem necessárias para alcançar a propriedade aninhada desejada no arquivo. Para obter mais informações sobre JsonPath notação, consulte o JsonPath repositório.


<step_name>.properties.<property>.<property>

O exemplo a seguir mostra como especificar um bucket do Amazon S3 usando a propriedade ProcessingOutputConfig de uma etapa de processamento.


step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri

Para criar a dependência de dados, passe o bucket para uma etapa de treinamento da seguinte forma.


from sagemaker.workflow.pipeline_context import PipelineSession

sklearn_train = SKLearn(..., sagemaker_session=PipelineSession())

step_train = TrainingStep(
    name="CensusTrain",
    step_args=sklearn_train.fit(inputs=TrainingInput(
        s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
            "train_data"].S3Output.S3Uri
    ))
)

Dependência personalizada entre etapas

Quando você especifica uma dependência de dados, o SageMaker Pipelines fornece a conexão de dados entre as etapas. Como alternativa, uma etapa pode acessar os dados de uma etapa anterior sem usar diretamente os SageMaker Pipelines. Nesse caso, você pode criar uma dependência personalizada que diga aos SageMaker Pipelines que não iniciem uma etapa até que a execução de outra etapa seja concluída. Você cria uma dependência personalizada especificando o atributo de uma etapa DependsOn.

Como exemplo, o exemplo a seguir define uma etapa C que começa somente após a execução da etapa A e B da etapa.


{
  'Steps': [
    {'Name':'A', ...},
    {'Name':'B', ...},
    {'Name':'C', 'DependsOn': ['A', 'B']}
  ]
}

SageMaker Os pipelines lançam uma exceção de validação se a dependência criar uma dependência cíclica.

O exemplo a seguir cria uma etapa de treinamento que começa após a conclusão da execução de uma etapa de processamento.


processing_step = ProcessingStep(...)
training_step = TrainingStep(...)

training_step.add_depends_on([processing_step])

O exemplo a seguir cria uma etapa de treinamento que não começa até que duas etapas de processamento diferentes terminem de ser executadas.


processing_step_1 = ProcessingStep(...)
processing_step_2 = ProcessingStep(...)

training_step = TrainingStep(...)

training_step.add_depends_on([processing_step_1, processing_step_2])

O seguinte fornece uma forma alternativa de criar a dependência personalizada.


training_step.add_depends_on([processing_step_1])
training_step.add_depends_on([processing_step_2])

O exemplo a seguir cria uma etapa de treinamento que recebe informações de uma etapa de processamento e aguarda a conclusão da execução de uma etapa de processamento diferente.


processing_step_1 = ProcessingStep(...)
processing_step_2 = ProcessingStep(...)

training_step = TrainingStep(
    ...,
    inputs=TrainingInput(
        s3_data=processing_step_1.properties.ProcessingOutputConfig.Outputs[
            "train_data"
        ].S3Output.S3Uri
    )

training_step.add_depends_on([processing_step_2])

O exemplo a seguir mostra como recuperar uma lista de strings das dependências personalizadas de uma etapa.


custom_dependencies = training_step.depends_on

Use uma imagem personalizada em uma etapa

Você pode usar qualquer uma das imagens disponíveis do SageMaker Deep Learning Container ao criar uma etapa em seu pipeline.

Também é possível usar seu próprio contêiner com etapas do pipeline. Como você não pode criar uma imagem no Amazon SageMaker Studio Classic, você deve criar sua imagem usando outro método antes de usá-la com o Amazon SageMaker Model Building Pipelines.

Para usar seu próprio contêiner ao criar as etapas do seu pipeline, inclua o URI da imagem na definição do estimador. Para obter mais informações sobre como usar seu próprio contêiner com SageMaker, consulte Usando contêineres do Docker com SageMaker.

Atenção O Javascript está desativado ou não está disponível no seu navegador.

Para usar a documentação da AWS, o Javascript deve estar ativado. Consulte as páginas de Ajuda do navegador para obter instruções.

Convenções do documento

Parâmetros do pipeline

Código L ift-and-shift Python com o decorador @step

Etapas do pipeline de construção de modelos

Tópicos

Tipos de etapas

nota

decorador @step

Processamento de etapa

Etapa de treinamento

Etapa de Ajuste

Importante

Etapa do AutoML

nota

Etapa do modelo

Criar um modelo

Registrar um modelo

CreateModel Etapa

Importante

RegisterModel Etapa

Importante

Etapa de transformação

Etapa de condição

Etapa de retorno de chamada

Importante

nota

Etapa do Lambda

Importante

nota

ClarifyCheck Etapa

Configurando a etapa ClarifyCheck

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

Trabalhar com linhas de base

exemplo Crie uma etapa ClarifyCheck para verificação de viés de dados

QualityCheck Etapa

Configurando a etapa QualityCheck

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

Trabalhar com linhas de base

exemplo Crie uma etapa QualityCheck para verificação da qualidade dos dados

Etapa do EMR

nota

Etapa de trabalho do Notebook

Etapa de falha

nota

Limitações de uso FailStep

Propriedades da etapa

Paralelismo de etapas

Dependência de dados entre etapas

Dependência personalizada entre etapas

Use uma imagem personalizada em uma etapa

exemplo Crie uma etapa `ClarifyCheck` para verificação de viés de dados

exemplo Crie uma etapa `QualityCheck` para verificação da qualidade dos dados

Limitações de uso `FailStep`