Aufrufen einer Remote-Funktion

Verwenden Sie eine der folgenden Methoden, um im @remote Decorator eine Funktion aufzurufen:

Wenn Sie eine Funktion nach der Methode des @remote Decorator verwenden aufrufen, wartet der Trainingsauftrag, bis die Funktion abgeschlossen ist, bevor eine neue Aufgabe begonnen wird. Wenn Sie dagegen die RemoteExecutor API verwenden, können Sie mehr als einen Auftrag parallel ausführen. In den folgenden Abschnitten werden beide Möglichkeiten zum Aufrufen einer Funktion gezeigt.

Eine Funktion mit Hilfe eines @remote Decorators aufrufen

Sie können den @remote -Decorator verwenden, um eine Funktion mit Anmerkungen zu versehen. SageMaker Die KI wird den Code im Dekorateur in einen SageMaker Trainingsjob umwandeln. Der Trainingsauftrag ruft dann die Funktion im Decorator auf und wartet, bis der Auftrag abgeschlossen ist. Das folgende Codebeispiel zeigt, wie Sie die erforderlichen Bibliotheken importieren, eine SageMaker AI-Instanz starten und eine Matrixmultiplikation mit dem @remote -Dekorator annotieren.

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

Der Decorator ist wie folgt definiert.

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

Wenn Sie eine dekorierte Funktion aufrufen, lädt das SageMaker Python SDK alle durch einen Fehler ausgelösten Ausnahmen in den lokalen Speicher. Im folgenden Beispielcode wird der erste Aufruf der Funktion „Teilen“ erfolgreich abgeschlossen und das Ergebnis in den lokalen Speicher geladen. Beim zweiten Aufruf der Funktion „Teilen“ gibt der Code einen Fehler zurück, der in den lokalen Speicher geladen wird.

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)

So ändern Sie den Wert einer lokalen Variablen

Die Decorator-Funktion wird auf einem Remote-Computer ausgeführt. Wenn eine nicht lokale Variable oder Eingabeargumente innerhalb einer dekorierten Funktion geändert werden, ändert sich der lokale Wert nicht.

Im folgenden Beispielcode werden an die Decorator-Funktion eine Liste und ein Dict angehängt. Dies ändert sich nicht, wenn die Decorator-Funktion aufgerufen wird.

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

Um den Wert einer lokalen Variablen zu ändern, die innerhalb einer Decorator-Funktion deklariert wurde, geben Sie die Variable aus der Funktion zurück. Der folgende Beispielcode zeigt, dass der Wert einer lokalen Variablen geändert wird, wenn sie von der Funktion zurückgegeben wird.

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

Serialisierung und Deserialisierung von Daten

Wenn Sie eine Remote-Funktion aufrufen, serialisiert SageMaker AI Ihre Funktionsargumente automatisch während der Eingabe- und Ausgabeschritte. Funktionsargumente und Rückgaben werden mit Cloudpickle serialisiert. https://github.com/cloudpipe/cloudpickle SageMaker AI unterstützt die Serialisierung der folgenden Python-Objekte und -Funktionen.

Die folgenden können mit Einschränkungen verwendet werden.

Der folgende Abschnitt enthält bewährte Methoden für die Verwendung der vorherigen Python-Klassen mit einigen Einschränkungen in Ihrer Remote-Funktion sowie Informationen darüber, wo SageMaker AI Ihre serialisierten Daten speichert und wie Sie den Zugriff darauf verwalten.

Bewährte Methoden für Python-Klassen mit eingeschränkter Unterstützung für die Serialisierung von Remote-Daten

Die in diesem Abschnitt aufgeführten Python-Klassen können Sie mit Einschränkungen verwenden. In den nächsten Abschnitten werden bewährte Methoden für die Verwendung der folgenden Python-Klassen erörtert.

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

Wo SageMaker KI Ihre serialisierten Daten speichert

Wenn Sie eine Remote-Funktion aufrufen, serialisiert SageMaker KI automatisch Ihre Funktionsargumente und Rückgabewerte während der Eingabe- und Ausgabeschritte. Diese serialisierten Daten werden in einem Stammverzeichnis in Ihrem S3-Bucket gespeichert. Das Stammverzeichnis,<s3_root_uri>, geben Sie in einer Konfigurationsdatei an. Der Parameter job_name wird automatisch für Sie generiert.

Unter dem Stammverzeichnis erstellt SageMaker AI einen <job_name> Ordner, der Ihr aktuelles Arbeitsverzeichnis, die serialisierte Funktion, die Argumente für Ihre serialisierte Funktion, Ergebnisse und alle Ausnahmen enthält, die beim Aufrufen der serialisierten Funktion aufgetreten sind.

Das Verzeichnis <job_name> enthält unter workdir ein ZIP-Archiv Ihres aktuellen Arbeitsverzeichnisses. Das komprimierte Archiv enthält ggf. Python-Dateien in Ihrem Arbeitsverzeichnis sowie die requirements.txt Datei, in der alle Abhängigkeiten angegeben sind, die für die Ausführung Ihrer Remote-Funktion erforderlich sind.

Im Folgenden finden Sie ein Beispiel für die Ordnerstruktur unter einem S3-Bucket, den Sie in Ihrer Konfigurationsdatei angeben.

<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

Das Stammverzeichnis, das Sie in Ihrem S3-Bucket angeben, ist nicht zur langfristigen Speicherung vorgesehen. Die serialisierten Daten sind eng mit der Python-Version und der Framework-Version für Machine Learning (ML) verknüpft, die während der Serialisierung verwendet wurden. Wenn Sie die Python-Version oder das ML-Framework aktualisieren, können Sie Ihre serialisierten Daten möglicherweise nicht verwenden. Tun Sie stattdessen folgendes:

Weitere Informationen dazu, was in einer Konfigurationsdatei für eine Remote-Funktion enthalten sein muss, finden Sie unter Konfigurationsdatei.

Zugriff auf Ihre serialisierten Daten

Standardmäßig speichert SageMaker AI serialisierte Daten im Standard-AI-Bucket, der möglicherweise von SageMaker mehreren Projekten gemeinsam genutzt wird. Um unbefugte Änderungen zu verhindern, verwenden Sie den s3_root_uri Parameter des @remote Dekorators oder. RemoteExecutor Geben Sie einen dedizierten S3-Bucket oder ein Präfix für Ihre serialisierten Funktionsargumente, Ergebnisse und Codearchive an.

Das folgende Beispiel zeigt, wie Sie mit dem @remote Decorator einen dedizierten S3-Speicherort angeben:

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

Standardmäßig verschlüsselt AWS Key Management Service (AWS KMS) die serialisierten Daten. Sie können den Speicherort und die Verschlüsselungseinstellungen in einer Konfigurationsdatei konfigurieren. Sie können die Konfigurationsdatei projekt- und aufgabenübergreifend gemeinsam nutzen. Weitere Informationen zur Konfigurationsdatei finden Sie unter Konfigurationsdatei.

Verwenden Sie die RemoteExecutor API, um eine Funktion aufzurufen

Sie können die RemoteExecutor API verwenden, um eine Funktion aufzurufen. SageMaker Das AI Python SDK wandelt den Code im RemoteExecutor Aufruf in einen SageMaker KI-Trainingsjob um. Der Trainingsauftrag ruft dann die Funktion als asynchronen Vorgang auf und gibt ein Future zurück. Wenn Sie die API RemoteExecutor verwenden, können Sie mehr als einen Trainingsauftrag parallel ausführen. Weitere Informationen zu Futures in Python finden Sie unter Futures.

Das folgende Codebeispiel zeigt, wie Sie die erforderlichen Bibliotheken importieren, eine Funktion definieren, eine SageMaker AI-Instanz starten und die API verwenden, um eine Anfrage zur parallelen Ausführung von 2 Jobs zu senden.

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

Die Klasse RemoteExecutor ist eine Implementierung der Bibliothek Concurrent.Futures.Executor.

Der folgende Beispielcode zeigt, wie eine Funktion definiert und mit der RemoteExecutorAPI aufgerufen wird. In diesem Beispiel reichen die RemoteExecutor insgesamt 4 Jobs ein, aber nur 2 parallel. Bei den letzten beiden Jobs werden die Cluster mit minimalem Aufwand wiederverwendet.

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

Der Parameter max_parallel_job dient lediglich als Mechanismus zur Ratenbegrenzung, ohne die Zuweisung von Rechenressourcen zu optimieren. Im vorangegangenen Beispielcode reserviert RemoteExecutor keine Rechenressourcen für die beiden parallelen Aufträge, bevor Aufträge eingereicht werden. Weitere Informationen zu max_parallel_job oder sonstigen Parametern für den @remote Decorator finden Sie unter Angabe von Klassen und Methoden für Remote-Funktionen.

Future-Klasse für die API RemoteExecutor

Eine Future-Klasse ist eine öffentliche Klasse, die die Rückgabefunktion des Trainingsauftrags darstellt, wenn er asynchron aufgerufen wird. Die Future-Klasse implementiert die Klasse Concurrent.Futures.Future. Diese Klasse kann für Operationen am zugrundeliegenden Auftrag verwendet werden und dafür, Daten in den Speicher zu laden.