As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
Definir um pipeline
Para orquestrar seus fluxos de trabalho com o Amazon SageMaker Pipelines, você deve gerar um gráfico acíclico direcionado (DAG) na forma de uma definição de pipeline JSON. O DAG especifica as diferentes etapas envolvidas em seu processo de ML, como pré-processamento de dados, treinamento de modelos, avaliação de modelos e implantação de modelos, bem como as dependências e o fluxo de dados entre essas etapas. O tópico a seguir mostra como gerar uma definição de pipeline.
Você pode gerar sua definição de pipeline JSON usando o SageMaker Python SDK ou o recurso drag-and-drop visual Pipeline Designer no Amazon Studio. SageMaker A imagem a seguir é uma representação do DAG de pipeline que você cria neste tutorial:

O pipeline que você define nas seções a seguir resolve um problema de regressão para determinar a idade de um abalone com base em suas medidas físicas. Para um notebook Jupyter executável que inclui o conteúdo deste tutorial, consulte Orquestração de trabalhos com
nota
Você pode referenciar a localização do modelo como uma propriedade da etapa de treinamento, conforme mostrado no end-to-end exemplo de CustomerChurn pipeline
Tópicos
O passo a passo a seguir orienta você pelas etapas para criar um pipeline básico usando o Pipeline Designer. drag-and-drop Se você precisar pausar ou encerrar sua sessão de edição do Pipeline no designer visual a qualquer momento, clique na opção Exportar. Isso permite que você baixe a definição atual do seu Pipeline para o seu ambiente local. Posteriormente, quando quiser retomar o processo de edição do Pipeline, você pode importar o mesmo arquivo de definição JSON para o designer visual.
Criar uma etapa de processamento
Para criar uma etapa do trabalho de processamento de dados, faça o seguinte:
-
Abra o console do Studio seguindo as instruções em Inicie o Amazon SageMaker Studio.
-
No painel de navegação à esquerda, selecione Pipelines.
-
Escolha Criar.
-
Escolha Em branco.
-
Na barra lateral esquerda, escolha Processar dados e arraste-o para o canvas.
-
No canvas, escolha a etapa Processar dados que você adicionou.
-
Para adicionar um conjunto de dados de entrada, escolha Adicionar em Dados (entrada) na barra lateral direita e selecione um conjunto de dados.
-
Para adicionar um local para salvar conjuntos de dados de saída, escolha Adicionar em Dados (saída) na barra lateral direita e navegue até o destino.
-
Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte sagemaker.workflow.steps. ProcessingStep
.
Criar uma etapa de treinamento
Para configurar uma etapa de treinamento de modelo, faça o seguinte:
-
Na barra lateral esquerda, escolha Treinar modelo e arraste-o até o canvas.
-
No canvas, escolha a etapa Treinar modelo que você adicionou.
-
Para adicionar um conjunto de dados de entrada, escolha Adicionar em Dados (entrada) na barra lateral direita e selecione um conjunto de dados.
-
Para escolher um local para salvar seus artefatos de modelo, insira um URI do Amazon S3 no campo Localização (URI do S3) ou escolha Procurar no S3 para navegar até o local de destino.
-
Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte sagemaker.workflow.steps. TrainingStep
. -
Clique e arraste o cursor da etapa Processar dados que você adicionou na seção anterior à etapa Treinar modelo para criar uma borda conectando as duas etapas.
Criar um pacote de modelo com a etapa de registro de modelo
Para criar um pacote de modelo com uma etapa de registro de modelo, faça o seguinte:
-
Na barra lateral esquerda, escolha Registrar modelo e arraste-o para o canvas.
-
No canvas, escolha a etapa Registrar modelo que você adicionou.
-
Para selecionar um modelo para registrar, escolha Adicionar em Modelo (entrada).
-
Escolha Criar um grupo de modelos para adicionar seu modelo a um novo grupo de modelos.
-
Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte sagemaker.workflow.step_collections. RegisterModel
. -
Clique e arraste o cursor da etapa Treinar modelo que você adicionou na seção anterior à etapa Registrar modelo para criar uma borda conectando as duas etapas.
Implante o modelo em um endpoint com uma etapa de implantação de modelo (endpoint)
Para implantar seu modelo usando uma etapa de implantação de modelo, faça o seguinte:
-
Na barra lateral esquerda, escolha Implantar modelo (endpoint) e arraste-o até o canvas.
-
No canvas, escolha a etapa Implantar modelo (endpoint) que você adicionou.
-
Para escolher um modelo para implantar, escolha Adicionar em Modelo (entrada).
-
Escolha o botão de opção Criar endpoint para criar um novo endpoint.
-
Insira um Nome e uma Descrição para o endpoint.
-
Clique e arraste o cursor da etapa Registrar modelo que você adicionou na seção anterior à etapa Implantar modelo (endpoint) para criar uma borda conectando as duas etapas.
-
Preencha os campos restantes na barra lateral direita.
Definir os parâmetros do Pipeline
Você pode configurar um conjunto de parâmetros do Pipeline cujos valores podem ser atualizados para cada execução. Para definir os parâmetros de pipeline e configurar os valores padrão, clique no ícone de engrenagem na parte inferior do designer visual.
Salvar pipeline
Depois de inserir todas as informações necessárias para criar seu pipeline, clique em Salvar na parte inferior do designer visual. Isso valida seu pipeline quanto a possíveis erros em runtime e notifica você. A operação Salvar não será bem-sucedida até que você resolva todos os erros sinalizados pelas verificações de validação automatizadas. Se quiser continuar a edição posteriormente, você pode salvar seu pipeline em andamento como uma definição JSON em seu ambiente local. Você pode exportar seu Pipeline como um arquivo de definição JSON clicando no botão Exportar na parte inferior do designer visual. Posteriormente, para continuar a atualização do Pipeline, faça o upload desse arquivo de definição JSON clicando no botão Importar.
Pré-requisitos
Para executar o tutorial a seguir, faça o seguinte:
-
Configure sua instância de caderno conforme descrito em Criar uma instância de caderno. Isso dá à sua função permissões para ler e gravar no Amazon S3 e criar trabalhos de treinamento, transformação em lote e processamento em SageMaker IA.
-
Conceda ao seu caderno permissões para obter e transmitir seu próprio perfil, conforme mostrado em Modificar uma política de permissões de perfil. Adicione o seguinte trecho de código do JSON para anexar essa política ao seu perfil: Substitua
<your-role-arn>
pelo ARN usado para criar sua instância de caderno.{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "iam:GetRole", "iam:PassRole" ], "Resource": "
<your-role-arn>
" } ] } -
Confie no diretor do serviço de SageMaker IA seguindo as etapas em Modificar a política de confiança de uma função. Adicione o seguinte fragmento de declaração à relação de confiança do seu perfil:
{ "Sid": "", "Effect": "Allow", "Principal": { "Service": "sagemaker.amazonaws.com" }, "Action": "sts:AssumeRole" }
Configurar o ambiente
Crie uma nova sessão de SageMaker IA usando o bloco de código a seguir. Isso retorna o ARN do perfil para a sessão. Esse ARN do perfil deve ser o ARN do perfil de execução que você configura como pré-requisito.
import boto3 import sagemaker import sagemaker.session from sagemaker.workflow.pipeline_context import PipelineSession region = boto3.Session().region_name sagemaker_session = sagemaker.session.Session() role = sagemaker.get_execution_role() default_bucket = sagemaker_session.default_bucket() pipeline_session = PipelineSession() model_package_group_name = f"AbaloneModelPackageGroupName"
Criar um pipeline
Importante
Políticas personalizadas do IAM que permitem que o Amazon SageMaker SageMaker Studio ou o Amazon Studio Classic criem SageMaker recursos da Amazon também devem conceder permissões para adicionar tags a esses recursos. A permissão para adicionar tags aos recursos é necessária porque o Studio e o Studio Classic marcam automaticamente todos os recursos que eles criam. Se uma política do IAM permitir que o Studio e o Studio Classic criem recursos, mas não permitisse a marcação, erros AccessDenied "" podem ocorrer ao tentar criar recursos. Para obter mais informações, consulte Forneça permissões para marcar recursos de SageMaker IA.
AWS políticas gerenciadas para Amazon SageMaker AIque dão permissões para criar SageMaker recursos já incluem permissões para adicionar tags ao criar esses recursos.
Execute as etapas a seguir em sua instância de notebook de SageMaker IA para criar um pipeline que inclua etapas para:
-
pré-processamento
-
treinamento
-
evaluation (avaliação)
-
avaliação condicional
-
registro de modelo
nota
Você pode usar ExecutionVariablesExecutionVariables
é resolvido em tempo de execução. Por exemplo, ExecutionVariables.PIPELINE_EXECUTION_ID
é resolvido com o ID da execução atual, que pode ser usado como um identificador exclusivo em diferentes execuções.
Etapa 1: baixe o conjunto de dados
Este caderno usa o conjunto de dados do UCI Machine Learning Abalone. O conjunto de dados contém os seguintes atributos:
-
length
: A medida de concha mais longa do abalone. -
diameter
: O diâmetro do abalone perpendicular ao seu comprimento. -
height
: A altura do abalone com carne na concha. -
whole_weight
: O peso do abalone inteiro. -
shucked_weight
: O peso da carne retirada do abalone. -
viscera_weight
: O peso das vísceras do abalone após o sangramento. -
shell_weight
: O peso da concha do abalone após a remoção e secagem da carne. -
sex
: O gênero do abalone. Entre 'M', 'F' ou 'I', em que 'I' é um abalone infantil. -
rings
: O número de anéis na concha do abalone.
O número de anéis na concha do abalone é uma boa aproximação de sua idade usando a fórmula age=rings + 1.5
. No entanto, obter esse número é uma tarefa demorada. Você deve cortar a concha pelo cone, marcar a seção e contar o número de anéis com um microscópio. No entanto, as outras medidas físicas são mais fáceis de conseguir. Este caderno usa o conjunto de dados para criar um modelo preditivo dos anéis variáveis usando as outras medidas físicas.
Para fazer download do conjunto de dados
-
Faça o download do conjunto de dados no bucket do Amazon S3 padrão da sua conta.
!mkdir -p data local_path = "data/abalone-dataset.csv" s3 = boto3.resource("s3") s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file( "dataset/abalone-dataset.csv", local_path ) base_uri = f"s3://{default_bucket}/abalone" input_data_uri = sagemaker.s3.S3Uploader.upload( local_path=local_path, desired_s3_uri=base_uri, ) print(input_data_uri)
-
Faça o download de um segundo conjunto de dados para transformação em lote após a criação do modelo.
local_path = "data/abalone-dataset-batch.csv" s3 = boto3.resource("s3") s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file( "dataset/abalone-dataset-batch", local_path ) base_uri = f"s3://{default_bucket}/abalone" batch_data_uri = sagemaker.s3.S3Uploader.upload( local_path=local_path, desired_s3_uri=base_uri, ) print(batch_data_uri)
Etapa 2: definir parâmetros do pipeline
Esse bloco de código define os seguintes parâmetros para seu pipeline:
-
processing_instance_count
: A contagem de instâncias do trabalho de processamento. -
input_data
: O local dos dados de entrada no Amazon S3. -
batch_data
: O local dos dados de entrada do Amazon S3 para transformação em lote. -
model_approval_status
: O status de aprovação com o qual registrar o modelo treinado para CI/CD. Para obter mais informações, consulte MLOps Automação com SageMaker projetos.
from sagemaker.workflow.parameters import ( ParameterInteger, ParameterString, ) processing_instance_count = ParameterInteger( name="ProcessingInstanceCount", default_value=1 ) model_approval_status = ParameterString( name="ModelApprovalStatus", default_value="PendingManualApproval" ) input_data = ParameterString( name="InputData", default_value=input_data_uri, ) batch_data = ParameterString( name="BatchData", default_value=batch_data_uri, )
Etapa 3: defina uma etapa de processamento para engenharia de atributos
Esta seção mostra como criar uma etapa de processamento para preparar os dados do conjunto de dados para treinamento.
Para criar uma etapa de processamento
-
Crie um diretório para o script de processamento.
!mkdir -p abalone
-
No diretório
/abalone
, crie um arquivo denominadopreprocessing.py
com o conteúdo a seguir. O script de pré-processamento é transmitido para a etapa de processamento para execução nos dados de entrada. Depois, a etapa de treinamento usa os rótulos e atributos de treinamento pré-processados para treinar um modelo. A etapa de avaliação usa o modelo treinado e os atributos e rótulos de teste pré-processados para avaliar o modelo. O script usascikit-learn
para fazer o seguinte:-
Preencha os dados categóricos
sex
ausentes e codifique-os para que sejam adequados para treinamento. -
Dimensione e normalize todos os campos numéricos, exceto
rings
esex
. -
Divida os dados em conjuntos de dados de treinamento, teste e validação.
%%writefile abalone/preprocessing.py import argparse import os import requests import tempfile import numpy as np import pandas as pd from sklearn.compose import ColumnTransformer from sklearn.impute import SimpleImputer from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler, OneHotEncoder # Because this is a headerless CSV file, specify the column names here. feature_columns_names = [ "sex", "length", "diameter", "height", "whole_weight", "shucked_weight", "viscera_weight", "shell_weight", ] label_column = "rings" feature_columns_dtype = { "sex": str, "length": np.float64, "diameter": np.float64, "height": np.float64, "whole_weight": np.float64, "shucked_weight": np.float64, "viscera_weight": np.float64, "shell_weight": np.float64 } label_column_dtype = {"rings": np.float64} def merge_two_dicts(x, y): z = x.copy() z.update(y) return z if __name__ == "__main__": base_dir = "/opt/ml/processing" df = pd.read_csv( f"{base_dir}/input/abalone-dataset.csv", header=None, names=feature_columns_names + [label_column], dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype) ) numeric_features = list(feature_columns_names) numeric_features.remove("sex") numeric_transformer = Pipeline( steps=[ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()) ] ) categorical_features = ["sex"] categorical_transformer = Pipeline( steps=[ ("imputer", SimpleImputer(strategy="constant", fill_value="missing")), ("onehot", OneHotEncoder(handle_unknown="ignore")) ] ) preprocess = ColumnTransformer( transformers=[ ("num", numeric_transformer, numeric_features), ("cat", categorical_transformer, categorical_features) ] ) y = df.pop("rings") X_pre = preprocess.fit_transform(df) y_pre = y.to_numpy().reshape(len(y), 1) X = np.concatenate((y_pre, X_pre), axis=1) np.random.shuffle(X) train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))]) pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False) pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False) pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)
-
-
Crie uma instância de um
SKLearnProcessor
para transmitir para a etapa de processamento.from sagemaker.sklearn.processing import SKLearnProcessor framework_version = "0.23-1" sklearn_processor = SKLearnProcessor( framework_version=framework_version, instance_type="ml.m5.xlarge", instance_count=processing_instance_count, base_job_name="sklearn-abalone-process", sagemaker_session=pipeline_session, role=role, )
-
Crie uma etapa de processamento. Essa etapa inclui o
SKLearnProcessor
, os canais de entrada e saída e o scriptpreprocessing.py
que você criou. Isso é muito semelhante aorun
método de uma instância de processador no SDK SageMaker AI Python. O parâmetroinput_data
transmitido paraProcessingStep
são os dados de entrada da própria etapa. Esses dados de entrada são usados pela instância do processador quando ela é executada.Observe os canais denominados
"train
,"validation
e"test"
especificados na configuração de saída do trabalho de processamento. A etapaProperties
, como essas, pode ser usada em etapas subsequentes e resolução para seus valores de runtime no runtime.from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep processor_args = sklearn_processor.run( inputs=[ ProcessingInput(source=input_data, destination="/opt/ml/processing/input"), ], outputs=[ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ], code="abalone/preprocessing.py", ) step_process = ProcessingStep( name="AbaloneProcess", step_args=processor_args )
Etapa 4: defina uma etapa de treinamento
Esta seção mostra como usar o XGBoostalgoritmo de SageMaker IA para treinar um modelo na saída de dados de treinamento das etapas de processamento.
Para definir uma etapa de treinamento
-
Especifique o caminho do modelo em que você deseja salvar os modelos do treinamento.
model_path = f"s3://{default_bucket}/AbaloneTrain"
-
Configure um estimador para o XGBoost algoritmo e o conjunto de dados de entrada. O tipo de instância de treinamento é transmitido para o estimador. Um roteiro de treinamento típico:
-
carrega dados dos canais de entrada
-
configura o treinamento com hiperparâmetros
-
treina um modelo
-
salva um modelo em
model_dir
para que ele possa ser hospedado posteriormente
SageMaker A IA carrega o modelo para o Amazon S3 na forma de
model.tar.gz
um no final do trabalho de treinamento.from sagemaker.estimator import Estimator image_uri = sagemaker.image_uris.retrieve( framework="xgboost", region=region, version="1.0-1", py_version="py3", instance_type="ml.m5.xlarge" ) xgb_train = Estimator( image_uri=image_uri, instance_type="ml.m5.xlarge", instance_count=1, output_path=model_path, sagemaker_session=pipeline_session, role=role, ) xgb_train.set_hyperparameters( objective="reg:linear", num_round=50, max_depth=5, eta=0.2, gamma=4, min_child_weight=6, subsample=0.7, silent=0 )
-
-
Crie um
TrainingStep
usando a instância do estimador e as propriedades deProcessingStep
. PasseS3Uri
do"train"
e o canal de saída"validation"
paraTrainingStep
.from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TrainingStep train_args = xgb_train.fit( inputs={ "train": TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train" ].S3Output.S3Uri, content_type="text/csv" ), "validation": TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "validation" ].S3Output.S3Uri, content_type="text/csv" ) }, ) step_train = TrainingStep( name="AbaloneTrain", step_args = train_args )
Etapa 5: defina uma etapa de processamento para avaliação do modelo
Esta seção mostra como criar uma etapa de processamento para avaliar a precisão do modelo. O resultado dessa avaliação de modelo é usado na etapa de condição para determinar qual caminho de execução seguir.
Para definir uma etapa de processamento para avaliação do modelo
-
Crie um arquivo denominado
evaluation.py
no diretório/abalone
. Esse script é usado em uma etapa de processamento para realizar a avaliação do modelo. Ele usa um modelo treinado e o conjunto de dados de teste como entrada e, em seguida, produz um arquivo JSON contendo métricas de avaliação de classificação.%%writefile abalone/evaluation.py import json import pathlib import pickle import tarfile import joblib import numpy as np import pandas as pd import xgboost from sklearn.metrics import mean_squared_error if __name__ == "__main__": model_path = f"/opt/ml/processing/model/model.tar.gz" with tarfile.open(model_path) as tar: tar.extractall(path=".") model = pickle.load(open("xgboost-model", "rb")) test_path = "/opt/ml/processing/test/test.csv" df = pd.read_csv(test_path, header=None) y_test = df.iloc[:, 0].to_numpy() df.drop(df.columns[0], axis=1, inplace=True) X_test = xgboost.DMatrix(df.values) predictions = model.predict(X_test) mse = mean_squared_error(y_test, predictions) std = np.std(y_test - predictions) report_dict = { "regression_metrics": { "mse": { "value": mse, "standard_deviation": std }, }, } output_dir = "/opt/ml/processing/evaluation" pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True) evaluation_path = f"{output_dir}/evaluation.json" with open(evaluation_path, "w") as f: f.write(json.dumps(report_dict))
-
Crie uma instância de um
ScriptProcessor
que seja usada para criar umaProcessingStep
.from sagemaker.processing import ScriptProcessor script_eval = ScriptProcessor( image_uri=image_uri, command=["python3"], instance_type="ml.m5.xlarge", instance_count=1, base_job_name="script-abalone-eval", sagemaker_session=pipeline_session, role=role, )
-
Crie um
ProcessingStep
usando a instância do processador, os canais de entrada e saída e o scriptevaluation.py
. Transmita:-
a propriedade
S3ModelArtifacts
da etapa de treinamentostep_train
-
o
S3Uri
do canal de saída"test"
da etapa de processamentostep_process
Isso é muito semelhante ao
run
método de uma instância de processador no SDK SageMaker AI Python.from sagemaker.workflow.properties import PropertyFile evaluation_report = PropertyFile( name="EvaluationReport", output_name="evaluation", path="evaluation.json" ) eval_args = script_eval.run( inputs=[ ProcessingInput( source=step_train.properties.ModelArtifacts.S3ModelArtifacts, destination="/opt/ml/processing/model" ), ProcessingInput( source=step_process.properties.ProcessingOutputConfig.Outputs[ "test" ].S3Output.S3Uri, destination="/opt/ml/processing/test" ) ], outputs=[ ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"), ], code="abalone/evaluation.py", ) step_eval = ProcessingStep( name="AbaloneEval", step_args=eval_args, property_files=[evaluation_report], )
-
Etapa 6: Definir uma CreateModelStep para transformação em lote
Importante
Recomendamos usar Etapa do modelo para criar modelos a partir da versão 2.90.0 do SDK do Python SageMaker . CreateModelStep
continuará funcionando nas versões anteriores do SDK do SageMaker Python, mas não tem mais suporte ativo.
Esta seção mostra como criar um modelo de SageMaker IA a partir do resultado da etapa de treinamento. Esse modelo é usado para transformação em lote em um novo conjunto de dados. Essa etapa é transmitida para a etapa de condição e só é executada se a etapa de condição for avaliada como true
.
Para definir uma CreateModelStep para transformação em lote
-
Crie um modelo de SageMaker IA. Transmita a propriedade
S3ModelArtifacts
a partir da etapa de treinamentostep_train
.from sagemaker.model import Model model = Model( image_uri=image_uri, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, sagemaker_session=pipeline_session, role=role, )
-
Defina a entrada do modelo para seu modelo de SageMaker IA.
from sagemaker.inputs import CreateModelInput inputs = CreateModelInput( instance_type="ml.m5.large", accelerator_type="ml.eia1.medium", )
-
Crie sua
CreateModelStep
usando aCreateModelInput
instância do modelo de SageMaker IA que você definiu.from sagemaker.workflow.steps import CreateModelStep step_create_model = CreateModelStep( name="AbaloneCreateModel", model=model, inputs=inputs, )
Etapa 7: Definir uma TransformStep para realizar a transformação em lote
Esta seção mostra como criar uma TransformStep
para realizar a transformação em lote em um conjunto de dados após o treinamento do modelo. Essa etapa é transmitida para a etapa de condição e só é executada se a etapa de condição for avaliada como true
.
Para definir uma TransformStep para realizar a transformação em lote
-
Crie uma instância transformadora com o tipo de instância de computação apropriado, a contagem de instâncias e o URI de saída desejado do bucket Amazon S3. Transmita a propriedade
ModelName
a partir das etapasstep_create_model
eCreateModel
.from sagemaker.transformer import Transformer transformer = Transformer( model_name=step_create_model.properties.ModelName, instance_type="ml.m5.xlarge", instance_count=1, output_path=f"s3://{default_bucket}/AbaloneTransform" )
-
Crie uma
TransformStep
usando a instância do transformador que você definiu e o parâmetro do pipelinebatch_data
.from sagemaker.inputs import TransformInput from sagemaker.workflow.steps import TransformStep step_transform = TransformStep( name="AbaloneTransform", transformer=transformer, inputs=TransformInput(data=batch_data) )
Etapa 8: Definir uma RegisterModel etapa para criar um pacote de modelo
Importante
Recomendamos usar Etapa do modelo para registrar modelos a partir da versão 2.90.0 do SDK do Python SageMaker . RegisterModel
continuará funcionando nas versões anteriores do SDK do SageMaker Python, mas não tem mais suporte ativo.
Esta seção mostra como criar uma instância de RegisterModel
. O resultado da execução do RegisterModel
em um pipeline é um pacote de modelo. Um pacote de modelo é uma abstração de artefatos de modelo reutilizável que empacota todos os ingredientes necessários para a inferência. Ele consiste em uma especificação de inferência que define a imagem de inferência a ser usada junto com uma localização opcional de pesos do modelo. Um grupo de pacotes de modelos é uma coleção de pacotes de modelos. Você pode usar um ModelPackageGroup
para que o Pipelines adicione uma nova versão e pacote de modelo ao grupo para cada execução de pipeline. Para obter mais informações sobre registro de modelos, consulte Implantação do registro do modelo com o Model Registry.
Essa etapa é transmitida para a etapa de condição e só é executada se a etapa de condição for avaliada como true
.
Para definir uma RegisterModel etapa para criar um pacote de modelo
-
Construa uma etapa
RegisterModel
usando a instância do estimador que você usou para a etapa de treinamento. Transmita a propriedadeS3ModelArtifacts
a partir da etapa de treinamentostep_train
e especifique umModelPackageGroup
. O Pipelines cria oModelPackageGroup
para você.from sagemaker.model_metrics import MetricsSource, ModelMetrics from sagemaker.workflow.step_collections import RegisterModel model_metrics = ModelMetrics( model_statistics=MetricsSource( s3_uri="
{}/evaluation.json
".format( step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"] ), content_type="application/json" ) ) step_register = RegisterModel( name="AbaloneRegisterModel
", estimator=xgb_train, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, content_types=["text/csv"], response_types=["text/csv"], inference_instances=["ml.t2.medium", "ml.m5.xlarge
"], transform_instances=["ml.m5.xlarge
"], model_package_group_name=model_package_group_name, approval_status=model_approval_status, model_metrics=model_metrics )
Etapa 9: defina uma etapa de condição para verificar a precisão do modelo
O ConditionStep
permite que o Pipelines ofereça compatibilidade com a execução condicional em seu DAG de pipeline com base na condição das propriedades da etapa. Nesse caso, você só registra um pacote de modelo se a precisão desse modelo exceder o valor necessário. A precisão do modelo é determinada pela etapa de avaliação de modelo. Se a precisão exceder o valor exigido, o pipeline também cria um modelo de SageMaker IA e executa a transformação em lote em um conjunto de dados. Esta seção mostra como definir a etapa de Condição.
Para definir uma etapa de condição para verificar a precisão do modelo
-
Defina uma condição
ConditionLessThanOrEqualTo
usando o valor de precisão encontrado na saída da etapa de processamento da avaliação do modelo,step_eval
. Obtenha essa saída usando o arquivo de propriedades que você indexou na etapa de processamento e o respectivo valor médio JSONPath do erro quadrático,."mse"
from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo from sagemaker.workflow.condition_step import ConditionStep from sagemaker.workflow.functions import JsonGet cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step_name=step_eval.name, property_file=evaluation_report, json_path="regression_metrics.mse.value" ), right=6.0 )
-
Construa uma
ConditionStep
. Transmita a condiçãoConditionEquals
e, em seguida, defina as etapas de registro do pacote modelo e de transformação em lote como as próximas etapas, caso a condição seja aprovada.step_cond = ConditionStep( name="AbaloneMSECond", conditions=[cond_lte], if_steps=[step_register, step_create_model, step_transform], else_steps=[], )
Etapa 10: criar um pipeline
Agora que você criou todas as etapas, combine-as em um pipeline.
Para criar um pipeline
-
Defina o seguinte para seu pipeline:
name
,parameters
esteps
. Os nomes devem ser exclusivos dentro de um par(account, region)
.nota
Uma etapa só pode aparecer uma vez na lista de etapas do pipeline ou nas listas de etapas hipotéticas da etapa de condição. Ela não pode aparecer em ambas.
from sagemaker.workflow.pipeline import Pipeline pipeline_name = f"AbalonePipeline" pipeline = Pipeline( name=pipeline_name, parameters=[ processing_instance_count, model_approval_status, input_data, batch_data, ], steps=[step_process, step_train, step_eval, step_cond], )
-
(Opcional) Examine a definição do pipeline JSON para garantir que ela esteja bem formada.
import json json.loads(pipeline.definition())
Essa definição de pipeline está pronta para ser enviada à SageMaker IA. No próximo tutorial, você envia esse pipeline para a SageMaker IA e inicia uma execução.
Você também pode usar o boto3
{'Version': '2020-12-01', 'Metadata': {}, 'Parameters': [{'Name': 'ProcessingInstanceType', 'Type': 'String', 'DefaultValue': 'ml.m5.xlarge'}, {'Name': 'ProcessingInstanceCount', 'Type': 'Integer', 'DefaultValue': 1}, {'Name': 'TrainingInstanceType', 'Type': 'String', 'DefaultValue': 'ml.m5.xlarge'}, {'Name': 'ModelApprovalStatus', 'Type': 'String', 'DefaultValue': 'PendingManualApproval'}, {'Name': 'ProcessedData', 'Type': 'String', 'DefaultValue': '
S3_URL
', {'Name': 'InputDataUrl', 'Type': 'String', 'DefaultValue': 'S3_URL
', 'PipelineExperimentConfig': {'ExperimentName': {'Get': 'Execution.PipelineName'}, 'TrialName': {'Get': 'Execution.PipelineExecutionId'}}, 'Steps': [{'Name': 'ReadTrainDataFromFS', 'Type': 'Processing', 'Arguments': {'ProcessingResources': {'ClusterConfig': {'InstanceType': 'ml.m5.4xlarge', 'InstanceCount': 2, 'VolumeSizeInGB': 30}}, 'AppSpecification': {'ImageUri': 'IMAGE_URI
', 'ContainerArguments': [....]}, 'RoleArn': 'ROLE
', 'ProcessingInputs': [...], 'ProcessingOutputConfig': {'Outputs': [.....]}, 'StoppingCondition': {'MaxRuntimeInSeconds': 86400}}, 'CacheConfig': {'Enabled': True, 'ExpireAfter': '30d'}}, ... ... ... }
Próxima etapa: Execute um pipeline