Reduzca los costos y el tiempo de desarrollo con el modo local de Amazon SageMaker Pipelines

La creación de canalizaciones de aprendizaje automático (ML) robustas y reutilizables puede ser un proceso complejo y lento. Los desarrolladores generalmente prueban sus scripts de procesamiento y entrenamiento localmente, pero las canalizaciones en sí generalmente se prueban en la nube. Crear y ejecutar una canalización completa durante la experimentación agrega gastos generales y costos no deseados al ciclo de vida del desarrollo. En esta publicación, detallamos cómo puede usar Amazon SageMaker Pipelines modo local para ejecutar canalizaciones de ML localmente para reducir tanto el desarrollo de la canalización como el tiempo de ejecución al mismo tiempo que se reducen los costos. Después de que la canalización se haya probado completamente localmente, puede volver a ejecutarla fácilmente con Amazon SageMaker recursos administrados con solo unas pocas líneas de cambios de código.

Descripción general del ciclo de vida de ML

Uno de los principales impulsores de las nuevas innovaciones y aplicaciones en ML es la disponibilidad y la cantidad de datos junto con opciones informáticas más económicas. En varios dominios, ML ha demostrado ser capaz de resolver problemas que antes no podían resolverse con big data clásico y técnicas analíticas, y la demanda de profesionales de ciencia de datos y ML está aumentando constantemente. Desde un nivel muy alto, el ciclo de vida de ML consta de muchas partes diferentes, pero la construcción de un modelo de ML generalmente consta de los siguientes pasos generales:

1. Limpieza y preparación de datos (ingeniería de funciones)
2. Entrenamiento y ajuste de modelos
3. Evaluación modelo
4. Implementación de modelos (o transformación por lotes)

En el paso de preparación de datos, los datos se cargan, manipulan y transforman en el tipo de entradas o funciones que espera el modelo de ML. Escribir los scripts para transformar los datos suele ser un proceso iterativo, donde los bucles de retroalimentación rápidos son importantes para acelerar el desarrollo. Por lo general, no es necesario usar el conjunto de datos completo al probar los scripts de ingeniería de funciones, por lo que puede usar el función de modo local de procesamiento de SageMaker. Esto le permite ejecutar localmente y actualizar el código de forma iterativa, utilizando un conjunto de datos más pequeño. Cuando el código final está listo, se envía al trabajo de procesamiento remoto, que usa el conjunto de datos completo y se ejecuta en instancias administradas de SageMaker.

El proceso de desarrollo es similar al paso de preparación de datos tanto para el entrenamiento del modelo como para los pasos de evaluación del modelo. Los científicos de datos utilizan el función de modo local de SageMaker Training para iterar rápidamente con conjuntos de datos más pequeños localmente, antes de usar todos los datos en un clúster administrado por SageMaker de instancias optimizadas para ML. Esto acelera el proceso de desarrollo y elimina el costo de ejecutar instancias de ML administradas por SageMaker mientras se experimenta.

A medida que aumenta la madurez de ML de una organización, puede usar Canalizaciones de Amazon SageMaker para crear canalizaciones de ML que unen estos pasos, creando flujos de trabajo de ML más complejos que procesan, entrenan y evalúan modelos de ML. SageMaker Pipelines es un servicio completamente administrado para automatizar los diferentes pasos del flujo de trabajo de ML, incluida la carga de datos, la transformación de datos, la capacitación y el ajuste de modelos y la implementación de modelos. Hasta hace poco, podía desarrollar y probar sus scripts localmente, pero tenía que probar sus canalizaciones de ML en la nube. Esto hizo que la iteración en el flujo y la forma de las canalizaciones de ML fuera un proceso lento y costoso. Ahora, con la función de modo local agregada de SageMaker Pipelines, puede iterar y probar sus canalizaciones de ML de manera similar a como prueba e itera sus scripts de procesamiento y entrenamiento. Puede ejecutar y probar sus canalizaciones en su máquina local, utilizando un pequeño subconjunto de datos para validar la sintaxis y las funcionalidades de la canalización.

Tuberías de SageMaker

SageMaker Pipelines proporciona una forma totalmente automatizada de ejecutar flujos de trabajo de ML simples o complejos. Con SageMaker Pipelines, puede crear flujos de trabajo de ML con un SDK de Python fácil de usar y luego visualizar y administrar su flujo de trabajo usando Estudio Amazon SageMaker. Sus equipos de ciencia de datos pueden ser más eficientes y escalar más rápido almacenando y reutilizando los pasos del flujo de trabajo que crea en SageMaker Pipelines. También puede usar plantillas prediseñadas que automatizan la creación de infraestructuras y repositorios para crear, probar, registrar e implementar modelos dentro de su entorno de ML. Estas plantillas están disponibles automáticamente para su organización y se aprovisionan mediante Catálogo de servicios de AWS por la seguridad alimentaria

SageMaker Pipelines aporta prácticas de integración continua e implementación continua (CI/CD) a ML, como el mantenimiento de la paridad entre los entornos de desarrollo y producción, el control de versiones, las pruebas bajo demanda y la automatización integral, lo que le ayuda a escalar ML en todo su organización. Los profesionales de DevOps saben que algunos de los principales beneficios del uso de técnicas de CI/CD incluyen un aumento en la productividad a través de componentes reutilizables y un aumento en la calidad a través de pruebas automatizadas, lo que conduce a un ROI más rápido para sus objetivos comerciales. Estos beneficios ahora están disponibles para los profesionales de MLOps mediante el uso de SageMaker Pipelines para automatizar el entrenamiento, las pruebas y la implementación de modelos de ML. Con el modo local, ahora puede iterar mucho más rápido mientras desarrolla scripts para usar en una canalización. Tenga en cuenta que las instancias de canalización local no se pueden ver ni ejecutar dentro del IDE de Studio; sin embargo, pronto estarán disponibles opciones de visualización adicionales para las tuberías locales.

El SDK de SageMaker proporciona un propósito general configuración de modo local que permite a los desarrolladores ejecutar y probar procesadores y estimadores compatibles en su entorno local. Puede usar el entrenamiento en modo local con varias imágenes de marco compatibles con AWS (TensorFlow, MXNet, Chainer, PyTorch y Scikit-Learn), así como imágenes que usted mismo proporcione.

SageMaker Pipelines, que crea un gráfico acíclico dirigido (DAG) de pasos de flujo de trabajo orquestados, admite muchas actividades que forman parte del ciclo de vida de ML. En modo local, se admiten los siguientes pasos:

• Pasos de trabajo de procesamiento – Una experiencia administrada y simplificada en SageMaker para ejecutar cargas de trabajo de procesamiento de datos, como ingeniería de funciones, validación de datos, evaluación de modelos e interpretación de modelos
• Pasos de trabajo de entrenamiento – Un proceso iterativo que enseña a un modelo a hacer predicciones mediante la presentación de ejemplos de un conjunto de datos de entrenamiento
• Trabajos de ajuste de hiperparámetros – Una forma automatizada de evaluar y seleccionar los hiperparámetros que producen el modelo más preciso
• Pasos de ejecución condicionales – Un paso que proporciona una ejecución condicional de ramas en una canalización
• paso modelo – Usando argumentos CreateModel, este paso puede crear un modelo para usar en pasos de transformación o implementación posterior como un punto final
• Transformar pasos de trabajo – Un trabajo de transformación por lotes que genera predicciones a partir de grandes conjuntos de datos y ejecuta inferencias cuando no se necesita un punto final persistente
• Pasos fallidos – Un paso que detiene la ejecución de una canalización y marca la ejecución como fallida

Resumen de la solución

Nuestra solución demuestra los pasos esenciales para crear y ejecutar SageMaker Pipelines en modo local, lo que significa usar CPU local, RAM y recursos de disco para cargar y ejecutar los pasos del flujo de trabajo. Su entorno local podría estar ejecutándose en una computadora portátil, usando IDE populares como VSCode o PyCharm, o podría estar alojado en SageMaker usando instancias de portátiles clásicas.

El modo local permite a los científicos de datos unir pasos, que pueden incluir trabajos de procesamiento, capacitación y evaluación, y ejecutar todo el flujo de trabajo localmente. Cuando haya terminado de probar localmente, puede volver a ejecutar la canalización en un entorno administrado de SageMaker reemplazando el LocalPipelineSession objeto con PipelineSession, que aporta consistencia al ciclo de vida de ML.

Para esta muestra de cuaderno, usamos un conjunto de datos estándar disponible públicamente, el Conjunto de datos de abulón de aprendizaje automático de UCI. El objetivo es entrenar un modelo ML para determinar la edad de un caracol de abulón a partir de sus medidas físicas. En el fondo, este es un problema de regresión.

Todo el código necesario para ejecutar este ejemplo de bloc de notas está disponible en GitHub en el amazon-sagemaker-ejemplos repositorio. En este ejemplo de cuaderno, cada paso del flujo de trabajo de la canalización se crea de forma independiente y luego se conecta para crear la canalización. Creamos los siguientes pasos:

• Paso de procesamiento (ingeniería de características)
• Paso de entrenamiento (entrenamiento modelo)
• Paso de procesamiento (evaluación del modelo)
• Paso de condición (precisión del modelo)
• Crear paso de modelo (modelo)
• Paso de transformación (transformación por lotes)
• Registrar paso de modelo (paquete de modelo)
• Paso fallido (ejecución fallida)

El siguiente diagrama ilustra nuestra canalización.

Requisitos previos

Para seguir esta publicación, necesita lo siguiente:

Una vez que se cumplen estos requisitos previos, puede ejecutar el cuaderno de muestra como se describe en las siguientes secciones.

Construye tu canalización

En este ejemplo de cuaderno, usamos Modo de secuencia de comandos de SageMaker para la mayoría de los procesos de ML, lo que significa que proporcionamos el código Python real (scripts) para realizar la actividad y pasar una referencia a este código. El modo de secuencia de comandos brinda una gran flexibilidad para controlar el comportamiento dentro del procesamiento de SageMaker al permitirle personalizar su código mientras aprovecha los contenedores preconstruidos de SageMaker como XGBoost o Scikit-Learn. El código personalizado se escribe en un archivo de secuencia de comandos de Python usando celdas que comienzan con el comando mágico %%writefile, como el siguiente:

%%writefile code/evaluation.py

El habilitador principal del modo local es el LocalPipelineSession objeto, que se instancia desde Python SDK. Los siguientes segmentos de código muestran cómo crear una canalización de SageMaker en modo local. Aunque puede configurar una ruta de datos local para muchos de los pasos de la canalización local, Amazon S3 es la ubicación predeterminada para almacenar la salida de datos de la transformación. El nuevo LocalPipelineSession El objeto se pasa al SDK de Python en muchas de las llamadas a la API de flujo de trabajo de SageMaker descritas en esta publicación. Tenga en cuenta que puede utilizar el local_pipeline_session variable para recuperar referencias al depósito predeterminado de S3 y el nombre de la región actual.

from sagemaker.workflow.pipeline_context import LocalPipelineSession

# Create a `LocalPipelineSession` object so that each 
# pipeline step will run locally
# To run this pipeline in the cloud, you must change 
# the `LocalPipelineSession()` to `PipelineSession()`
local_pipeline_session = LocalPipelineSession()
region = local_pipeline_session.boto_region_name

default_bucket = local_pipeline_session.default_bucket()
prefix = "sagemaker-pipelines-local-mode-example"

Antes de crear los pasos de canalización individuales, establecemos algunos parámetros utilizados por la canalización. Algunos de estos parámetros son literales de cadena, mientras que otros se crean como tipos enumerados especiales proporcionados por el SDK. La tipificación enumerada garantiza que se proporcionen configuraciones válidas a la canalización, como esta, que se pasa a la ConditionLessThanOrEqualTo paso más abajo:

mse_threshold = ParameterFloat(name="MseThreshold", default_value=7.0)

Para crear un paso de procesamiento de datos, que se usa aquí para realizar la ingeniería de características, usamos el SKLearnProcessor para cargar y transformar el conjunto de datos. pasamos el local_pipeline_session variable al constructor de la clase, que indica al paso del flujo de trabajo que se ejecute en modo local:

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    role=role,
    sagemaker_session=local_pipeline_session,
)

A continuación, creamos nuestro primer paso de canalización real, un ProcessingStep objeto, tal como se importó desde SageMaker SDK. Los argumentos del procesador se devuelven de una llamada al SKLearnProcessor método ejecutar(). Este paso del flujo de trabajo se combina con otros pasos hacia el final del cuaderno para indicar el orden de operación dentro de la canalizació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="code/preprocessing.py",
)

step_process = ProcessingStep(name="AbaloneProcess", step_args=processor_args)

A continuación, proporcionamos un código para establecer un paso de entrenamiento creando primero una instancia de un estimador estándar mediante el SDK de SageMaker. Pasamos lo mismo local_pipeline_session variable al estimador, llamado xgb_train, como el sagemaker_session argumento. Como queremos entrenar un modelo XGBoost, debemos generar un URI de imagen válido especificando los siguientes parámetros, incluido el marco y varios parámetros de versión:

from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

model_path = f"s3://{default_bucket}/{prefix}/model"
image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.5-1",
    py_version="py3",
    instance_type=instance_type,
)

xgb_train = Estimator(
    image_uri=image_uri,
    entry_point="code/abalone.py",
    instance_type=instance_type,
    instance_count=training_instance_count,
    output_path=model_path,
    role=role,
    sagemaker_session=local_pipeline_session,
)

Opcionalmente podemos llamar a métodos de estimación adicionales, por ejemplo set_hyperparameters(), para proporcionar configuraciones de hiperparámetros para el trabajo de entrenamiento. Ahora que tenemos un estimador configurado, estamos listos para crear el paso de entrenamiento real. Una vez más, importamos el TrainingStep clase de la biblioteca SageMaker SDK:

from sagemaker.workflow.steps import TrainingStep

step_train = TrainingStep(name="AbaloneTrain", step_args=train_args)

A continuación, construimos otro paso de procesamiento para realizar la evaluación del modelo. Esto se hace creando un ScriptProcessor instancia y pasando el local_pipeline_session Objeto como parámetro:

from sagemaker.processing import ScriptProcessor

script_eval = ScriptProcessor(
    image_uri=image_uri,
    command=["python3"],
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="script-abalone-eval",
    role=role,
    sagemaker_session=local_pipeline_session,
)

Para permitir la implementación del modelo entrenado, ya sea en un Punto final en tiempo real de SageMaker o a una transformación por lotes, necesitamos crear un Model objeto pasando los artefactos del modelo, el URI de la imagen adecuada y, opcionalmente, nuestro código de inferencia personalizado. Luego pasamos esto Model objetar a un ModelStep, que se agrega a la canalización local. Ver el siguiente código:

from sagemaker.model import Model

model = Model(
    image_uri=image_uri,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    source_dir="code",
    entry_point="inference.py",
    role=role,
    sagemaker_session=local_pipeline_session,
)

from sagemaker.workflow.model_step import ModelStep

step_create_model = ModelStep(name="AbaloneCreateModel", 
    step_args=model.create(instance_type=instance_type)
)

A continuación, creamos un paso de transformación por lotes en el que enviamos un conjunto de vectores de características y realizamos inferencias. Primero tenemos que crear un Transformer objeto y pasar el local_pipeline_session parámetro para ello. Luego creamos un TransformStep, pasando los argumentos requeridos, y agregue esto a la definición de canalización:

from sagemaker.transformer import Transformer

transformer = Transformer(
    model_name=step_create_model.properties.ModelName,
    instance_type=instance_type,
    instance_count=transform_instance_count,
    output_path=f"s3://{default_bucket}/{prefix}/transform",
    sagemaker_session=local_pipeline_session,
)

from sagemaker.workflow.steps import TransformStep

transform_args = transformer.transform(transform_data, content_type="text/csv")

step_transform = TransformStep(name="AbaloneTransform", step_args=transform_args)

Finalmente, queremos agregar una condición de rama al flujo de trabajo para que solo ejecutemos la transformación por lotes si los resultados de la evaluación del modelo cumplen con nuestros criterios. Podemos indicar este condicional agregando un ConditionStep con un tipo de condición particular, como ConditionLessThanOrEqualTo. Luego enumeramos los pasos para las dos ramas, esencialmente definiendo las ramas if/else o true/false de la canalización. Los if_steps proporcionados en el ConditionStep (paso_crear_modelo, paso_transformar) se ejecutan cada vez que la condición se evalúa como True.

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

step_cond = ConditionStep(
    name="AbaloneMSECond",
    conditions=[cond_lte],
    if_steps=[step_create_model, step_transform],
    else_steps=[step_fail],
)

El siguiente diagrama ilustra esta rama condicional y los pasos if/else asociados. Solo se ejecuta una rama, según el resultado del paso de evaluación del modelo en comparación con el paso de condición.

Ahora que tenemos todos nuestros pasos definidos y las instancias de clase subyacentes creadas, podemos combinarlos en una canalización. Proporcionamos algunos parámetros y definimos de manera crucial el orden de operación simplemente enumerando los pasos en el orden deseado. Tenga en cuenta que el TransformStep no se muestra aquí porque es el objetivo del paso condicional y se proporcionó como argumento de paso para el ConditionalStep más temprano.

from sagemaker.workflow.pipeline import Pipeline

pipeline_name = f"LocalModelPipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        mse_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
    sagemaker_session=local_pipeline_session,
)

Para ejecutar la canalización, debe llamar a dos métodos: pipeline.upsert(), que carga la tubería al servicio subyacente, y pipeline.start(), que comienza a ejecutar la canalización. Puede usar varios otros métodos para consultar el estado de ejecución, enumerar los pasos de la canalización y más. Debido a que usamos la sesión de canalización en modo local, todos estos pasos se ejecutan localmente en su procesador. El resultado de la celda debajo del método de inicio muestra el resultado de la canalización:

pipeline.upsert(role_arn=role)
execution = pipeline.start()

Debería ver un mensaje en la parte inferior de la salida de la celda similar al siguiente:

Pipeline execution d8c3e172-089e-4e7a-ad6d-6d76caf987b7 SUCCEEDED

Revertir a recursos administrados

Una vez que hayamos confirmado que la canalización se ejecuta sin errores y que estemos satisfechos con el flujo y la forma de la canalización, podemos volver a crear la canalización pero con los recursos administrados de SageMaker y volver a ejecutarla. El único cambio requerido es usar el PipelineSession objeto en lugar de LocalPipelineSession:

en sagemaker.workflow.pipeline_context importar LocalPipelineSession
from sagemaker.workflow.pipeline_context import PipelineSession

local_pipeline_session = LocalPipelineSession()
pipeline_session = PipelineSession()

Esto informa al servicio que ejecute cada paso que haga referencia a este objeto de sesión en los recursos administrados de SageMaker. Dado el pequeño cambio, ilustramos solo los cambios de código requeridos en la siguiente celda de código, pero el mismo cambio debería implementarse en cada celda usando el local_pipeline_session objeto. Sin embargo, los cambios son idénticos en todas las celdas porque solo reemplazamos el local_pipeline_session objeto con el pipeline_session objeto.

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    role=role,
    sagemaker_session=pipeline_session,  # non-local session
)

Después de que el objeto de la sesión local se haya reemplazado en todas partes, recreamos la canalización y la ejecutamos con los recursos administrados de SageMaker:

from sagemaker.workflow.pipeline import Pipeline

pipeline_name = f"LocalModelPipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        mse_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
    sagemaker_session=pipeline_session, # non-local session
)

pipeline.upsert(role_arn=role)
execution = pipeline.start()

Limpiar

Si desea mantener ordenado el entorno de Studio, puede usar los siguientes métodos para eliminar la canalización y el modelo de SageMaker. El código completo se puede encontrar en la muestra. cuaderno.

# delete models 
sm_client = boto3.client("sagemaker")
model_prefix="AbaloneCreateModel"
delete_models(sm_client, model_prefix)

# delete managed pipeline
pipeline_to_delete = 'SM-Managed-Pipeline'
delete_sagemaker_pipeline(sm_client, pipeline_to_delete)

Conclusión

Hasta hace poco, podía usar la función de modo local de SageMaker Processing y SageMaker Training para iterar en sus scripts de procesamiento y capacitación localmente, antes de ejecutarlos en todos los datos con los recursos administrados de SageMaker. Con la nueva función de modo local de SageMaker Pipelines, los profesionales de ML ahora pueden aplicar el mismo método al iterar en sus canalizaciones de ML, uniendo los diferentes flujos de trabajo de ML. Cuando la canalización está lista para la producción, ejecutarla con los recursos administrados de SageMaker requiere solo unas pocas líneas de cambios en el código. Esto reduce el tiempo de ejecución de la canalización durante el desarrollo, lo que lleva a un desarrollo de la canalización más rápido con ciclos de desarrollo más rápidos, al tiempo que reduce el costo de los recursos administrados de SageMaker.

Para obtener más información, visite Canalizaciones de Amazon SageMaker or Use SageMaker Pipelines para ejecutar sus trabajos localmente.

Sobre los autores

pablo hargis ha centrado sus esfuerzos en el aprendizaje automático en varias empresas, incluidas AWS, Amazon y Hortonworks. Le gusta crear soluciones tecnológicas y enseñar a las personas cómo aprovecharlas al máximo. Antes de su puesto en AWS, fue arquitecto principal de Exportaciones y Expansiones de Amazon, ayudando a amazon.com a mejorar la experiencia de los compradores internacionales. A Paul le gusta ayudar a los clientes a expandir sus iniciativas de aprendizaje automático para resolver problemas del mundo real.

palma de niklas es arquitecto de soluciones en AWS en Estocolmo, Suecia, donde ayuda a los clientes de los países nórdicos a tener éxito en la nube. Es un apasionado de las tecnologías sin servidor junto con el IoT y el aprendizaje automático. Fuera del trabajo, Niklas es un ávido esquiador de fondo y snowboarder, así como un maestro hervidor de huevos.

kirit thadaka es un arquitecto de soluciones de aprendizaje automático que trabaja en el equipo de SageMaker Service SA. Antes de unirse a AWS, Kirit trabajó en empresas emergentes de IA en etapa inicial, seguido de algún tiempo como consultor en varios roles en investigación de IA, MLOps y liderazgo técnico.