GS Caltex utiliza Amazon SageMaker Canvas y Amazon SageMaker Pipelines para conectar los modelos de los ingenieros de procesos con flujos de trabajo de producción. La empresa energética y química surcoreana divide el desarrollo y las operaciones de los modelos entre ingenieros con habilidades complementarias.

Los ingenieros de procesos crean y validan modelos en SageMaker Canvas sin escribir código, aprovechando su conocimiento del proceso. Los ingenieros de aprendizaje automático (ML) configuran flujos de trabajo de SageMaker Pipelines para implementación y inferencia programada.

Mover activos de SageMaker Canvas a SageMaker Pipelines aún requería trabajo de integración. Los ingenieros de ML tenían que establecer rutas de datos para producción, suministrar la configuración del modelo y conectar trabajos de preprocesamiento y inferencia. GS Caltex automatizó esta entrega mientras reutilizaba los activos creados en SageMaker Canvas.

GS Caltex informa que sus ingenieros de ML ahora implementan modelos en cuestión de horas, en comparación con alrededor de una semana antes. La empresa ejecuta inferencia por lotes cada 30 minutos, reduciendo la necesidad de puntos finales en ejecución continua. GS Caltex informa que esto ha reducido significativamente los costos de inferencia.

Esta publicación explica cómo GS Caltex conecta SageMaker Canvas con SageMaker Pipelines y reutiliza flujos de preprocesamiento validados y modelos en producción.

Visión general de la solución

El flujo de trabajo de operaciones de aprendizaje automático (MLOps) tiene dos caminos: preparar una nueva versión de modelo para implementación y ejecutar inferencia por lotes con el modelo implementado.

De un modelo validado a un flujo de trabajo repetible

Puede utilizar SageMaker Canvas para preparar datos de forma visual, crear modelos y generar predicciones por lotes. También puede registrar modelos validados en Amazon SageMaker Model Registry desde SageMaker Canvas. GS Caltex creó un proceso de entrega separado para convertir estos activos en pasos ejecutables de SageMaker Pipelines.

Los ingenieros de procesos entienden las variables y los datos detrás de un proceso y evalúan si las predicciones de un modelo son útiles. Los ingenieros de ML convierten los activos resultantes en trabajos de producción y administran versiones de modelo y registro de ejecución.

Arquitectura de referencia

El siguiente diagrama muestra cómo los ingenieros de procesos entregan activos de SageMaker Canvas a ingenieros de ML para implementación y inferencia programada. Los ingenieros de ML utilizan Amazon Simple Storage Service (Amazon S3) para flujos de preprocesamiento y Model Registry para versiones de modelo, y luego programan ejecuciones de pipeline a través de Amazon EventBridge.

Figura 1: Activos de SageMaker Canvas conectados a un flujo de trabajo de SageMaker Pipelines para inferencia por lotes programada

Los ingenieros de ML reciben el flujo de preprocesamiento y el modelo de SageMaker Canvas y revisan los formatos de datos y los resultados de predicción. Después de esa revisión, aprueban el modelo en Model Registry y lo implementan. Esta preparación tiene lugar cuando se implementa una nueva versión de modelo. Las ejecuciones de inferencia posteriores reutilizan el modelo implementado.

Amazon EventBridge inicia cada ejecución programada. SageMaker Pipelines prepara los datos de entrada desde Amazon Simple Storage Service (Amazon S3), ejecuta el preprocesamiento y SageMaker Batch Transform, y almacena los resultados en S3. Los ingenieros de ML pueden inspeccionar el estado de los pasos y los registros de Amazon CloudWatch para monitorear el flujo de trabajo.

La pipeline de producción reutiliza la lógica de preprocesamiento y el modelo entrenado creado en SageMaker Canvas. Los ingenieros de ML configuran las rutas de entrada y salida, establecen el orden de ejecución y escriben el código que conecta la aprobación y la programación al flujo de trabajo.

Requisitos previos

Antes de comenzar, prepare lo siguiente:

• Un modelo construido y evaluado en SageMaker Canvas, junto con su flujo de preprocesamiento de Data Wrangler
• Datos CSV para inferencia y un bucket de S3 para almacenar los archivos.
• Un rol de ejecución de SageMaker y un rol de Amazon EventBridge con permiso para iniciar el pipeline
• Un entorno de Python con Boto3 y SageMaker Python SDK 2.x instalados.

La implementación de referencia completa está disponible en el repositorio de GitHub. Los ejemplos de código en esta publicación se centran en las conexiones entre componentes. El repositorio utiliza un modelo de clasificación de préstamos de ejemplo, así que ajuste la configuración para que coincida con los formatos de entrada y salida de su modelo.

Implementación técnica

Los siguientes pasos cubren la reutilización de preprocesamiento, la aprobación y implementación del modelo, la construcción del pipeline y la inferencia programada.

1. Ejecutar el flujo de preprocesamiento de SageMaker Canvas con SageMaker Processing

Un flujo de preprocesamiento creado con Data Wrangler en SageMaker Canvas registra las fuentes de datos y los pasos de transformación. Para configurar la implementación de referencia, exporte datos a Amazon S3 utilizando SageMaker Processing, luego recupere la configuración del trabajo y el archivo .flow utilizado por ese trabajo.

En la configuración de exportación, desactive Configuración de trabajo automática y seleccione SageMaker Processing. Después de que comience la exportación, abra los detalles del trabajo de procesamiento en la consola de Amazon SageMaker AI. Registre la URI de la imagen del contenedor, la URI de S3 de la entrada con nombre flow, el rol de ejecución y la configuración de salida. Utilice estos valores para configurar el paso del pipeline que ejecuta el archivo .flow.

El rol de ejecución de SageMaker debe tener permisos para leer los datos de S3 requeridos, escribir los resultados y acceder a la imagen del contenedor. Elija el tipo y la cantidad de instancias en función del volumen de datos que necesite procesar.

Mantenga la lógica de preprocesamiento validada por el ingeniero de procesos y ajuste las fuentes de datos, los caminos y otros ajustes para el entorno de producción. Verifique la imagen del contenedor y el nodo de salida contra la configuración del trabajo original.

Algunos archivos .flow contienen metadatos de ejecución como internal_metadata.dw_job.container_uri. Verifique la configuración real del trabajo de procesamiento en lugar de suponer que cada exportación contiene este campo. Si mueve el flujo de trabajo a otra región de AWS, verifique que la imagen del contenedor esté disponible allí.

La implementación de referencia separa las salidas intermedias y finales por ejecución bajo runs/<execution-id>/. Su paso PrepareInput escribe datos en una ubicación de S3 fija referenciada por el archivo .flow. Evite que las ejecuciones que comparten esa ruta de entrada se superpongan.

Para aislar los datos de entrada para cada ejecución también, agregue un paso de preparación que actualice la URI de origen en una copia del archivo .flow. Declarar un parámetro de SageMaker Pipelines no reemplaza automáticamente una URI dentro del archivo.

2. Manejar las diferencias entre las entradas de entrenamiento y inferencia

La columna de destino es una de las diferencias entre los datos de entrenamiento y inferencia. Un flujo construido con datos de entrenamiento puede contener transformaciones que esperan la columna de destino en el esquema de entrada. Los datos de inferencia no contienen ese resultado conocido.

La implementación de referencia agrega un paso PrepareInput antes del preprocessamiento. Si falta la columna de destino, este paso la agrega con un valor de marcador de posición. Después del preprocessamiento, DropTargetColumn elimina la columna. Coincida con el tipo de datos y la posición de la columna del marcador de posición con el esquema de entrenamiento y pase solo las características esperadas al modelo en su orden original.

PrepareInput -> PreprocessNewData -> DropTargetColumn
             -> BatchPredict -> AssembleResults

Utilice este enfoque solo cuando el destino del placeholder no afecte los valores de la característica. Un flujo que filtra filas por el destino, deriva características de él o aplica codificación de destino necesita un tratamiento adicional. En esos casos, separe las transformaciones y el estado aprendido requerido para la inferencia.

La validación debe cubrir el orden y el recuento de columnas, los tipos de datos, el manejo de valores perdidos y el recuento de filas. Después de eliminar el destino, verifique que las columnas restantes coincidan con el orden de características utilizado durante el entrenamiento.

3. Registrar, aprobar y implementar el modelo

Registre la versión de modelo de SageMaker Canvas validada en el Registro de modelos. La versión registrada puede incluir los archivos del modelo, la configuración de inferencia y los resultados de evaluación. Los ingenieros de ML revisan esta información antes de cambiar el estado de aprobación.

El archivo de modelo, como model.tar.gz, debe usarse con el contenedor de inferencia y la configuración de entorno correctos. Un paquete de modelo en el Registro de modelos agrupa esta información. En esta arquitectura de lote, la implementación crea un recurso de modelo de SageMaker a partir de un paquete aprobado para que Batch Transform lo reference.

El siguiente ejemplo verifica un paquete que ya ha sido aprobado, luego crea el recurso de modelo. Proporcione model_package_arn, model_name y sagemaker_role_arn durante la implementación. Este ejemplo utiliza un nombre de modelo único para cada versión.

import boto3

sm_client = boto3.client("sagemaker")
package = sm_client.describe_model_package(
    ModelPackageName=model_package_arn
)

if package["ModelPackageStatus"] != "Completed":
    raise RuntimeError("Model package is not ready")
if package["ModelApprovalStatus"] != "Approved":
    raise RuntimeError("Model package is not approved")

sm_client.create_model(
    ModelName=model_name,
    ExecutionRoleArn=sagemaker_role_arn,
    PrimaryContainer={"ModelPackageName": model_package_arn},
)

El Registro de modelos registra el estado de aprobación, pero no aplica una verificación de aprobación en cada solicitud de inferencia. Verifique ese estado en el código de implementación y conserve el ARN del paquete seleccionado y el nombre del modelo en la configuración de la canalización. Si se revoca la aprobación, actualice los horarios y las configuraciones de ejecución que utilizan el modelo.

El registro y la implementación tienen lugar cuando se libera una versión del modelo. La canalización que se ejecuta cada 30 minutos crea trabajos de Batch Transform utilizando ese modelo, por lo que no necesita registrar el modelo nuevamente en cada ejecución.

4. Conectar la preprocesamiento, la inferencia y la ensambladura de resultados

La canalización de inferencia tiene cinco pasos.

Configure cada trabajo antes de conectar los pasos. En el siguiente ejemplo, prepare_args, preprocess_args, drop_target_args y assemble_args contienen los argumentos de paso devueltos por las llamadas run() en procesadores que utilizan un PipelineSession. Los procesadores y transformer comparten el objeto pipeline_session. Configure el transformer con el model_name previamente implementado.

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TransformStep

prepare = ProcessingStep(
    name="PrepareInput", step_args=prepare_args
)
preprocess = ProcessingStep(
    name="PreprocessNewData",
    step_args=preprocess_args,
    depends_on=[prepare],
)
drop_target = ProcessingStep(
    name="DropTargetColumn",
    step_args=drop_target_args,
    depends_on=[preprocess],
)

# The DropTargetColumn job defines an output named "features".
features_uri = (
    drop_target.properties.ProcessingOutputConfig
    .Outputs["features"].S3Output.S3Uri
)
predict = TransformStep(
    name="BatchPredict",
    step_args=transformer.transform(
        data=features_uri,
        content_type="text/csv",
        split_type="Line",
    ),
)
assemble = ProcessingStep(
    name="AssembleResults",
    step_args=assemble_args,
    depends_on=[predict],
)

pipeline = Pipeline(
    name="canvas-process-inference",
    steps=[prepare, preprocess, drop_target, predict, assemble],
    sagemaker_session=pipeline_session,
)
pipeline_arn = pipeline.upsert(role_arn=sagemaker_role_arn)["PipelineArn"]

El siguiente gráfico de ejecución muestra los cinco pasos en el flujo de trabajo de inferencia del modelo de préstamo de muestra. Preparan las entradas, aplican la preprocesamiento, eliminan la columna de destino, ejecutan la inferencia por lotes y ensamblan los resultados. La ejecución estaba en el estado Executing cuando se tomó esta captura de pantalla.

Figura 2: Gráfico de ejecución de SageMaker Pipelines para el modelo de préstamo de muestra

Utilice depends_on para controlar el orden de ejecución y pase los datos a través de las propiedades de salida de paso. Por ejemplo, assemble_args debe recibir la ubicación de los datos preprocesados y predict.properties.TransformOutput.S3OutputPath. El nombre de salida "features" pertenece a DropTargetColumn. El paso Data Wrangler utiliza el ID de salida definido en su flujo de preprocesamiento.

En la implementación de referencia, AssembleResults combina los datos preprocesados y las predicciones por posición de fila. Antes de utilizarlo en producción, agregue controles para que coincidan el orden de archivo y los recuentos de filas, y detenga la generación de resultados si no coinciden. Si el preprocesamiento ordena o elimina filas, mantenga un mapeo explícito entre los ID de registros de origen y las predicciones.

5. Programar la inferencia cada 30 minutos

GS Caltex ejecuta inferencia cada 30 minutos. El siguiente ejemplo utiliza una regla de Amazon EventBridge para programar ejecuciones de SageMaker Pipelines. Cree la canalización y verifíquela con una ejecución manual antes de adjuntar el programa.

import boto3

events_client = boto3.client("events")
rule_name = "canvas-process-inference-every-30-minutes"

events_client.put_rule(
    Name=rule_name,
    ScheduleExpression="rate(30 minutes)",
    State="ENABLED",
)

result = events_client.put_targets(
    Rule=rule_name,
    Targets=[{
        "Id": "canvas-process-inference",
        "Arn": pipeline_arn,
        "RoleArn": eventbridge_role_arn,
    }],
)
if result["FailedEntryCount"]:
    raise RuntimeError(result["FailedEntries"])

Establezca eventbridge_role_arn en un rol que confíe en el principal de servicio de Amazon EventBridge, events.amazonaws.com, y tenga permiso sagemaker:StartPipelineExecution para la canalización de destino. El valor pipeline_arn identifica la canalización creada en el paso anterior.

La expresión rate(30 minutes) programa ejecuciones a intervalos de 30 minutos. Para ejecutar en la hora y media hora, use cron(0/30 * * * ? *). Decida qué ventana de datos procesa cada ejecución en el paso de preparación de entrada y regístrela con la ejecución.

Consideraciones operativas

Antes de habilitar ejecuciones programadas, verifique el costo total, el tiempo de ejecución y el manejo de datos inválidos o no coincidentes.

Evaluar costos por lotes en todo el flujo de trabajo

Amazon SageMaker Pipelines es un servicio de orquestación de flujo de trabajo sin servidor. Administra el orden de ejecución y las dependencias sin un servidor de flujo de trabajo que deba aprovisionar o mantener. Los trabajos de procesamiento y Batch Transform utilizan recursos de cómputo separados y se facturan según el tipo de instancia y la duración.

Un punto final de tiempo real aprovisionado es adecuado para solicitudes que necesitan una respuesta de baja latencia consistente. Batch Transform es adecuado para cargas de trabajo que pueden procesar un conjunto de registros juntos. Libera los recursos de cómputo del trabajo cuando el trabajo finaliza, por lo que puede ejecutar predicciones sin mantener un punto final de inferencia en ejecución continuamente para cada modelo.

Un calendario de 30 minutos produce 48 ejecuciones programadas por día. Incluya el preprocesamiento, el inicio de instancia, la inferencia y el posprocesamiento al estimar el tiempo de ejecución total. La comparación de costos también debe tener en cuenta el Procesamiento, S3, CloudWatch y otros servicios utilizados por el flujo de trabajo.

Evalúe el costo de cada carga de trabajo utilizando su uso medido, incluidos los servicios de soporte y el tiempo de ejecución completo de cada trabajo programado.

Verificar el tiempo de ejecución y las ejecuciones superpuestas

Un calendario que inicia una ejecución cada 30 minutos no garantiza que cada ejecución finalice dentro de ese intervalo. Mida el tiempo de ejecución completo del pipeline con datos representativos, incluido el inicio de instancia y el tiempo necesario para reintentaar trabajos fallidos.

Decida cómo manejar una ejecución programada que llega antes de que la ejecución anterior haya finalizado. Controlar los pasos paralelos dentro de un pipeline no evita que las ejecuciones de pipeline separadas se superpongan. Verifique las ejecuciones activas, evite el procesamiento duplicado de la misma ventana de entrada y separe las rutas de salida para cada ejecución.

Registre la ventana de datos de entrada, la versión del flujo de preprocesamiento, el paquete de modelo ARN, el ARN de ejecución del pipeline y la ubicación de salida para cada ejecución. Estos detalles ayudan a reproducir resultados e investigar errores.

Validar salidas y comportamiento de fallas

Complete las siguientes verificaciones antes de habilitar el calendario:

1. Ejecute un conjunto de datos de validación a través de SageMaker Canvas y Procesamiento. Compare las columnas de salida, los tipos de datos, los recuentos de filas y los valores transformados clave.
2. Verifique que agregar un objetivo de marcador de posición no cambie los valores de las características y que el objetivo se elimine de la entrada del modelo.
3. Ejecute Batch Transform con la versión de modelo aprobada y verifique la asociación entre los registros de origen y las predicciones.
4. Confirme que un esquema de entrada inválido o una falla de preprocesamiento evite la inferencia y la publicación de resultados posteriores.
5. Mida el tiempo de ejecución y el costo, pruebe las ejecuciones superpuestas y el reprocesamiento de la misma ventana de entrada, y luego habilite el calendario de 30 minutos.

Agregue alertas de CloudWatch para fallas de trabajo y retrasos de procesamiento, junto con el monitoreo de la calidad de los datos o el modelo según sea necesario. Establezca umbrales iniciales a partir del tiempo de ejecución medido y las características de los datos, y revíselos cuando el modelo o el volumen de datos cambie.

Enfoque de adopción y resultados

Con una implementación escalonada, un equipo puede verificar la transferencia antes de estandarizarla en todos los modelos. Las siguientes secciones describen un enfoque de adopción y los resultados que informó GS Caltex.

Pasar de desarrollo de modelo a operación regular

Comience validando la ruta a producción para un modelo, luego estandarice los pasos que se repiten en las implementaciones. La siguiente secuencia es un enfoque de adopción sugerido basado en la experiencia de GS Caltex.

A medida que se expande la automatización, los ingenieros de procesos continúan evaluando si las predicciones son útiles para el proceso. Los ingenieros de ML gestionan la ejecución del pipeline y los cambios.

Resultados en GS Caltex

GS Caltex redujo el trabajo repetido necesario para conectar los modelos de SageMaker Canvas a producción. Los ingenieros de ML ahora implementan modelos en unas pocas horas, en comparación con aproximadamente una semana anteriormente. También reutilizan los activos que los ingenieros de procesos validaron en SageMaker Canvas.

Los ingenieros de procesos comenzaron a proponer aplicaciones en otros procesos después de ver sus modelos utilizados en producción. Los ingenieros de ML dedicaron menos tiempo a repetir el trabajo de integración y pudieron centrarse en operar y monitorear los pipelines.

Limpieza de recursos de prueba

Para una implementación de prueba, desactive la regla de Amazon EventBridge y elimine sus objetivos primero. Después de que las ejecuciones de pipeline y trabajos activos hayan terminado, elimine el pipeline de prueba y sus recursos de modelo de SageMaker.

Si creó un endpoint en tiempo real para pruebas, elimínelo también. Quite los archivos de entrada, intermedios y predicción de S3 y limpie los registros de CloudWatch según su política de retención. Verifique si otras ejecuciones o requisitos de auditoría aún dependen de los archivos del modelo y las versiones del Model Registry antes de eliminarlos.

Conclusión

En GS Caltex, los ingenieros de procesos crean y validan modelos en SageMaker Canvas, y los ingenieros de ML utilizan esos activos para ejecutar flujos de trabajo de producción en SageMaker Pipelines. Esta entrega redujo el trabajo de integración repetido. GS Caltex informa que los ingenieros de ML ahora completan las implementaciones de modelos en unas pocas horas, en comparación con aproximadamente una semana antes.

La entrega debe incluir el flujo de preprocesamiento, el esquema de entrada, la configuración de inferencia y el proceso de aprobación junto con los archivos del modelo. Verifique que el flujo de trabajo reproduzca los resultados esperados y pueda ejecutarse de manera confiable en el programa requerido.

Para probar este enfoque, comience con un modelo que haya validado en SageMaker Canvas y utilice el repositorio de GitHub. Consulte la guía para desarrolladores de SageMaker Pipelines para obtener detalles sobre la definición y ejecución de flujos de trabajo. Verifique los resultados de preprocesamiento y predicción por lotes con una ejecución manual, luego programe las ejecuciones en el intervalo que necesite su aplicación.

En la parte 2, mostramos cómo GS Caltex utilizó Amazon Bedrock para automatizar el traslado de modelos existentes a esta canalización.

Acerca de los autores