Adición de un paso

En la interfaz de usuario de arrastrar y soltar de Canalizaciones, puede usar un paso Ejecutar código para ejecutar su propio código como un paso de canalización. Puede cargar una función, un script o un cuaderno de Python para que se ejecute como parte de su canalización. Debe utilizar este paso si quiere orquestar un trabajo de ML personalizado que utilice las características avanzadas de SageMaker AI u otros servicios de AWS.

El paso Ejecutar código carga los archivos a su bucket de Amazon S3 predeterminado para Amazon SageMaker AI. Es posible que este bucket no tenga establecidos los permisos de uso compartido de recursos entre orígenes (CORS). Para obtener más información sobre cómo configurar permisos de CORS, consulte Requisito de CORS para datos de imagen de entrada.

El paso Ejecutar código utiliza un trabajo de entrenamiento de Amazon SageMaker para ejecutar el código. Asegúrese de que su rol de IAM tenga los permisos de la API sagemaker:DescribeTrainingJob y sagemaker:CreateTrainingJob. Para obtener más información sobre todos los permisos necesarios para Amazon SageMaker AI y cómo configurarlos, consulte Permisos de la API de Amazon SageMaker AI: referencia de acciones, permisos y recursos.

Para añadir un paso de ejecución de código a una canalización mediante el diseñador de canalizaciones, haga lo siguiente:

Utilice un paso de procesamiento para crear un trabajo de procesamiento para el procesamiento de datos. Para obtener más información sobre los trabajos de procesamiento, consulte Process Data and Evaluate Models.

Puede usar un paso de entrenamiento para crear un trabajo de entrenamiento para entrenar un modelo. Para obtener más información sobre los trabajos de entrenamiento, consulte Entrenamiento de un modelo con Amazon SageMaker AI.

Un paso de entrenamiento requiere un estimador, así como la introducción y la validación de los datos de entrenamiento.

Se utiliza un paso de ajuste para crear un trabajo de ajuste de hiperparámetros, también conocido como optimización de hiperparámetros (HPO). Un trabajo de ajuste de hiperparámetros ejecuta varios trabajos de entrenamiento, cada uno de los cuales produce una versión del modelo. Para obtener más información sobre el ajuste de hiperparámetros, consulte Ajuste automático del modelo con SageMaker AI.

El trabajo de ajuste está asociado al experimento de SageMaker AI para la canalización. Los trabajos de entrenamiento se crean como pruebas. Para obtener más información, consulte Integración de experimentos.

Un paso de ajuste requiere un HyperparameterTuner y entradas de entrenamiento. Puede volver a entrenar los trabajos de ajuste anteriores si especifica el parámetro warm_start_config del HyperparameterTuner. Para obtener más información sobre el inicio en caliente y el ajuste de hiperparámetros, consulte Ejecución de un trabajo de ajuste de hiperparámetros de inicio en caliente.

Use el método get_top_model_s3_uri de la clase sagemaker.workflow.steps.TuningStep para obtener el artefacto del modelo de una de las versiones del modelo con mejor rendimiento. Para ver un cuaderno que muestre cómo utilizar un paso de ajuste en una canalización de SageMaker AI, consulte sagemaker-pipelines-tuning-step.ipynb.

El siguiente ejemplo muestra cómo crear una definición de TuningStep.

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep

tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession())
    
step_tuning = TuningStep(
    name = "HPTuning",
    step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://amzn-s3-demo-bucket/my-data"))
)

Obtención de la mejor versión del modelo

El siguiente ejemplo muestra cómo obtener la mejor versión del modelo a partir del trabajo de ajuste mediante el método get_top_model_s3_uri. Como máximo, están disponibles las 50 versiones con mejor rendimiento, clasificadas según HyperParameterTuningJobObjective. El argumento top_k es un índice de las versiones, donde top_k=0 es la versión con mejor rendimiento y top_k=49 es la versión con peor rendimiento.

best_model = Model(
    image_uri=image_uri,
    model_data=step_tuning.get_top_model_s3_uri(
        top_k=0,
        s3_bucket=sagemaker_session.default_bucket()
    ),
    ...
)

Para obtener más información sobre los requisitos del paso de ajuste, consulte la documentación de sagemaker.workflow.steps.TuningStep.

El afinamiento entrena un modelo fundacional previamente entrenado de Amazon SageMaker JumpStart en un nuevo conjunto de datos. Este proceso, también conocido como aprendizaje por transferencia, puede producir modelos precisos con conjuntos de datos más pequeños y con menos tiempo de entrenamiento. Al afinar un modelo, puede usar el conjunto de datos predeterminado o elegir sus propios datos. Para obtener más información sobre cómo afinar un modelo fundacional desde JumpStart, consulte Ajuste de un modelo.

El paso de afinamiento utiliza un trabajo de entrenamiento de Amazon SageMaker para personalizar el modelo. Asegúrese de que su rol de IAM tenga los permisos de API sagemaker:DescribeTrainingJob y sagemaker:CreateTrainingJob para ejecutar el trabajo de afinamiento en su canalización. Para obtener más información sobre los permisos necesarios para Amazon SageMaker AI y cómo configurarlos, consulte Permisos de la API de Amazon SageMaker AI: referencia de acciones, permisos y recursos.

Para añadir un paso Afinar modelo a su canalización mediante el editor de arrastrar y soltar, siga estos pasos:

Utilice la API de AutoML para crear un trabajo de AutoML para entrenar automáticamente un modelo. Para obtener más información sobre los trabajos de AutoML, consulte Automate model development with Amazon SageMaker Autopilot.

El siguiente ejemplo muestra cómo crear una definición mediante AutoMLStep.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.workflow.automl_step import AutoMLStep

pipeline_session = PipelineSession()

auto_ml = AutoML(...,
    role="<role>",
    target_attribute_name="my_target_attribute_name",
    mode="ENSEMBLING",
    sagemaker_session=pipeline_session) 

input_training = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-validation-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="validation",
)

step_args = auto_ml.fit(
    inputs=[input_training, input_validation]
)

step_automl = AutoMLStep(
    name="AutoMLStep",
    step_args=step_args,
)     

Obtención de la mejor versión del modelo

El paso AutoML entrena automáticamente a varios candidatos de modelos. Obtenga el modelo con la mejor métrica objetiva del trabajo AutoML utilizando el método get_best_auto_ml_model siguiente. También debe utilizar un role de IAM para acceder a los artefactos del modelo.

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

Para obtener más información, consulte el paso AutoML del SageMaker Python SDK.

Utilice un ModelStep para crear o registrar un modelo de SageMaker AI. Para obtener más información sobre los requisitos de ModelStep, consulte la documentación de sagemaker.workflow.model_step.ModelStep.

Crear un modelo

Puede utilizar un ModelStep para crear un modelo de SageMaker AI. Un ModelStep requiere artefactos del modelo e información sobre el tipo de instancia de SageMaker AI que debe usar para crear el modelo. Para obtener más información sobre los modelos de SageMaker AI, consulte Entrenamiento de un modelo con Amazon SageMaker AI.

El siguiente ejemplo muestra cómo crear una definición de ModelStep.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.model import Model
from sagemaker.workflow.model_step import ModelStep

step_train = TrainingStep(...)
model = Model(
    image_uri=pytorch_estimator.training_image_uri(),
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=PipelineSession(),
    role=role,
)

step_model_create = ModelStep(
   name="MyModelCreationStep",
   step_args=model.create(instance_type="ml.m5.xlarge"),
)

Registrar un modelo

Puede utilizar un ModelStep para registrar un sagemaker.model.Model o un sagemaker.pipeline.PipelineModel en el Registro de modelos de Amazon SageMaker. Un PipelineModel representa una canalización de inferencia, que es un modelo compuesto de una secuencia lineal de contenedores que procesan solicitudes de inferencia. Para obtener más información sobre cómo registrar un modelo, consulte Implementación del registro de modelos con el registro de modelos.

El siguiente ejemplo muestra cómo crear un ModelStep que registre un PipelineModel.

import time

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

pipeline_session = PipelineSession()

code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix)

sklearn_model = SKLearnModel(
   model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri,
   entry_point='inference.py',
   source_dir='sklearn_source_dir/',
   code_location=code_location,
   framework_version='1.0-1',
   role=role,
   sagemaker_session=pipeline_session,
   py_version='py3'
)

xgboost_model = XGBoostModel(
   model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
   entry_point='inference.py',
   source_dir='xgboost_source_dir/',
   code_location=code_location,
   framework_version='0.90-2',
   py_version='py3',
   sagemaker_session=pipeline_session,
   role=role
)

from sagemaker.workflow.model_step import ModelStep
from sagemaker import PipelineModel

pipeline_model = PipelineModel(
   models=[sklearn_model, xgboost_model],
   role=role,sagemaker_session=pipeline_session,
)

register_model_step_args = pipeline_model.register(
    content_types=["application/json"],
   response_types=["application/json"],
   inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
   transform_instances=["ml.m5.xlarge"],
   model_package_group_name='sipgroup',
)

step_model_registration = ModelStep(
   name="AbaloneRegisterModel",
   step_args=register_model_step_args,
)

Utilice un paso Crear modelo para crear un modelo de SageMaker AI. Para obtener más información sobre los modelos de SageMaker AI, consulte Entrenar un modelo con Amazon SageMaker..

El paso Crear modelo requiere artefactos del modelo e información sobre el tipo de instancia de SageMaker AI que tiene que utilizar para crear el modelo. Los siguientes ejemplos muestran cómo crear una definición del paso Crear modelo. Para obtener más información sobre los requisitos del paso Crear modelo, consulte la documentación de sagemaker.workflow.steps.CreateModelStep

El paso Registrar modelo registra un modelo en el Registro de modelos de Amazon SageMaker.

En el diseñador de canalizaciones, utilice el paso Implementar modelo (punto de conexión) para implementar el modelo en un punto de conexión. Puede crear un nuevo punto de conexión o utilizar un punto de conexión existente. La inferencia en tiempo real es idónea para cargas de trabajo de inferencia con requisitos en tiempo real, interactivos y de baja latencia. Puede implementar su modelo en los servicios de alojamiento de SageMaker AI y obtener un punto de conexión en tiempo real que pueda utilizarse para la inferencia. Estos puntos de conexión están totalmente administrados y admiten escalado automático. Para obtener más información sobre la inferencia en tiempo real en SageMaker AI, consulte Inferencia en tiempo real.

Antes de añadir un paso del modelo de implementación a su canalización, asegúrese de que su rol de IAM tenga los siguientes permisos:

Para obtener más información sobre todos los permisos necesarios para SageMaker AI y cómo configurarlos, consulte Permisos de la API de Amazon SageMaker AI: referencia de acciones, permisos y recursos.

Para añadir un paso Implementar modelo a su canalización mediante el editor de arrastrar y soltar, siga estos pasos:

Utilice un paso de transformación por lotes para ejecutar la inferencia en todo un conjunto de datos. Para obtener más información acerca de la transformación por lotes, consulte Transformaciones por lotes con canalizaciones de inferencia.

Un paso de transformación requiere un transformador y los datos sobre los que se ejecutará la transformación por lotes. El siguiente ejemplo muestra cómo crear una definición del paso Transformar. Para obtener más información sobre los requisitos del paso de entrenamiento, consulte la documentación de sagemaker.workflow.steps.TransformStep.

Se utiliza un paso de condición para evaluar la condición de las propiedades del paso y evaluar qué acción se debe realizar a continuación en la canalización.

Un paso de condición requiere lo siguiente:

Utilice un paso Callback para añadir procesos y servicios de AWS adicionales a su flujo de trabajo que Canalizaciones de Amazon SageMaker no proporciona directamente. Cuando se ejecuta un paso Callback, se lleva a cabo el siguiente procedimiento:

Para obtener más información sobre los requisitos del paso Callback, consulte la documentación de sagemaker.workflow.callback_step.CallbackStep. Para obtener una solución completa, consulte Extend SageMaker Pipelines to include custom steps using callback steps.

En el siguiente ejemplo, se muestra una implementación del procedimiento anterior.

from sagemaker.workflow.callback_step import CallbackStep

step_callback = CallbackStep(
    name="MyCallbackStep",
    sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue",
    inputs={...},
    outputs=[...]
)

callback_handler_code = '
    import boto3
    import json

    def handler(event, context):
        sagemaker_client=boto3.client("sagemaker")

        for record in event["Records"]:
            payload=json.loads(record["body"])
            token=payload["token"]

            # Custom processing

            # Call SageMaker AI to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

Comportamiento de detención

Un proceso de canalización no se detiene mientras se ejecuta un paso Callback.

Al llamar a StopPipelineExecution en un proceso de canalización con un paso Callback en ejecución, Canalizaciones envía un mensaje adicional de Amazon SQS a la cola de SQS especificada. El cuerpo del mensaje de SQS contiene un campo Estado, que se establece en Stopping. En el siguiente ejemplo se muestra un cuerpo de mensaje SQS.

{
  "token": "26vcYbeWsZ",
  "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a",
  "arguments": {
    "number": 5,
    "stringArg": "some-arg",
    "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv"
  },
  "status": "Stopping"
}

Debe añadir lógica a su consumidor de mensajes de Amazon SQS para tomar las medidas necesarias (por ejemplo, limpiar los recursos) al recibir el mensaje. A continuación, añada una llamada a SendPipelineExecutionStepSuccess o SendPipelineExecutionStepFailure.

Solo cuando Canalizaciones recibe una de estas llamadas detiene el proceso de canalización.

Se utiliza un paso Lambda para ejecutar una función AWS Lambda. Puede ejecutar una función de Lambda existente o SageMaker AI puede crear y ejecutar una función de Lambda nueva. Si decide utilizar una función de Lambda existente, debe estar en la misma Región de AWS que la canalización de SageMaker AI. Para ver un cuaderno que muestre cómo utilizar un paso Lambda en una canalización de SageMaker AI, consulte sagemaker-pipelines-lambda-step.ipynb.

SageMaker AI proporciona la clase sagemaker.lambda_helper.Lambda para crear, actualizar, invocar y eliminar funciones de Lambda. Lambda tiene la siguiente firma.

Lambda(
    function_arn,       # Only required argument to invoke an existing Lambda function

    # The following arguments are required to create a Lambda function:
    function_name,
    execution_role_arn,
    zipped_code_dir,    # Specify either zipped_code_dir and s3_bucket, OR script
    s3_bucket,          # S3 bucket where zipped_code_dir is uploaded
    script,             # Path of Lambda function script
    handler,            # Lambda handler specified as "lambda_script.lambda_handler"
    timeout,            # Maximum time the Lambda function can run before the lambda step fails
    ...
)

La clase sagemaker.workflow.lambda_step.LambdaStep tiene un argumento lambda_func de tipo Lambda. Para invocar una función de Lambda existente, el único requisito es proporcionar el nombre de recurso de Amazon (ARN) de la función al function_arn. Si no proporciona un valor para function_arn, debe especificar el handler y una de las siguientes opciones:

El siguiente ejemplo muestra cómo crear una definición de paso Lambda que invoque una función de Lambda existente.

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
        function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda"
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

El siguiente ejemplo muestra cómo crear una definición de paso Lambda que cree e invoque una función de Lambda mediante un script de función de Lambda.

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
      function_name="split-dataset-lambda",
      execution_role_arn=execution_role_arn,
      script="lambda_script.py",
      handler="lambda_script.lambda_handler",
      ...
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

Entradas y salidas

Si la función Lambda tiene entradas o salidas, estas también deben definirse en el paso Lambda.

Al definir el paso Lambda, inputs debe se un diccionario de pares clave-valor. Cada valor del diccionario de inputs debe ser de un tipo primitivo (cadena, entero o flotante). No se admiten objetos anidados. Si no se define, el valor de inputs predeterminado es None.

El valor de outputs debe ser una lista de claves. Estas claves hacen referencia a un diccionario definido en la salida de la función Lambda. Como en inputs, estas claves deben ser de tipos primitivos y no se admiten objetos anidados.

Comportamiento de tiempo de espera y detención

La clase Lambda tiene un argumento timeout que especifica el tiempo máximo que puede ejecutarse la función de Lambda. El valor predeterminado es de 120 segundos, con un máximo de 10 minutos. Si la función de Lambda se está ejecutando cuando se alcanza el tiempo de espera, el paso Lambda producirá un erro; sin embargo, la función de Lambda seguirá ejecutándose.

No se puede detener un proceso de canalización mientras se está ejecutando un paso Lambda porque la función de Lambda invocada por el paso Lambda no se puede detener. Si detiene el proceso mientras se ejecuta la función de Lambda, la canalización esperará a que finalice la función o hasta que se agote el tiempo de espera. Esto depende de lo que ocurra primero. A continuación, el proceso se detiene. Si la función de Lambda finaliza, el estado del proceso de canalización es Stopped. Si se agota el tiempo de espera, el estado del proceso de canalización es Failed.

Puede utilizar el paso ClarifyCheck para comprobar la desviación de la referencia con respecto a las líneas de base anteriores para analizar los sesgos y la explicabilidad del modelo. A continuación, puede generar y registrar las líneas de base con el método model.register() y pasar la salida de ese método a Paso de modelo con step_args. El Monitor de modelos de Amazon SageMaker puede utilizar estas referencias para los puntos de conexión del modelo. Como resultado, no necesita hacer una sugerencia de referencia por separado.

El paso ClarifyCheck también permite extraer líneas de base para la comprobación de desviaciones del registro del modelo. El paso ClarifyCheck utiliza el contenedor prediseñado de SageMaker Clarify. Este contenedor proporciona una serie de capacidades de supervisión de modelos, incluida la sugerencia de restricciones y la validación de restricciones con respecto a una referencia determinada. Para obtener más información, consulte Contenedores prediseñados de SageMaker Clarify.

Configuración del paso ClarifyCheck

Puede configurar el paso ClarifyCheck para que realice solo uno de los siguientes tipos de comprobaciones cada vez que se utilice en una canalización.

Para ello, defina el parámetro clarify_check_config con uno de los siguientes valores de tipo de comprobación:

El paso ClarifyCheck inicializa un trabajo de procesamiento que ejecuta el contenedor prediseñado de SageMaker AI Clarify y requiere configuraciones específicas para el trabajo de verificación y procesamiento. ClarifyCheckConfig y CheckJobConfig son funciones auxiliares para estas configuraciones. Estas funciones auxiliares están alineadas con la forma en que el trabajo de procesamiento de SageMaker Clarify computa para verificar el sesgo del modelo, el sesgo de los datos o la explicabilidad del modelo. Para obtener más información, consulte Ejecutar trabajos de procesamiento de SageMaker Clarify para obtener análisis del sesgo y explicabilidad.

Control del comportamiento del paso para comprobaciones de desviación

El paso ClarifyCheck requiere los dos indicadores booleanos siguientes para controlar su comportamiento:

Para obtener más información, consulte Cálculo de referencia, detección de desviaciones y ciclo de vida con los pasos ClarifyCheck y QualityCheck en Canalizaciones de Amazon SageMaker.

Trabajo con líneas de base

Puede especificar model_package_group_name para localizar la referencia existente. A continuación, el paso ClarifyCheck extrae DriftCheckBaselines en el último paquete de modelos aprobado del grupo de paquetes de modelos.

O bien, puede proporcionar una referencia anterior a través del parámetro supplied_baseline_constraints. Si especifica tanto model_package_group_name como supplied_baseline_constraints, el paso ClarifyCheck utiliza la referencia especificada por el parámetro supplied_baseline_constraints.

Para obtener más información sobre el uso de los requisitos del paso ClarifyCheck, consulte sagemaker.workflow.steps.ClarifyCheckStep en el Amazon SageMaker AI SageMaker AI SDK para Python. Para ver un cuaderno de Amazon SageMaker Studio Classic que muestra cómo usar el paso ClarifyCheck en Canalizaciones, consulte sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_bias_data_config = DataConfig(
    s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']),
    label=0,
    dataset_type="text/csv",
    s3_analysis_config_output_path=data_bias_analysis_cfg_output_path,
)

data_bias_config = BiasConfig(
    label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]]  
)

data_bias_check_config = DataBiasCheckConfig(
    data_config=data_bias_data_config,
    data_bias_config=data_bias_config,
)h

data_bias_check_step = ClarifyCheckStep(
    name="DataBiasCheckStep",
    clarify_check_config=data_bias_check_config,
    check_job_config=check_job_config,
    skip_check=False,
    register_new_baseline=False
   supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json",
    model_package_group_name="MyModelPackageGroup"
)

Utilice el paso QualityCheck para realizar sugerencias de referencia y comprobar las desviaciones con respecto a una referencia anterior para determinar la calidad de los datos o la calidad del modelo en una canalización. A continuación, puede generar y registrar las referencias con el método model.register() y pasar la salida de ese método a Paso de modelo con step_args.

El monitor de modelos utiliza estas referencias para la comprobación de desviaciones en los puntos de conexión del modelo, de manera que no sea necesario hacer una sugerencia de referencia aparte. El paso QualityCheck también permite extraer líneas de base para la comprobación de desviaciones del registro del modelo. El paso QualityCheck aprovecha el contenedor prediseñado del Monitor de modelos de Amazon SageMaker AI. Este contenedor dispone de una serie de funciones de supervisión del modelo que incluyen la sugerencia de restricciones, la generación de estadísticas y la validación de restricciones con respecto a una referencia. Para obtener más información, consulte Contenedor prediseñado del monitor de modelos de Amazon SageMaker.

Configuración del paso QualityCheck

Puede configurar el paso QualityCheck para que ejecute solo uno de los siguientes tipos de comprobaciones cada vez que se utilice en una canalización.

Para ello, defina el parámetro quality_check_config con uno de los siguientes valores de tipo de comprobación:

El paso QualityCheck inicia un trabajo de procesamiento que ejecuta el contenedor prediseñado del monitor de modelos y requiere configuraciones específicas para el trabajo de verificación y procesamiento. QualityCheckConfig y CheckJobConfig son funciones auxiliares para estas configuraciones. Estas funciones auxiliares están alineadas con la forma en que el Monitor de modelos crea una referencia para supervisar la calidad del modelo o la calidad de los datos. Para obtener más información sobre las sugerencias de referencia del monitor de modelos, consulte Crear una referencia y Creación de una referencia de calidad del modelo.

Control del comportamiento del paso para comprobaciones de desviación

El paso QualityCheck requiere los dos indicadores booleanos siguientes para controlar su comportamiento:

Para obtener más información, consulte Cálculo de referencia, detección de desviaciones y ciclo de vida con los pasos ClarifyCheck y QualityCheck en Canalizaciones de Amazon SageMaker.

Trabajo con líneas de base

Puede especificar una referencia anterior directamente a través de los parámetros supplied_baseline_statistics y supplied_baseline_constraints. También puede especificar model_package_group_name y el paso QualityCheck extrae las DriftCheckBaselines del paquete de modelos aprobado más reciente del grupo de paquetes de modelos.

Cuando especifica lo siguiente, el paso QualityCheck utiliza la referencia especificada por supplied_baseline_constraints y supplied_baseline_statistics en el tipo de comprobación del paso QualityCheck.

Para obtener más información sobre el uso de los requisitos del paso QualityCheck, consulte sagemaker.workflow.steps.QualityCheckStep en el Amazon SageMaker AI SageMaker AI SDK para Python. Para ver un cuaderno de Amazon SageMaker Studio Classic que muestra cómo usar el paso QualityCheck en Canalizaciones, consulte sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_quality_check_config = DataQualityCheckConfig(
    baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"),
    output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep'])
)

data_quality_check_step = QualityCheckStep(
    name="DataQualityCheckStep",
    skip_check=False,
    register_new_baseline=False,
    quality_check_config=data_quality_check_config,
    check_job_config=check_job_config,
    supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json",
    supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json",
    model_package_group_name="MyModelPackageGroup"
)

Utilice el paso EMR de Canalizaciones de Amazon SageMaker para:

Para obtener más información sobre Amazon EMR, consulte Introducción a Amazon EMR.

El paso EMR requiere que EMRStepConfig incluya la ubicación del archivo JAR que utilizará el clúster de Amazon EMR y cualquier argumento que se vaya a pasar. También debe proporcionar el ID del clúster de Amazon EMR si desea ejecutar el paso en un clúster de EMR en ejecución. También puede pasar la configuración del clúster para ejecutar el paso EMR en un clúster que cree, administre y finalice en su nombre. Las siguientes secciones incluyen ejemplos y enlaces a cuadernos de muestra que muestran ambos métodos.

Lanzamiento de un nuevo trabajo en un clúster de Amazon EMR en ejecución

Para inicializar un nuevo trabajo en un clúster de Amazon EMR en ejecución, debe pasar el ID del clúster en forma de cadena al argumento cluster_id del EMRStep. El siguiente ejemplo demuestra este procedimiento.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

step_emr = EMRStep (
    name="EMRSampleStep", # required
    cluster_id="j-1ABCDEFG2HIJK", # include cluster_id to use a running cluster
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

Para ver un ejemplo de cuaderno que le guíe a través de un ejemplo completo, consulte Pipelines EMR Step With Running EMR Cluster.

Lanzamiento de un nuevo trabajo en un clúster de Amazon EMR nuevo

Para inicializar un nuevo trabajo en un nuevo clúster que EMRStep cree para usted, proporcione la configuración de su clúster como diccionario. El diccionario debe tener la misma estructura que una solicitud RunJobFlow. Sin embargo, no incluya los siguientes campos en la configuración del clúster:

Todos los demás argumentos RunJobFlow están disponibles para su uso en la configuración del clúster. Para obtener más información sobre la sintaxis de la solicitud, consulte RunJobFlow.

El siguiente ejemplo pasa una configuración de clúster a una definición de paso EMR. Esto solicita al paso que inicialice un nuevo trabajo en un nuevo clúster de EMR. La configuración del clúster de EMR de este ejemplo incluye especificaciones para los nodos del clúster de EMR principales y básicos. Para obtener más información sobre los tipos de nodos de Amazon EMR, consulte Descripción de los tipos de nodos: principales, básicos y de tarea.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_step_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

# include your cluster configuration as a dictionary
emr_cluster_config = {
    "Applications": [
        {
            "Name": "Spark", 
        }
    ],
    "Instances":{
        "InstanceGroups":[
            {
                "InstanceRole": "MASTER",
                "InstanceCount": 1,
                "InstanceType": "m5.2xlarge"
            },
            {
                "InstanceRole": "CORE",
                "InstanceCount": 2,
                "InstanceType": "m5.2xlarge"
            }
        ]
    },
    "BootstrapActions":[],
    "ReleaseLabel": "emr-6.6.0",
    "JobFlowRole": "job-flow-role",
    "ServiceRole": "service-role"
}

emr_step = EMRStep(
    name="emr-step",
    cluster_id=None,
    display_name="emr_step",
    description="MyEMRStepDescription",
    step_config=emr_step_config,
    cluster_config=emr_cluster_config
)

Para ver un ejemplo de cuaderno que le guie a través de un ejemplo completo, consulte Pipelines EMR Step With Cluster Lifecycle Management.

Utilice NotebookJobStep para ejecutar su trabajo de cuaderno de SageMaker de forma no interactiva como un paso de la canalización. Si crea su canalización en la interfaz de usuario de arrastrar y soltar de Canalizaciones, use paso de ejecución del código para ejecutar su cuaderno. Para obtener más información acerca de los trabajos de cuadernos de SageMaker, consulte Trabajos del cuaderno de SageMaker.

Un NotebookJobStep requiere como mínimo un cuaderno de entrada, el URI de la imagen y el nombre del kernel. Para obtener más información sobre los requisitos del paso Trabajo de cuaderno y otros parámetros que puede configurar para personalizar su paso, consulte sagemaker.workflow.steps.NotebookJobStep.

En el siguiente ejemplo, se utilizan los argumentos mínimos para definir un NotebookJobStep.

from sagemaker.workflow.notebook_job_step import NotebookJobStep


notebook_job_step = NotebookJobStep(
    input_notebook=input_notebook,
    image_uri=image_uri,
    kernel_name=kernel_name
)

El paso de la canalización NotebookJobStep se trata como un trabajo de cuaderno de SageMaker. Como resultado, realice un seguimiento del estado de ejecución en el panel de trabajos del cuaderno de la IU de Studio Classic incluyendo etiquetas específicas con el argumento tags. Para obtener más información acerca de las etiquetas que se deben incluir, consulte Consulte los trabajos de cuaderno en el panel de la interfaz de usuario de Studio.

Además, si programa el trabajo del cuaderno con el SageMaker Python SDK, solo podrá especificar determinadas imágenes para ejecutar el trabajo del cuaderno. Para obtener más información, consulte Restricciones de imágenes para los trabajos de cuaderno del SageMaker AI Python SDK.

Utilice un paso Error para detener una ejecución de Canalizaciones de Amazon SageMaker cuando no se alcance una condición o un estado deseados. El paso Error también le permite introducir un mensaje de error personalizado en el que se indique la causa del error de la ejecución de la canalización.

Limitaciones del uso del paso Error

Puede crear el mensaje del paso Error en forma de cadena de texto estático. Como alternativa, también puede utilizar Parámetros de canalización, una operación de unión u otras propiedades del paso para crear un mensaje de error más informativo si usa el SDK.