Invoca una función remota

Para invocar una función dentro del decorador @remote, utilice cualquiera de los métodos siguientes:

Si utiliza el método decorador @remote para invocar una función, el trabajo de entrenamiento esperará a que la función se complete antes de iniciar una nueva tarea. Sin embargo, si usa el RemoteExecutorAPI, puede ejecutar más de un trabajo en paralelo. En las siguientes secciones se muestran las dos formas de invocar una función.

Use un decorador @remote para invocar una función

Puedes usar el decorador @remote para anotar una función. SageMaker transformará el código del decorador en un SageMaker trabajo de formación. A continuación, el trabajo de entrenamiento invocará la función dentro del decorador y esperará a que se complete el trabajo. El siguiente ejemplo de código muestra cómo importar las bibliotecas necesarias, iniciar una SageMaker instancia y anotar una multiplicación de matrices con el decorador @remote.

from sagemaker.remote_function import remote
import numpy as np

@remote(instance_type="ml.m5.large")
def matrix_multiply(a, b):
    return np.matmul(a, b)
    
a = np.array([[1, 0],
             [0, 1]])
b = np.array([1, 2])

assert (matrix_multiply(a, b) == np.array([1,2])).all()

El decorador se define de la siguiente manera.

def remote(
    *,
    **kwarg):
        ...

Cuando se invoca una función decorada, SageMaker Python SDK carga cualquier excepción provocada por un error en la memoria local. En el siguiente ejemplo de código, la primera llamada a la función de división se completa correctamente y el resultado se carga en la memoria local. En la segunda llamada a la función de división, el código devuelve un error y este error se carga en la memoria local.

from sagemaker.remote_function import remote
import pytest

@remote()
def divide(a, b):
    return a/b

# the underlying job is completed successfully 
# and the function return is loaded
assert divide(10, 5) == 2

# the underlying job fails with "AlgorithmError" 
# and the function exception is loaded into local memory 
with pytest.raises(ZeroDivisionError):
    divide(10, 0)

Cómo cambiar el valor de una variable local

La función decoradora se ejecuta en una máquina remota. Cambiar una variable no local o los argumentos de entrada dentro de una función decorada no cambiará el valor local.

En el siguiente ejemplo de código, se añaden una lista y un diccionario dentro de la función de decorador. Esto no cambia cuando se invoca la función de decorador.

a = []

@remote
def func():
    a.append(1)

# when func is invoked, a in the local memory is not modified        
func() 
func()

# a stays as []
    
a = {}
@remote
def func(a):
    # append new values to the input dictionary
    a["key-2"] = "value-2"
    
a = {"key": "value"}
func(a)

# a stays as {"key": "value"}

Para cambiar el valor de una variable local declarada dentro de una función de decorador, devuelva la variable de la función. El siguiente ejemplo de código muestra que el valor de una variable local cambia cuando la función la devuelve.

a = {"key-1": "value-1"}

@remote
def func(a):
    a["key-2"] = "value-2"
    return a

a = func(a)

-> {"key-1": "value-1", "key-2": "value-2"}                                    

Serialización y deserialización de datos

Al invocar una función remota, serializa SageMaker automáticamente los argumentos de la función durante las etapas de entrada y salida. Los argumentos y los retornos de la función se serializan con cloudpickle. SageMaker admite la serialización de los siguientes objetos y funciones de Python.

Se puede utilizar lo siguiente con algunas limitaciones.

La siguiente sección contiene las prácticas recomendadas para usar las clases de Python anteriores con algunas limitaciones en la función remota, información sobre dónde se SageMaker almacenan los datos serializados y cómo administrar el acceso a ellos.

Prácticas recomendadas para las clases de Python con asistencia limitada para la serialización remota de datos

Puede usar las clases de Python enumeradas en esta sección con limitaciones. En las siguientes secciones, se analizan las prácticas recomendadas para utilizar las siguientes clases de Python.

#Do not pass a Dask DataFrame  to your remote function as follows
def clean(df: dask.DataFrame ):
    cleaned = df[] \ ...                            

import dask.dataframe as dd

@remote(
   instance_type='ml.m5.24xlarge',
   volume_size=300, 
   keep_alive_period_in_seconds=600)
#pass the data path to your remote function rather than the Dask DataFrame  itself
def clean(raw_data_path: str, output_data_path: str: split_ratio: list[float]):
    df = dd.read_parquet(raw_data_path) #pass the path to your DataFrame 
    cleaned = df[(df.column_a >= 1) & (df.column_a < 5)]\
        .drop(['column_b', 'column_c'], axis=1)\
        .persist() #keep the data in memory to facilitate the following random_split operation

    train_df, test_df = cleaned.random_split(split_ratio, random_state=10)

    train_df.to_parquet(os.path.join(output_data_path, 'train')
    test_df.to_parquet(os.path.join(output_data_path, 'test'))
    
clean("s3://amzn-s3-demo-bucket/raw/", "s3://amzn-s3-demo-bucket/cleaned/", split_ratio=[0.7, 0.3])                            

executor = RemoteExecutor(
    instance_type='ml.m5.24xlarge',
    volume_size=300, 
    keep_alive_period_in_seconds=600)

future = executor.submit(lambda: dd.read_parquet("s3://amzn-s3-demo-bucket/raw/").describe().compute())

future.result()                            

import xgboost as xgb

@remote
def train(df, params):
    #Convert a pandas dataframe into a DMatrix DataFrame and use it for training
    dtrain = DMatrix(df) 
    return xgb.train(dtrain, params)                        

import tensorflow as tf
import tensorflow_io as tfio

@remote
def train(data_path: str, params):
    
    dataset = tf.data.TextLineDataset(tf.data.Dataset.list_files(f"{data_path}/*.txt"))
    ...
    
train("s3://amzn-s3-demo-bucket/data", {})                        

# Do not return a model trained on GPUs to a CPU-only environment as follows

@remote(instance_type='ml.g4dn.xlarge')
def train(...):
    if torch.cuda.is_available():
        device = torch.device("cuda")
    else:
        device = torch.device("cpu") # a device without GPU capabilities
    
    model = Net().to(device)
    
    # train the model
    ...
    
    return model
    
model = train(...) #returns a DeserializationError if run on a device with GPU                        

import s3fs

model_path = "s3://amzn-s3-demo-bucket/folder/"

@remote(instance_type='ml.g4dn.xlarge')
def train(...):
    if torch.cuda.is_available():
        device = torch.device("cuda")
    else:
        device = torch.device("cpu")
    
    model = Net().to(device)
    
    # train the model
    ...
    
    fs = s3fs.FileSystem()
    with fs.open(os.path.join(model_path, 'model.pt'), 'wb') as file:
        torch.save(model.state_dict(), file) #this writes the model in a device-agnostic way (CPU vs GPU)
    
train(...) #use the model to train on either CPUs or GPUs

model = Net()
fs = s3fs.FileSystem()with fs.open(os.path.join(model_path, 'model.pt'), 'rb') as file:
    model.load_state_dict(torch.load(file, map_location=torch.device('cpu')))

¿Dónde se SageMaker almacenan los datos serializados

Al invocar una función remota, serializa SageMaker automáticamente los argumentos de la función y los valores devueltos durante las etapas de entrada y salida. Estos datos serializados se almacenan en un directorio raíz en el bucket de S3. El directorio raíz, <s3_root_uri>, se especifica en un archivo de configuración. El parámetro job_name se genera automáticamente para usted.

En el directorio raíz, SageMaker crea una <job_name> carpeta que contiene el directorio de trabajo actual, la función serializada, los argumentos de la función serializada, los resultados y cualquier excepción que haya surgido al invocar la función serializada.

Bajo <job_name>, el directorio workdir contiene un archivo comprimido de su directorio de trabajo actual. El archivo comprimido incluye todos los archivos de Python del directorio de trabajo y el archivo requirements.txt, que especifica las dependencias necesarias para ejecutar la función remota.

A continuación, se muestra un ejemplo de la estructura de carpetas de un bucket de S3 que se especifica en el archivo de configuración.

<s3_root_uri>/ # specified by s3_root_uri or S3RootUri
    <job_name>/ #automatically generated for you
        workdir/workspace.zip # archive of the current working directory (workdir)
        function/ # serialized function
        arguments/ # serialized function arguments
        results/ # returned output from the serialized function including the model
        exception/ # any exceptions from invoking the serialized function

El directorio raíz que especifique en su bucket de S3 no está diseñado para el almacenamiento a largo plazo. Los datos serializados están estrechamente relacionados con la versión de Python y la versión del marco de machine learning (ML) que se utilizaron durante la serialización. Si actualiza la versión de Python o el marco ML, es posible que no pueda usar sus datos serializados. En su lugar, haga lo siguiente.

Para obtener más información sobre qué incluir en un archivo de configuración para una función remota, consulte Archivo de configuración.

Acceder a los datos serializados

Los administradores pueden proporcionar la configuración de los datos serializados, incluida su ubicación y cualquier configuración de cifrado en un archivo de configuración. De forma predeterminada, los datos serializados se cifran con una clave AWS Key Management Service ()AWS KMS. Los administradores también pueden restringir el acceso al directorio raíz que especifique en el archivo de configuración mediante una política de bucket. El archivo de configuración se puede compartir y usar en todos los proyectos y trabajos. Para obtener más información sobre los archivos de configuración, consulte Archivo de configuración.

Utilice la RemoteExecutor API para invocar una función

Puede usar el RemoteExecutor API para invocar una función. SageMaker Python SDK transformará el código de la RemoteExecutor llamada en un trabajo de SageMaker formación. A continuación, el trabajo de entrenamiento invocará la función como una operación asíncrona y devolverá un future. Si usa el RemoteExecutorAPI, puede ejecutar más de un trabajo de entrenamiento en paralelo. Para obtener más información sobre los futuros en Python, consulte Futures.

El siguiente ejemplo de código muestra cómo importar las bibliotecas necesarias, definir una función, iniciar una SageMaker instancia y utilizarlas API para enviar una solicitud para ejecutar 2 trabajos en paralelo.

from sagemaker.remote_function import RemoteExecutor

def matrix_multiply(a, b):
    return np.matmul(a, b)


a = np.array([[1, 0],
             [0, 1]])
b = np.array([1, 2])

with RemoteExecutor(max_parallel_job=2, instance_type="ml.m5.large") as e:
    future = e.submit(matrix_multiply, a, b)

assert (future.result() == np.array([1,2])).all()

La clase RemoteExecutor es una implementación de la biblioteca concurrent.futures.Executor.

En el ejemplo de código siguiente se muestra cómo definir una función y cómo llamarla utilizando la RemoteExecutorAPI. En este ejemplo, el RemoteExecutor enviará 4 trabajos en total, pero solo 2 en paralelo. Los dos últimos trabajos reutilizarán los clústeres con una sobrecarga mínima.

from sagemaker.remote_function.client import RemoteExecutor

def divide(a, b):
    return a/b 

with RemoteExecutor(max_parallel_job=2, keep_alive_period_in_seconds=60) as e:
    futures = [e.submit(divide, a, 2) for a in [3, 5, 7, 9]]

for future in futures:
    print(future.result())

El parámetro max_parallel_job solo sirve como mecanismo de limitación de velocidad sin optimizar la asignación de los recursos de computación. En el ejemplo de código anterior, RemoteExecutor no reserva recursos de computación para los dos trabajos paralelos antes de que se envíe ningún trabajo. Para obtener más información sobre max_parallel_job u otros parámetros del decorador @remote, consulte Especificación de métodos y clases de funciones remotas.

Clase futura para RemoteExecutor API

Una clase future es una clase pública que representa la función de retorno del trabajo de entrenamiento cuando se invoca de forma asíncrona. La clase future implementa la clase Concurrent.futures.Future. Esta clase se puede usar para realizar operaciones en el trabajo subyacente y cargar datos en la memoria.