Invoquer une fonction à distance

Pour invoquer une fonction dans le décorateur @remote, utilisez l’une des méthodes suivantes :

Si vous utilisez la méthode @remote decorator pour invoquer une fonction, la tâche d’entraînement attendra que la fonction soit terminée avant de démarrer une nouvelle tâche. Toutefois, si vous utilisez l’API RemoteExecutor, vous pouvez exécuter plusieurs tâches en parallèle. Les sections suivantes montrent les deux manières d’invoquer une fonction.

Utilisation d’un décorateur @remote pour invoquer une fonction

Vous pouvez utiliser le décorateur @remote pour annoter une fonction. SageMaker L'IA transformera le code du décorateur en une tâche de SageMaker formation. La tâche d’entraînement invoquera ensuite la fonction dans le décorateur et attendra la fin de la tâche. L'exemple de code suivant montre comment importer les bibliothèques requises, démarrer une instance SageMaker AI et annoter une multiplication matricielle avec le décorateur @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()

Le décorateur est défini comme suit.

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

Lorsque vous invoquez une fonction décorée, le SDK SageMaker Python charge toutes les exceptions déclenchées par une erreur dans la mémoire locale. Dans l’exemple de code suivant, le premier appel à la fonction de division se termine correctement et le résultat est chargé dans la mémoire locale. Lors du deuxième appel à la fonction de division, le code renvoie une erreur et cette erreur est chargée dans la mémoire locale.

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)

Comment modifier la valeur d’une variable locale

La fonction de décorateur est exécutée sur une machine distante. La modification d’une variable non locale ou d’arguments d’entrée dans une fonction décorée ne modifiera pas la valeur locale.

Dans l’exemple de code suivant, une liste et un dictionnaire sont ajoutés dans la fonction de décoration. Cela ne change pas lorsque la fonction de décorateur est invoquée.

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"}

Pour modifier la valeur d’une variable locale déclarée dans une fonction de décoration, renvoyez la variable depuis la fonction. L’exemple de code suivant montre que la valeur d’une variable locale est modifiée lorsqu’elle est renvoyée par la fonction.

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"}                                    

Sérialisation et désérialisation des données

Lorsque vous invoquez une fonction distante, l' SageMaker IA sérialise automatiquement les arguments de votre fonction pendant les étapes d'entrée et de sortie. Les arguments et les retours des fonctions sont sérialisés à l'aide de https://github.com/cloudpipe/cloudpickle cloudpickle. SageMaker L'IA prend en charge la sérialisation des objets et fonctions Python suivants.

Les éléments suivants peuvent être utilisés avec certaines restrictions.

La section suivante contient les meilleures pratiques pour utiliser les classes Python précédentes avec certaines limitations de votre fonction à distance, des informations sur l'endroit où SageMaker AI stocke vos données sérialisées et comment gérer l'accès à celles-ci.

Bonnes pratiques pour les classes Python avec une prise en charge limitée de la sérialisation de données distantes

Vous pouvez utiliser les classes Python répertoriées dans cette section avec certaines restrictions. Les sections suivantes présentent les bonnes pratiques relatives à l’utilisation des classes Python suivantes.

#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')))

Où SageMaker l'IA stocke vos données sérialisées

Lorsque vous invoquez une fonction distante, l' SageMaker IA sérialise automatiquement les arguments de votre fonction et renvoie des valeurs pendant les étapes d'entrée et de sortie. Ces données sérialisées sont stockées dans un répertoire racine de votre compartiment S3. Vous spécifiez le répertoire racine, <s3_root_uri>, dans un fichier de configuration. Le paramètre job_name est automatiquement généré pour vous.

Dans le répertoire racine, SageMaker AI crée un <job_name> dossier contenant votre répertoire de travail actuel, votre fonction sérialisée, les arguments de votre fonction sérialisée, les résultats et toutes les exceptions résultant de l'invocation de la fonction sérialisée.

Sous <job_name>, le répertoire workdir contient une archive compressée de votre répertoire de travail actuel. L’archive compressée inclut tous les fichiers Python de votre répertoire de travail ainsi que le fichier requirements.txt, qui spécifie les dépendances nécessaires pour exécuter votre fonction distante.

Voici un exemple de structure de dossiers sous un compartiment S3 que vous spécifiez dans votre fichier de configuration.

<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

Le répertoire racine que vous spécifiez dans votre compartiment S3 n’est pas destiné au stockage à long terme. Les données sérialisées sont étroitement liées à la version Python et à la version du framework de machine learning (ML) utilisées lors de la sérialisation. Si vous mettez à niveau la version Python ou le framework de machine learning, vous ne pourrez peut-être pas utiliser vos données sérialisées. Procédez plutôt comme suit.

Pour plus d’informations sur les éléments à inclure dans un fichier de configuration pour une fonction distante, consultez Fichier de configuration.

Accès à vos données sérialisées

Par défaut, SageMaker AI stocke les données sérialisées dans le compartiment SageMaker AI par défaut, qui peut être partagé entre les projets. Pour empêcher toute modification non autorisée, utilisez le s3_root_uri paramètre du @remote décorateur ouRemoteExecutor. Spécifiez un compartiment ou un préfixe S3 dédié pour les arguments, les résultats et les archives de code de vos fonctions sérialisées.

L'exemple suivant montre comment spécifier un emplacement S3 dédié avec le @remote décorateur :

from sagemaker.remote_function import remote

@remote(
    instance_type="ml.m5.xlarge",
    s3_root_uri="s3://amzn-s3-demo-bucket/sagemaker-jobs"
)
def my_training_function(data):
    # Your training logic here
    return model

Par défaut, AWS Key Management Service (AWS KMS) chiffre les données sérialisées. Vous pouvez configurer l'emplacement de stockage et les paramètres de chiffrement dans un fichier de configuration. Vous pouvez partager le fichier de configuration entre les projets et les tâches. Pour plus d'informations sur le fichier de configuration, voir Fichier de configuration.

Utilisation de l’API RemoteExecutor pour invoquer une fonction

Vous pouvez utiliser l'RemoteExecutorAPI pour invoquer une fonction. SageMaker Le SDK AI Python transformera le code contenu dans l'RemoteExecutorappel en une tâche de formation à l' SageMaker IA. La tâche d’entraînement invoquera ensuite la fonction en tant qu’opération asynchrone et renverra un objet Future. Si vous utilisez l’API RemoteExecutor, vous pouvez exécuter plusieurs tâches d’entraînement en parallèle. Pour plus d’informations sur les objets Future dans Python, consultez Objets Future.

L'exemple de code suivant montre comment importer les bibliothèques requises, définir une fonction, démarrer une instance d' SageMaker IA et utiliser l'API pour soumettre une demande d'exécution de 2 tâches en parallèle.

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 classe RemoteExecutor est une implémentation de la bibliothèque concurrent.futures.Executor.

L’exemple de code suivant montre comment définir et appeler une fonction avec RemoteExecutorAPI. Dans cet exemple, RemoteExecutor soumettra 4 tâches au total, mais uniquement 2 en parallèle. Les deux dernières tâches réutiliseront les clusters avec une surcharge minimale.

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())

Le paramètre max_parallel_job sert uniquement de mécanisme de limitation du débit sans optimiser l’allocation des ressources de calcul. Dans l’exemple de code précédent, RemoteExecutor ne réserve pas de ressources de calcul pour les deux tâches parallèles avant que les tâches ne soient soumises. Pour plus d’informations sur max_parallel_job ou sur d’autres paramètres du décorateur @remote, consultez Spécification des classes et méthodes de fonctions distantes.

Classe Future pour l’API RemoteExecutor

Une classe Future est une classe publique qui représente la fonction de retour de la tâche d’entraînement lorsqu’elle est invoquée de manière asynchrone. La classe Future implémente la classe concurrent.futures.Future. Cette classe peut être utilisée pour effectuer des opérations sur la tâche sous-jacente et charger des données en mémoire.