Defina um pipeline de construção de modelos

Para orquestrar seus fluxos de trabalho com o Amazon SageMaker Model Building Pipelines, você precisa gerar um gráfico acíclico direcionado (DAG) na forma de uma definição de pipeline JSON. A imagem a seguir é uma representação do pipeline DAG que você cria neste tutorial:

Um exemplo de gráfico acíclico direcionado por pipeline (DAG).

Você pode gerar sua definição de pipeline JSON usando o SDK do SageMaker Python. O tutorial a seguir mostra como gerar uma definição de tubulação para uma tubulação que resolve um problema de regressão para determinar a idade de um abalone com base em suas medidas físicas. Para um notebook Jupyter que inclui o conteúdo deste tutorial que você pode executar, consulte Orquestração de trabalhos com o Amazon SageMaker Model Building Pipelines.

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
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 responsável pelo SageMaker serviço seguindo as etapas em Modificar uma política de confiança de 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 seu ambiente

Crie uma nova SageMaker sessão 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 ter mais informações, consulte Forneça permissões para recursos de marcação SageMaker.

AWS Políticas gerenciadas para a Amazon SageMakerque dão permissões para criar SageMaker recursos já incluem permissões para adicionar tags ao criar esses recursos.

Execute as etapas a seguir na instância do seu SageMaker notebook para criar um pipeline, incluindo etapas de pré-processamento, treinamento, avaliação, avaliação condicional e registro do modelo.

Etapa 1: download do conjunto de dados

Este caderno usa o conjunto de dados do UCI Machine Learning Abalone. O conjunto de dados contém os seguintes recursos:

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 determinar. 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 ter mais informações, consulte Automatize MLOPs com projetos SageMaker .


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: definir uma etapa de processamento para engenharia de recursos

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 denominado preprocessing.py com o conteúdo a seguir. Esse script de pré-processamento é transmitido para a etapa de processamento para execução nos dados de entrada. Então, a etapa de treinamento usa os recursos e rótulos de treinamento pré-processados para treinar um modelo e a etapa de avaliação usa o modelo treinado e os recursos 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)

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 script preprocessing.py que você criou. Isso é muito semelhante ao run método de uma instância de processador no SDK do SageMaker 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, podem ser usadas em etapas subsequentes e resolvidas para seus valores de tempo de execução na execução.
```
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: definir uma etapa de treinamento

Esta seção mostra como usar o algoritmo SageMaker XGBoost 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 algoritmo XGBoost e o conjunto de dados de entrada. O tipo de instância de treinamento é transmitido para o estimador. Um script de treinamento típico carrega dados dos canais de entrada, configura o treinamento com hiperparâmetros, treina um modelo e salva um modelo para que ele model_dir possa ser hospedado posteriormente. SageMaker carrega o modelo no 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 uma TrainingStep usando a instância do estimador e as propriedades da ProcessingStep. Em particular, transmita o canal no 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: definir 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 do 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 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,
)

Crie um ProcessingStep usando a instância do processador, os canais de entrada e saída e o evaluation.py script. Em particular, passe a S3ModelArtifacts propriedade da etapa de step_train treinamento, bem como a S3Uri do canal de "test" saída da etapa de step_process processamento. Isso é muito semelhante ao run método de uma instância de processador no SDK do SageMaker 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 . CreateModelStepcontinuará funcionando nas versões anteriores do SDK do SageMaker Python, mas não tem mais suporte ativo.

Esta seção mostra como criar um SageMaker modelo a partir da saída 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 SageMaker modelo. 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,
)

Defina a entrada do modelo para seu SageMaker modelo.


from sagemaker.inputs import CreateModelInput


inputs = CreateModelInput(
    instance_type="ml.m5.large",
    accelerator_type="ml.eia1.medium",
)

Crie seu CreateModelStep usando a instância CreateModelInput e SageMaker modelo 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 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"
)

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 Model Package

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.

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 for SageMaker Pipelines para adicionar 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 Registrar e implantar modelos com o Registro 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 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. SageMaker O Pipelines cria isso 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: definir uma etapa de condição para verificar a precisão do modelo

A ConditionStep permite que os SageMaker pipelines suportem 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, conforme determinada pela etapa de avaliação do modelo, exceder o valor necessário. Se a precisão exceder o valor necessário, o pipeline também cria um SageMaker modelo 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 JSONPath do valor médio 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çã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

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

(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. No próximo tutorial, você envia esse pipeline SageMaker e inicia uma execução.

Próxima etapa: Execute um pipeline

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

Criar e gerenciar o Pipelines

Execute um pipeline