Requisitos previos Creación de una canalización

Defina una canalización de creación de modelos

Para organizar sus flujos de trabajo con Amazon SageMaker Model Building Pipelines, debe generar un gráfico acíclico dirigido (DAG) en forma de definición de canalización en JSON. La siguiente imagen es una representación del DAG de canalización que ha creado en este tutorial:

Un ejemplo de gráfico acíclico (DAG) dirigido a una canalización.

Puedes generar tu definición de canalización de JSON mediante el SDK de SageMaker Python. El siguiente tutorial muestra cómo generar una definición de canalización para una canalización que resuelva un problema de regresión para determinar la edad de un abulón en función de sus medidas físicas. Para ver un cuaderno de Jupyter que incluya el contenido de este tutorial y que pueda ejecutar, consulte Cómo organizar tareas con Amazon SageMaker Model Building Pipelines.

Requisitos previos

Para ejecutar el siguiente tutorial, debe hacer lo siguiente:

Configure la instancia del cuaderno tal y como se describe en Create a notebook instance. Esto le da a su rol permisos para leer y escribir en Amazon S3 y crear trabajos de formación, transformación por lotes y procesamiento en Amazon SageMaker.
Conceda permisos a su cuaderno para obtener y transferir su propio rol, tal y como se muestra en Modificación de una política de permisos de rol. Agregue el siguiente fragmento de JSON para asociar esta política a su rol. Sustituya <your-role-arn> por el ARN utilizado para crear la instancia del cuaderno.
```
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "iam:GetRole",
                "iam:PassRole"
            ],
            "Resource": "<your-role-arn>"
        }
    ]
}
```
Confíe en el director del SageMaker servicio siguiendo los pasos que se indican en Modificación de la política de confianza de un rol. Agregue el siguiente fragmento de declaración a la relación de confianza de su rol:
```
{
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "sagemaker.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
```

Configurar su entorno

Cree una nueva SageMaker sesión mediante el siguiente bloque de código. Esto devuelve el ARN del rol de la sesión. El ARN de este rol debe ser el ARN del rol de ejecución que haya configurado como requisito previo.


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"

Creación de una canalización

importante

Las políticas de IAM personalizadas que permiten a Amazon SageMaker Studio o Amazon SageMaker Studio Classic crear SageMaker recursos de Amazon también deben conceder permisos para añadir etiquetas a esos recursos. El permiso para añadir etiquetas a los recursos es obligatorio porque Studio y Studio Classic etiquetan automáticamente todos los recursos que crean. Si una política de IAM permite a Studio y Studio Classic crear recursos, pero no permite el etiquetado, se pueden producir errores AccessDenied «» al intentar crear recursos. Para obtener más información, consulte Proporcione permisos para etiquetar SageMaker los recursos.

AWS Políticas gestionadas para Amazon SageMakerque otorgan permisos para crear SageMaker recursos ya incluyen permisos para añadir etiquetas al crear esos recursos.

Ejecuta los siguientes pasos desde la instancia de tu SageMaker bloc de notas para crear una canalización, incluidos los pasos para el preprocesamiento, la formación, la evaluación, la evaluación condicional y el registro del modelo.

Paso 1: Descargar el conjunto de datos

Este cuaderno utiliza el conjunto de datos de abulón de UCI Machine Learning. El conjunto de datos contiene las siguientes características:

length: la medida más larga de la concha del abulón.
diameter: el diámetro del abulón perpendicular a su longitud.
height: La talla del abulón con carne en la concha.
whole_weight: el peso de todo el abulón.
shucked_weight: el peso de la carne extraída del abulón.
viscera_weight: el peso de las vísceras del abulón tras el sangrado.
shell_weight: el peso de la concha de abulón tras retirar la carne y secarla.
sex: el sexo del abulón. Puede ser “M”, “F” o “I”, don “I” es un abulón cría.
rings: el número de anillos de la concha del abulón.

El número de anillos de la concha del abulón es una buena indicación de su edad según la fórmula age=rings + 1.5. Sin embargo, la obtención de este número es una tarea que lleva mucho tiempo. Debe cortar la concha a través del cono, teñir la sección y contar el número de anillos con un microscopio. Sin embargo, las demás medidas físicas son más fáciles de determinar. Este cuaderno utiliza el conjunto de datos para crear un modelo predictivo de los anillos variables a partir de las demás medidas físicas.

Para descargar los conjuntos de datos

Descargue el conjunto de datos en el bucket de Amazon S3 predeterminado de su cuenta.


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

Descargue un segundo conjunto de datos para la transformación por lotes una vez creado el 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)

Paso 2: Definir los parámetros de la canalización

Este bloque de código define los siguientes parámetros para la canalización:

processing_instance_count: el recuento de instancias del trabajo de procesamiento.
input_data: la ubicación de los datos de entrada en Amazon S3.
batch_data: la ubicación de los datos de entrada en Amazon S3 para la transformación por lotes.
model_approval_status: el estado de aprobación para registrar el modelo entrenado para la CI/CD. Para obtener más información, consulte Automatice los MLOP con proyectos 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,
)

Paso 3: Definir un paso de procesamiento para la ingeniería de características

En esta sección se muestra cómo crear un paso de procesamiento para preparar los datos del conjunto de datos para el entrenamiento.

Para crear un paso de procesamiento

Cree un directorio para el script de procesamiento.
```
!mkdir -p abalone
```

En el directorio /abalone, cree un archivo denominado preprocessing.py con el contenido siguiente. Este script de preprocesamiento se transfiere al paso de procesamiento para su ejecución en los datos de entrada. A continuación, el paso de entrenamiento usa las características y etiquetas de entrenamiento preprocesadas para entrenar un modelo, y el paso de evaluación usa el modelo entrenado y las características y etiquetas de prueba preprocesadas para evaluar el modelo. El script usa scikit-learn para hacer lo siguiente:

Rellenar los datos categóricos sex ausentes y codificarlos para que sean adecuados para el entrenamiento.
Escalar y normalizar todos los campos numéricos excepto rings y sex.
Dividir los datos en conjuntos de entrenamiento, validación y prueba.


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

Cree una instancia de un SKLearnProcessor para transferirla al paso de procesamiento.


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

Cree un paso de procesamiento. Este paso incluye SKLearnProcessor, los canales de entrada y salida y el script preprocessing.py que ha creado. Es muy similar al run método de una instancia de procesador en el SDK de SageMaker Python. El parámetro input_data que se pasa a ProcessingStep son los datos de entrada del propio paso. La instancia del procesador utiliza estos datos de entrada cuando se ejecuta.

Observe los canales denominados "train, "validation y "test" especificados en la configuración de salida para el trabajo de procesamiento. Las Properties del paso como estas pueden utilizarse en pasos posteriores y se resuelven a sus valores de tiempo de ejecución en la ejecución.
```
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
)
```

Paso 4: Definir un paso de entrenamiento

En esta sección se muestra cómo utilizar el algoritmo SageMaker XGBoost para entrenar un modelo a partir de los datos de entrenamiento generados por los pasos de procesamiento.

Para definir un paso de entrenamiento

Especifique la ruta del modelo en la que desea guardar los modelos del entrenamiento.
```
model_path = f"s3://{default_bucket}/AbaloneTrain"
```

Configure un estimador para el algoritmo XGBoost y el conjunto de datos de entrada. El tipo de instancia de entrenamiento se transfiere al estimador. Un script de entrenamiento típico carga los datos de los canales de entrada, configura el entrenamiento con hiperparámetros, entrena un modelo y model_dir lo guarda para poder alojarlo más adelante. SageMaker carga el modelo en Amazon S3 en forma de a model.tar.gz al final del trabajo de formación.


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
)

Cree un TrainingStep mediante la instancia del estimador y las propiedades del ProcessingStep. En particular, pase el S3Uri de los canales de salida "train" y "validation" al 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
)

Paso 5: Definir un paso de procesamiento para la evaluación del modelo

En esta sección se muestra cómo crear un paso de procesamiento para evaluar la precisión del modelo. El resultado de la evaluación del modelo se utiliza en el paso de condición para determinar qué ruta de ejecución se debe seguir.

Para definir un paso de procesamiento para la evaluación del modelo

En el directorio /abalone cree un archivo denominado evaluation.py. Este script se utiliza en un paso de procesamiento para realizar la evaluación del modelo. Toma un modelo entrenado y el conjunto de datos de prueba como entrada y, a continuación, produce un archivo JSON que contiene las métricas de evaluación de la clasificación.


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

Cree una instancia de un ScriptProcessor que se utilice para crear unProcessingStep.


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

Cree una ProcessingStep utilizando la instancia del procesador, los canales de entrada y salida y el evaluation.py script. En concreto, transfiera la S3ModelArtifacts propiedad de la etapa de step_train entrenamiento, así como la S3Uri del canal de "test" salida de la etapa de step_process procesamiento. Es muy similar al run método de una instancia de procesador en el SDK de 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],
)

Paso 6: Definir una transformación CreateModelStep para Batch

importante

Recomendamos usarlo Paso de modelo para crear modelos a partir de la versión 2.90.0 del SDK de Python SageMaker . CreateModelStepseguirá funcionando en las versiones anteriores del SDK de SageMaker Python, pero ya no es compatible activamente.

En esta sección se muestra cómo crear un SageMaker modelo a partir del resultado del paso de entrenamiento. Este modelo se utiliza para la transformación por lotes en un nuevo conjunto de datos. Este paso se transfiere al paso de condición y solo se ejecuta si el paso de condición se evalúa como true.

Para definir una transformación CreateModelStep por lotes

Cree un SageMaker modelo. Transfiera la propiedad S3ModelArtifacts desde el paso de entrenamiento 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 la entrada del modelo para su SageMaker modelo.


from sagemaker.inputs import CreateModelInput


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

Cree la suya CreateModelStep utilizando la instancia del SageMaker modelo CreateModelInput y la que definió.


from sagemaker.workflow.steps import CreateModelStep


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

Paso 7: TransformStep Definir una transformación por lotes para realizar

En esta sección se muestra cómo crear un TransformStep para realizar la transformación por lotes en un conjunto de datos después de entrenar el modelo. Este paso se transfiere al paso de condición y solo se ejecuta si el paso de condición se evalúa como true.

Para definir una transformación por lotes TransformStep para realizar

Cree una instancia de transformador con el tipo de instancia de cómputo, el recuento de instancias y el URI de bucket de Amazon S3 de salida adecuados. Transfiera la propiedad ModelName desde el paso CreateModel de step_create_model.


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

Cree un TransformStep con la instancia de transformador que definió y el parámetro de canalización 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)
)

Paso 8: Defina un RegisterModel paso para crear un paquete modelo

importante

Recomendamos usarlo Paso de modelo para registrar modelos a partir de la versión 2.90.0 del SDK de Python SageMaker . RegisterModelseguirá funcionando en las versiones anteriores del SDK de SageMaker Python, pero ya no es compatible activamente.

En esta sección se muestra cómo construir una instancia deRegisterModel. El resultado de la ejecución de RegisterModel en una canalización es un paquete de modelos. Un paquete de modelos es una abstracción de artefactos de modelos reutilizable que empaqueta todos los ingredientes necesarios para la inferencia. Consiste en una especificación de inferencia que define la imagen de inferencia que se va a utilizar junto con una ubicación opcional de las ponderaciones del modelo. Un grupo de paquetes de modelos es una colección de paquetes de modelos. Puedes usar ModelPackageGroup for SageMaker Pipelines para añadir una nueva versión y un paquete de modelos al grupo por cada ejecución de la canalización. Para obtener más información acerca del registro de modelos, consulte Registro e implementación de modelos con el registro de modelos.

Este paso se transfiere al paso de condición y solo se ejecuta si el paso de condición se evalúa como true.

Para definir un RegisterModel paso para crear un paquete modelo

Construya un paso RegisterModel con la instancia del estimador que utilizó para el paso de entrenamiento. Transfiera la propiedad S3ModelArtifacts desde el paso de entrenamiento step_train y especifique un ModelPackageGroup. SageMaker Pipelines lo crea ModelPackageGroup por usted.


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
)

Paso 9: Definir un paso de condición para verificar la precisión del modelo

A ConditionStep permite a SageMaker Pipelines admitir la ejecución condicional en tu DAG de canalización en función del estado de las propiedades de los pasos. En este caso, solo le interesa registrar un paquete de modelos si la precisión de ese modelo, determinada por el paso de evaluación del modelo, supera el valor requerido. Si la precisión supera el valor requerido, la canalización también crea un SageMaker modelo y ejecuta la transformación por lotes en un conjunto de datos. En esta sección se muestra cómo definir el paso de condición.

Para definir un paso de condición para verificar la precisión del modelo

Defina una condición ConditionLessThanOrEqualTo con el valor de precisión que se encuentra en al salida del paso de procesamiento de la evaluación del modelo, step_eval. Obtenga esta salida a través del archivo de propiedades que indexó en el paso de procesamiento y el JSONPath respectivo del valor del error cuadrático medio, "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
)

Construya un ConditionStep. Pase la condición ConditionEquals y, a continuación, establezca los pasos de registro del paquete de modelos y transformación por lotes como los siguientes pasos si se cumple la condición.
```
step_cond = ConditionStep(
    name="AbaloneMSECond",
    conditions=[cond_lte],
    if_steps=[step_register, step_create_model, step_transform],
    else_steps=[], 
)
```

Paso 10: Crear una canalización

Ahora que ha creado todos los pasos, debe combinarlos en una canalización.

Para crear una canalización

Defina lo siguiente para su canalización: name, parameters y steps. Los nombres deben ser únicos en un par (account, region).

nota

Un paso solo puede aparecer una vez en la lista de pasos de la canalización o en las listas de pasos if/else del paso de condición. No puede aparecer en 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],
)

De forma opcional, examine la definición de canalización de JSON para asegurarse de que esté bien formada.
```
import json

json.loads(pipeline.definition())
```

Esta definición de canalización está lista para enviarse a SageMaker. En el siguiente tutorial, debe enviar esta canalización SageMaker e iniciar una ejecución.

Siguiente paso: Ejecución de una canalización

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

Cree y administre canalizaciones

Ejecución de una canalización