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: ![\[Captura de tela da drag-and-drop interface visual do Pipelines no Studio.\]](http://docs.aws.amazon.com/pt_br/sagemaker/latest/dg/images/pipelines/pipelines-studio-overview.png) 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](https://sagemaker-examples.readthedocs.io/en/latest/sagemaker-pipelines/tabular/abalone_build_train_deploy/sagemaker-pipelines-preprocess-train-evaluate-batch-transform.html) o Amazon Model Building Pipelines. SageMaker **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](https://github.com/aws-samples/customer-churn-sagemaker-pipelines-sample/blob/main/pipelines/customerchurn/pipeline.py) no Github. **Topics** ## Definir um pipeline (Pipeline Designer) 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: 1. Abra o console do Studio seguindo as instruções em [Inicie o Amazon SageMaker Studio](studio-updated-launch.md). 1. No painel de navegação à esquerda, selecione **Pipelines**. 1. Escolha **Criar**. 1. Escolha **Em branco**. 1. Na barra lateral esquerda, escolha **Processar dados** e arraste-o para o canvas. 1. No canvas, escolha a etapa **Processar dados** que você adicionou. 1. Para adicionar um conjunto de dados de entrada, escolha **Adicionar** em **Dados (entrada)** na barra lateral direita e selecione um conjunto de dados. 1. 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. 1. Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte [sagemaker.workflow.steps. ProcessingStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.ProcessingStep). ### Criar uma etapa de treinamento Para configurar uma etapa de treinamento de modelo, faça o seguinte: 1. Na barra lateral esquerda, escolha **Treinar modelo** e arraste-o até o canvas. 1. No canvas, escolha a etapa **Treinar modelo** que você adicionou. 1. Para adicionar um conjunto de dados de entrada, escolha **Adicionar** em **Dados (entrada)** na barra lateral direita e selecione um conjunto de dados. 1. 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. 1. Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte [sagemaker.workflow.steps. TrainingStep](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TrainingStep). 1. 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: 1. Na barra lateral esquerda, escolha **Registrar modelo** e arraste-o para o canvas. 1. No canvas, escolha a etapa **Registrar modelo** que você adicionou. 1. Para selecionar um modelo para registrar, escolha **Adicionar** em **Modelo (entrada)**. 1. Escolha **Criar um grupo de modelos** para adicionar seu modelo a um novo grupo de modelos. 1. Preencha os campos restantes na barra lateral direita. Para obter informações sobre os campos nessas guias, consulte [sagemaker.workflow.step\$1collections. RegisterModel](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_collections.RegisterModel). 1. 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: 1. Na barra lateral esquerda, escolha **Implantar modelo (endpoint)** e arraste-o até o canvas. 1. No canvas, escolha a etapa **Implantar modelo (endpoint)** que você adicionou. 1. Para escolher um modelo para implantar, escolha **Adicionar** em **Modelo (entrada)**. 1. Escolha o botão de opção **Criar endpoint** para criar um novo endpoint. 1. Insira um **Nome** e uma **Descrição** para o endpoint. 1. 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. 1. 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**. ## Definir um pipeline (SageMaker Python SDK) ### 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](https://docs.aws.amazon.com/sagemaker/latest/dg/howitworks-create-ws.html). 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](https://docs.aws.amazon.com/IAM/latest/UserGuide/roles-managingrole-editing-console.html#roles-modify_permissions-policy). Adicione o seguinte trecho de código do JSON para anexar essa política ao seu perfil: Substitua `` pelo ARN usado para criar sua instância de caderno. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "iam:GetRole", "iam:PassRole" ], "Resource": "arn:aws:iam::111122223333:role/role-name" } ] } ``` ------ + Confie no diretor do serviço de SageMaker IA seguindo as etapas em [Modificar a política de confiança de uma função](https://docs.aws.amazon.com/IAM/latest/UserGuide/roles-managingrole-editing-cli.html#roles-managingrole_edit-trust-policy-cli). 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" } ``` #### Configure 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](security_iam_id-based-policy-examples.md#grant-tagging-permissions). [AWS políticas gerenciadas para Amazon SageMaker AI](security-iam-awsmanpol.md)que 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 [ExecutionVariables](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#execution-variables)a função [Join](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#execution-variables) para especificar seu local de saída. `ExecutionVariables`é 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** 1. Faça o download do conjunto de dados no bucket do Amazon S3 padrão da sua conta. ``` !mkdir -p data local_path = "data/abalone-dataset.csv" s3 = boto3.resource("s3") s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file( "dataset/abalone-dataset.csv", local_path ) base_uri = f"s3://{default_bucket}/abalone" input_data_uri = sagemaker.s3.S3Uploader.upload( local_path=local_path, desired_s3_uri=base_uri, ) print(input_data_uri) ``` 1. 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](sagemaker-projects.md). ``` 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** 1. Crie um diretório para o script de processamento. ``` !mkdir -p abalone ``` 1. No diretório `/abalone`, crie um arquivo denominado `preprocessing.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 usa `scikit-learn` para fazer o seguinte: + Preencha os dados categóricos `sex` ausentes e codifique-os para que sejam adequados para treinamento. + Dimensione e normalize todos os campos numéricos, exceto `rings` e `sex`. + Divida os dados em conjuntos de dados de treinamento, teste e validação. ``` %%writefile abalone/preprocessing.py import argparse import os import requests import tempfile import numpy as np import pandas as pd from sklearn.compose import ColumnTransformer from sklearn.impute import SimpleImputer from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler, OneHotEncoder # Because this is a headerless CSV file, specify the column names here. feature_columns_names = [ "sex", "length", "diameter", "height", "whole_weight", "shucked_weight", "viscera_weight", "shell_weight", ] label_column = "rings" feature_columns_dtype = { "sex": str, "length": np.float64, "diameter": np.float64, "height": np.float64, "whole_weight": np.float64, "shucked_weight": np.float64, "viscera_weight": np.float64, "shell_weight": np.float64 } label_column_dtype = {"rings": np.float64} def merge_two_dicts(x, y): z = x.copy() z.update(y) return z if __name__ == "__main__": base_dir = "/opt/ml/processing" df = pd.read_csv( f"{base_dir}/input/abalone-dataset.csv", header=None, names=feature_columns_names + [label_column], dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype) ) numeric_features = list(feature_columns_names) numeric_features.remove("sex") numeric_transformer = Pipeline( steps=[ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()) ] ) categorical_features = ["sex"] categorical_transformer = Pipeline( steps=[ ("imputer", SimpleImputer(strategy="constant", fill_value="missing")), ("onehot", OneHotEncoder(handle_unknown="ignore")) ] ) preprocess = ColumnTransformer( transformers=[ ("num", numeric_transformer, numeric_features), ("cat", categorical_transformer, categorical_features) ] ) y = df.pop("rings") X_pre = preprocess.fit_transform(df) y_pre = y.to_numpy().reshape(len(y), 1) X = np.concatenate((y_pre, X_pre), axis=1) np.random.shuffle(X) train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))]) pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False) pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False) pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False) ``` 1. 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, ) ``` 1. Crie uma etapa de processamento. Essa etapa inclui o `SKLearnProcessor`, os canais de entrada e saída e o script `preprocessing.py` que você criou. Isso é muito semelhante ao `run` método de uma instância de processador no SDK SageMaker AI Python. O parâmetro `input_data` transmitido para `ProcessingStep` são os dados de entrada da própria etapa. Esses dados de entrada são usados pela instância do processador quando ela é executada. Observe os canais denominados `"train`, `"validation` e `"test"` especificados na configuração de saída do trabalho de processamento. A etapa `Properties`, 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](https://docs.aws.amazon.com/sagemaker/latest/dg/xgboost.html) de SageMaker IA para treinar um modelo na saída de dados de treinamento das etapas de processamento. **Para definir uma etapa de treinamento** 1. Especifique o caminho do modelo em que você deseja salvar os modelos do treinamento. ``` model_path = f"s3://{default_bucket}/AbaloneTrain" ``` 1. 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 ) ``` 1. Crie um `TrainingStep` usando a instância do estimador e as propriedades de `ProcessingStep`. Passe `S3Uri` do `"train"` e o canal de saída `"validation"` para `TrainingStep`.  ``` 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** 1. Crie um arquivo denominado `evaluation.py` no diretório `/abalone`. Esse script é usado em uma etapa de processamento para realizar a avaliação do modelo. Ele usa um modelo treinado e o conjunto de dados de teste como entrada e, em seguida, produz um 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)) ``` 1. Crie uma instância de um `ScriptProcessor` que seja usada para criar uma `ProcessingStep`. ``` from sagemaker.processing import ScriptProcessor script_eval = ScriptProcessor( image_uri=image_uri, command=["python3"], instance_type="ml.m5.xlarge", instance_count=1, base_job_name="script-abalone-eval", sagemaker_session=pipeline_session, role=role, ) ``` 1. Crie um `ProcessingStep` usando a instância do processador, os canais de entrada e saída e o script `evaluation.py`. Transmita: + a propriedade `S3ModelArtifacts` da etapa de treinamento `step_train` + o `S3Uri` do canal de saída `"test"` da etapa de processamento `step_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](build-and-manage-steps-types.md#step-type-model) 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** 1. Crie um modelo de SageMaker IA. Transmita a propriedade `S3ModelArtifacts` a partir da etapa de treinamento `step_train`. ``` from sagemaker.model import Model model = Model( image_uri=image_uri, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, sagemaker_session=pipeline_session, role=role, ) ``` 1. 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", ) ``` 1. Crie sua `CreateModelStep` usando a `CreateModelInput` 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** 1. 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 etapas `step_create_model` e `CreateModel`. ``` from sagemaker.transformer import Transformer transformer = Transformer( model_name=step_create_model.properties.ModelName, instance_type="ml.m5.xlarge", instance_count=1, output_path=f"s3://{default_bucket}/AbaloneTransform" ) ``` 1. Crie uma `TransformStep` usando a instância do transformador que você definiu e o parâmetro do pipeline `batch_data`. ``` from sagemaker.inputs import TransformInput from sagemaker.workflow.steps import TransformStep step_transform = TransformStep( name="AbaloneTransform", transformer=transformer, inputs=TransformInput(data=batch_data) ) ``` #### Etapa 8: Definir uma RegisterModel etapa para criar um pacote de modelo **Importante** Recomendamos usar [Etapa do modelo](build-and-manage-steps-types.md#step-type-model) 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](model-registry.md). 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 propriedade `S3ModelArtifacts` a partir da etapa de treinamento `step_train` e especifique um `ModelPackageGroup`. O Pipelines cria o `ModelPackageGroup` para você. ``` from sagemaker.model_metrics import MetricsSource, ModelMetrics from sagemaker.workflow.step_collections import RegisterModel model_metrics = ModelMetrics( model_statistics=MetricsSource( s3_uri="{}/evaluation.json".format( step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"] ), content_type="application/json" ) ) step_register = RegisterModel( name="AbaloneRegisterModel", estimator=xgb_train, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, content_types=["text/csv"], response_types=["text/csv"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name=model_package_group_name, approval_status=model_approval_status, model_metrics=model_metrics ) ``` #### 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** 1. Defina uma condição `ConditionLessThanOrEqualTo` usando o valor de precisão encontrado na saída da etapa de processamento da avaliação do modelo, `step_eval`. Obtenha essa saída usando o arquivo de propriedades que você indexou na etapa de processamento e o respectivo valor médio JSONPath do erro quadrático,. `"mse"` ``` from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo from sagemaker.workflow.condition_step import ConditionStep from sagemaker.workflow.functions import JsonGet cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step_name=step_eval.name, property_file=evaluation_report, json_path="regression_metrics.mse.value" ), right=6.0 ) ``` 1. Construa uma `ConditionStep`. Transmita a condição `ConditionEquals` e, em seguida, defina as etapas de registro do pacote modelo e de transformação em lote como as próximas etapas, caso a condição seja aprovada. ``` step_cond = ConditionStep( name="AbaloneMSECond", conditions=[cond_lte], if_steps=[step_register, step_create_model, step_transform], else_steps=[], ) ``` #### Etapa 10: criar um pipeline Agora que você criou todas as etapas, combine-as em um pipeline. **Para criar um pipeline** 1. Defina o seguinte para seu pipeline: `name`, `parameters` e `steps`. Os nomes devem ser exclusivos dentro de um par `(account, region)`. **nota** Uma etapa só pode aparecer uma vez na lista de etapas do pipeline ou nas listas de etapas hipotéticas da etapa de condição. Ela não pode aparecer em ambas. ``` from sagemaker.workflow.pipeline import Pipeline pipeline_name = f"AbalonePipeline" pipeline = Pipeline( name=pipeline_name, parameters=[ processing_instance_count, model_approval_status, input_data, batch_data, ], steps=[step_process, step_train, step_eval, step_cond], ) ``` 1. (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. ## Definir um pipeline (JSON) Você também pode usar o [boto3](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sagemaker.html#SageMaker.Client.create_pipeline) ou [CloudFormation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sagemaker-pipeline.html) para criar um pipeline. A criação de um pipeline requer uma definição de pipeline, que é um objeto JSON que define cada etapa do pipeline. O SageMaker SDK oferece uma maneira simples de criar a definição do pipeline, que você pode usar com qualquer uma das mencionadas APIs anteriormente para criar o próprio pipeline. Sem usar o SDK, os usuários precisam escrever a definição bruta do JSON para criar o pipeline sem nenhuma das verificações de erro fornecidas pelo SDK do Python SageMaker . Para ver o esquema da definição JSON do pipeline, consulte Esquema JSON de definição do [pipeline do SageMaker AI Pipeline](https://aws-sagemaker-mlops.github.io/sagemaker-model-building-pipeline-definition-JSON-schema/). O exemplo de código a seguir mostra um exemplo de um objeto JSON de definição de pipeline de SageMaker IA: ``` {'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](run-pipeline.md)