Integración de dependencias personalizadas en Amazon Sagemaker Canvas Flows de trabajo

Al implementar aprendizaje automático (Ml) flujos de trabajo en Canvas de Amazon Sagemakerlas organizaciones pueden necesitar considerar dependencias externas necesarias para sus casos de uso específicos. Aunque el lienzo de Sagemaker proporciona potentes capacidades sin código y código bajo para la experimentación rápida, algunos proyectos pueden requerir dependencias y bibliotecas especializadas que no están incluidas por defecto en el lienzo de SageMaker. Esta publicación proporciona un ejemplo de cómo incorporar un código que se basa en dependencias externas en sus flujos de trabajo de lienzo de Sagemaker.

Amazon Sagemaker Canvas es un Código de bajo código (LCNC) Plataforma ML que guía a los usuarios a través de cada etapa del viaje ML, desde la preparación de datos inicial hasta la implementación final del modelo. Sin escribir una sola línea de código, los usuarios pueden explorar conjuntos de datos, transformar datos, crear modelos y generar predicciones.

Sagemaker Canvas ofrece capacidades integrales de disputas de datos que lo ayudan a preparar sus datos, incluidos:

• Más de 300 pasos de transformación incorporados
• Capacidades de ingeniería de características
• Normalización de datos y funciones de limpieza
• Un editor de código personalizado que apoya a Python, Pyspark y SparkSQL

En esta publicación, demostramos cómo incorporar dependencias almacenadas en Servicio de almacenamiento simple de Amazon (Amazon S3) dentro de un Amazon Sagemaker Data Wrangler fluir. Usando este enfoque, puede ejecutar scripts personalizados que dependen de módulos que no sean inherentemente compatibles con Sagemaker Canvas.

Descripción general de la solución

Para mostrar la integración de scripts y dependencias personalizados de Amazon S3 en SageMaker Canvas, exploramos el siguiente flujo de trabajo de ejemplo.

La solución sigue tres pasos principales:

1. Cargue scripts y dependencias personalizados en Amazon S3
2. Use Sagemaker Data Wrangler en Sagemaker Canvas para transformar sus datos utilizando el código cargado
3. Entrenar y exportar el modelo

El siguiente diagrama es la arquitectura de la solución.

En este ejemplo, trabajamos con dos conjuntos de datos complementarios disponibles en el lienzo de Sagemaker que contienen información de envío para entregas de pantalla de la computadora. Al unirnos a estos conjuntos de datos, creamos un conjunto de datos integral que captura varias métricas de envío y resultados de entrega. Nuestro objetivo es construir un modelo predictivo que pueda determinar si los envíos futuros llegarán a tiempo según los patrones y características de envío históricos.

Requisitos previos

Como requisito previo, necesita acceso a Amazon S3 y Amazon Sagemaker AI. Si aún no tiene un dominio de IA Sagemaker configurado en su cuenta, también necesita permisos para Crea un dominio de IA de Sagemaker.

Crear el flujo de datos

Para crear el flujo de datos, siga estos pasos:

1. En la consola AI de Amazon Sagemaker, en el panel de navegación, debajo Aplicaciones e IDESseleccionar Lienzocomo se muestra en la siguiente captura de pantalla. Es posible que deba crear un dominio de Sagemaker si aún no lo ha hecho.
2. Después de que se cree su dominio, elija Lienzo abierto.

3. En lienzo, seleccione el Conjuntos de datos pestaña y seleccione Canvas-sample-shipping-logs.csv, como se muestra en la siguiente captura de pantalla. Después de que aparezca la vista previa, elija + Crear un flujo de datos.

El flujo de datos inicial se abrirá con una fuente y un tipo de datos.

4. En la parte superior derecha de la pantalla y seleccione Agregar datos → Tabular. Elegir Conjuntos de datos de lona como la fuente y seleccione el lienzo-muestra-productos-descripciones.csv.
5. Elegir Próximo Como se muestra en la siguiente captura de pantalla. Luego elija Importar.

6. Después de que se hayan agregado ambos conjuntos de datos, seleccione el signo más. En el menú desplegable, elija Seleccionar Combinar datos. Desde el siguiente menú desplegable, elija Unirse.

7. Para realizar una unión interna en la columna ProductID, en el menú de la derecha, debajo Tipo de uniónelegir Unión interior. Bajo Unirse a las teclaselegir Productidcomo se muestra en la siguiente captura de pantalla.

8. Después de que se hayan unido los conjuntos de datos, seleccione el signo más. En el menú desplegable, seleccione + Agregar transformación. Se abrirá una vista previa del conjunto de datos.

El conjunto de datos contiene columnas XshippingDistance (Long) y YshippingDistance (largas). Para nuestros propósitos, queremos utilizar una función personalizada que encuentre la distancia total utilizando las coordenadas X e Y y luego soltar las columnas de coordenadas individuales. Para este ejemplo, encontramos la distancia total utilizando una función que se basa en la biblioteca MPMath.

9. Para llamar a la función personalizada, seleccione + Agregar transformación. En el menú desplegable, seleccione Transformación personalizada. Cambiar el editor a Python (pandas) e intente ejecutar la siguiente función desde el editor de Python:

from mpmath import sqrt  # Import sqrt from mpmath

def calculate_total_distance(df, x_col="XShippingDistance", y_col="YShippingDistance", new_col="TotalDistance"):

    # Use mpmath's sqrt to calculate the total distance for each row
    df[new_col] = df.apply(lambda row: float(sqrt(row[x_col]**2 + row[y_col]**2)), axis=1)
    
    # Drop the original x and y columns
    df = df.drop(columns=[x_col, y_col])
    
    return df

df = calculate_total_distance(df)

Ejecutar la función produce el siguiente error: ModulenotFoundError: No se llamó el módulo ‘MPMATH’, como se muestra en la siguiente captura de pantalla.

Este error ocurre porque MPMath no es un módulo inherentemente compatible con Sagemaker Canvas. Para usar una función que se basa en este módulo, necesitamos abordar el uso de una función personalizada de manera diferente.

Zip el script y las dependencias

Para usar una función que se basa en un módulo que no se admite de forma nativa en el lienzo, el script personalizado debe estar encendido con los módulos en el que se basa. Para este ejemplo, utilizamos nuestro entorno de desarrollo integrado local (IDE) para crear un script.py que se basa en la biblioteca MPMath.

El archivo script.py contiene dos funciones: una función que es compatible con el tiempo de ejecución de Python (pandas) (función calculate_total_distance), y uno que sea compatible con el tiempo de ejecución de Python (Pyspark) (función udf_total_distance).

def calculate_total_distance(df, x_col="XShippingDistance", y_col="YShippingDistance", new_col="TotalDistance"):
    from npmath import sqrt  # Import sqrt from npmath

    # Use npmath's sqrt to calculate the total distance for each row
    df[new_col] = df.apply(lambda row: float(sqrt(row[x_col]**2 + row[y_col]**2)), axis=1)

    # Drop the original x and y columns
    df = df.drop(columns=[x_col, y_col])

    return df

def udf_total_distance(df, x_col="XShippingDistance", y_col="YShippingDistance", new_col="TotalDistance"):
    from pyspark.sql import SparkSession
    from pyspark.sql.functions import udf
    from pyspark.sql.types import FloatType

    spark = SparkSession.builder \
        .master("local") \
        .appName("DistanceCalculation") \
        .getOrCreate()

    def calculate_distance(x, y):
        import sys

        # Add the path to npmath
        mpmath_path = "/tmp/maths"
        if mpmath_path not in sys.path:
            sys.path.insert(0, mpmath_path)

        from mpmath import sqrt
        return float(sqrt(x**2 + y**2))

    # Register and apply UDF
    distance_udf = udf(calculate_distance, FloatType())
    df = df.withColumn(new_col, distance_udf(df[x_col], df[y_col]))
    df = df.drop(x_col, y_col)

    return df

Para asegurarse de que el script pueda ejecutarse, instale mpmath en el mismo directorio que script.py ejecutando pip install mpmath.

Correr zip -r my_project.zip Para crear un archivo .zip que contenga la función y la instalación de MPMath. El directorio actual ahora contiene un archivo .zip, nuestro script de Python y la instalación de la que depende nuestro script, como se muestra en la siguiente captura de pantalla.

Subir a Amazon S3

Después de crear el archivo .zip, cárguelo a un cubo de Amazon S3.

Después de que el archivo zip se ha subido a Amazon S3, se puede acceder en el lienzo de Sagemaker.

Ejecutar el script personalizado

Regrese al flujo de datos en el lienzo de Sagemaker y reemplace el código de función personalizado anterior con el siguiente código y elija Actualizar.

import zipfile
import boto3
import sys
from pathlib import Path
import shutil
import importlib.util


def load_script_and_dependencies(bucket_name, zip_key, extract_to):
    """
    Downloads a zip file from S3, unzips it, and ensures dependencies are available.

    Args:
        bucket_name (str): Name of the S3 bucket.
        zip_key (str): Key for the .zip file in the bucket.
        extract_to (str): Directory to extract files to.

    Returns:
        str: Path to the extracted folder containing the script and dependencies.
    """
    
    s3_client = boto3.client("s3")
    
    # Local path for the zip file
    zip_local_path="/tmp/dependencies.zip"
    
    # Download the .zip file from S3
    s3_client.download_file(bucket_name, zip_key, zip_local_path)
    print(f"Downloaded zip file from S3: {zip_key}")

    # Unzip the file
    try:
        with zipfile.ZipFile(zip_local_path, 'r') as zip_ref:
            zip_ref.extractall(extract_to)
            print(f"Extracted files to {extract_to}")
    except Exception as e:
        raise RuntimeError(f"Failed to extract zip file: {e}")

    # Add the extracted folder to Python path
    if extract_to not in sys.path:
      sys.path.insert(0, extract_to)
          
    return extract_to
    


def call_function_from_script(script_path, function_name, df):
    """
    Dynamically loads a function from a Python script using importlib.
    """
    try:
        # Get the script name from the path
        module_name = script_path.split('/')[-1].replace('.py', '')
        
        # Load the module specification
        spec = importlib.util.spec_from_file_location(module_name, script_path)
        if spec is None:
            raise ImportError(f"Could not load specification for module {module_name}")
            
        # Create the module
        module = importlib.util.module_from_spec(spec)
        sys.modules[module_name] = module
        
        # Execute the module
        spec.loader.exec_module(module)
        
        # Get the function from the module
        if not hasattr(module, function_name):
            raise AttributeError(f"Function '{function_name}' not found in the script.")
            
        loaded_function = getattr(module, function_name)

        # Clean up: remove module from sys.modules after execution
        del sys.modules[module_name]
        
        # Call the function
        return loaded_function(df)
        
    except Exception as e:
        raise RuntimeError(f"Error loading or executing function: {e}")


bucket_name="canvasdatabuckett"  # S3 bucket name
zip_key = 'functions/my_project.zip'  # S3 path to the zip file with our custom dependancy
script_name="script.py"  # Name of the script in the zip file
function_name="udf" # Name of function to call from our script
extract_to = '/tmp/maths' # Local path to our custom script and dependancies

# Step 1: Load the script and dependencies
extracted_path = load_script_and_dependencies(bucket_name, zip_key, extract_to)

# Step 2: Call the function from the script
script_path = f"{extracted_path}/{script_name}"
df = call_function_from_script(script_path, function_name, df)

Este código de ejemplo descomprime el archivo .zip y agrega las dependencias requeridas a la ruta local para que estén disponibles para la función en el momento de ejecución. Debido a que MPMath se agregó a la ruta local, ahora puede llamar a una función que se basa en esta biblioteca externa.

El código anterior se ejecuta usando el tiempo de ejecución de Python (Pandas) y la función calculación_total_distance. Para usar el tiempo de ejecución de Python (Pyspark), actualice la variable function_name para llamar a la función UDF_TOTAL_DISTANCE en su lugar.

Completa el flujo de datos

Como último paso, elimine las columnas irrelevantes antes de entrenar el modelo. Sigue estos pasos:

1. En la consola de lienzo de Sagemaker, seleccione + Agregar transformación. En el menú desplegable, seleccione Administrar columnas
2. Bajo Transformarelegir Columna de caída. Bajo Columnas para soltaragregar productID_0, ProductID_1, y OrderId, como se muestra en la siguiente captura de pantalla.

El conjunto de datos final debe contener 13 columnas. El flujo de datos completo se muestra en la siguiente imagen.

Entrenar el modelo

Para entrenar el modelo, siga estos pasos:

1. En la parte superior derecha de la página, seleccione Crear modelo y nombra tu conjunto de datos y modelo.
2. Seleccionar Análisis predictivo como el tipo de problema y Ontimedelivery Como la columna objetivo, como se muestra en la captura de pantalla a continuación.

Al construir el modelo, puede elegir ejecutar una compilación rápida o una compilación estándar. Una construcción rápida prioriza la velocidad sobre la precisión y produce un modelo entrenado en menos de 20 minutos. Una construcción estándar prioriza la precisión sobre la latencia, pero el modelo tarda más en entrenar.

Resultados

Después de completar la construcción del modelo, puede ver la precisión del modelo, junto con métricas como F1, precisión y retiro. En el caso de una construcción estándar, el modelo alcanzó una precisión del 94.5%.

Después de completar el entrenamiento del modelo, hay cuatro formas en que puede usar su modelo:

1. Implementar el modelo directamente Desde el lienzo de Sagemaker hasta un punto final
2. Agregue el modelo al Registro del Modelo de Sagemaker
3. Exportar su modelo a un cuaderno Jupyter
4. Envía tu modelo a Amazon Quicksight Para usar en visualizaciones del tablero

Limpiar

Para gestionar los costos y evitar adicionales cargos de espacio de trabajoelegir Finalizar la sesión Para firmar el lienzo de Sagemaker cuando haya terminado de usar la aplicación, como se muestra en la siguiente captura de pantalla. También puede configurar el lienzo de Sagemaker para Apagado automáticamente cuando está inactivo.

Si creó un cubo S3 específicamente para este ejemplo, también es posible que desee vacío y borrar tu cubo.

Resumen

En esta publicación, demostramos cómo puede cargar dependencias personalizadas a Amazon S3 e integrarlas en los flujos de trabajo de Canvas SageMaker. Al caminar a través de un ejemplo práctico de implementar una función de cálculo de distancia personalizada con la biblioteca MPMath, mostramos cómo:

1. Paquete de código personalizado y dependencias en un archivo .zip
2. Almacene y acceda a estas dependencias desde Amazon S3
3. Implementar transformaciones de datos personalizadas en Sagemaker Data Wrangler
4. Entrenar un modelo predictivo utilizando los datos transformados

Este enfoque significa que los científicos y analistas de datos pueden extender las capacidades de lienzo de Sagemaker más allá de las más de 300 funciones incluidas.

Para probar las transformaciones personalizadas, consulte el Documentación de lienzo de Amazon Sagemaker e inicie sesión en Sagemaker Canvas hoy. Para obtener información adicional sobre cómo puede optimizar su implementación de lienzo de Sagemaker, recomendamos explorar estas publicaciones relacionadas:

Sobre el autor

Nadhya Polanco es un arquitecto de soluciones asociadas en AWS con sede en Bruselas, Bélgica. En este rol, apoya a las organizaciones que buscan incorporar la IA y el aprendizaje automático en sus cargas de trabajo. En su tiempo libre, a Nadhya le gusta disfrutar de su pasión por el café y explorar nuevos destinos.