Aufrufen einer -Funktion

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

• Eine Funktion mit Hilfe eines @remote Decorators aufrufen.

• Verwenden Sie die RemoteExecutorAPI, um 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 jedoch den verwenden RemoteExecutorAPI, können Sie mehr als einen Job 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 -Dekorator verwenden, um eine Funktion mit Anmerkungen zu versehen. SageMaker wandelt den Code im Decorator in einen SageMaker Trainingsjob um. 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 Instanz starten und eine Matrixmultiplikation mit Anmerkungen 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, SDK lädt SageMaker Python alle Ausnahmen, die durch einen Fehler ausgelöst wurden, 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)

Anmerkung

Die dekorierte Funktion wird als Remote-Job ausgeführt. Wenn der Thread unterbrochen wird, wird der zugrundeliegende Auftrag nicht abgebrochen.

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 Ihre Funktionsargumente während der Eingabe- und Ausgabephase SageMaker automatisch. Funktionsargumente und Rückgaben werden mit Cloudpickle serialisiert. SageMaker unterstützt die Serialisierung der folgenden Python-Objekte und -Funktionen.

• Integrierte Python-Objekte wie Dicts, Listen, Floats, Ints, Strings, boolesche Werte und Tupel

• Numpy-Arrays

• Pandas-Datenrahmen

• Scikit-Learn-Datensätze und Schätzer

• PyTorch Modelle

• TensorFlow Modelle

• Die Booster-Klasse für XGBoost

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

• Die Maske DataFrames

• Die XGBoost Dmatrix-Klasse

• TensorFlow Datensätze und Unterklassen

• PyTorch Modelle

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

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.

• Aufgabe DataFrames

• Die XGBoost DMatric Klasse

• TensorFlow Datensätze und Unterklassen

• PyTorch Modelle

Bewährte Methoden für Dask

Dask ist eine Open-Source-Bibliothek, die für Parallelberechnungen in Python verwendet wird. In diesem Abschnitt wird Folgendes gezeigt.

• Wie übergebe ich einen Dask DataFrame an Ihre Remote-Funktion

• Wie konvertiert man zusammenfassende Statistiken von einem Dask DataFrame in einen Pandas DataFrame

Wie übergebe ich einen Dask an deine DataFrame Remote-Funktion

Dask DataFrames werden häufig zur Verarbeitung großer Datenmengen verwendet, da sie Datensätze enthalten können, die mehr Speicher benötigen, als verfügbar ist. Das liegt daran, dass ein Dask Ihre lokalen DataFrame Daten nicht in den Speicher lädt. Wenn Sie einen Dask DataFrame als Funktionsargument an Ihre Remote-Funktion übergeben, kann Dask anstelle der Daten selbst einen Verweis auf die Daten auf Ihrer lokalen Festplatte oder Ihrem Cloud-Speicher übergeben. Der folgende Code zeigt ein Beispiel für die Übergabe eines Dask DataFrame innerhalb Ihrer Remote-Funktion, der mit einem leeren Objekt arbeitet. DataFrame

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

Dask lädt die Daten vom Dask nur dann DataFrame in den Speicher, wenn Sie den verwenden. DataFrame Wenn Sie einen Dask DataFrame innerhalb einer Remote-Funktion verwenden möchten, geben Sie den Pfad zu den Daten an. Dask liest den Datensatz dann direkt von dem Datenpfad, den Sie angeben, wenn der Code ausgeführt wird.

Das folgende Codebeispiel zeigt, wie ein Dask DataFrame innerhalb der Remote-Funktion verwendet wird. clean Im Codebeispiel raw_data_path wird an clean statt an den DataFrame Dask übergeben. Wenn der Code ausgeführt wird, wird der Datensatz direkt von dem Speicherort eines Amazon-S3-Buckets gelesen, der in raw_data_path angegeben ist. Dann behält die persist Funktion den Datensatz im Speicher, um die nachfolgende random_split Funktion zu erleichtern, und schreibt ihn mithilfe von DataFrame API Dask-Funktionen in den Ausgabedatenpfad in einem S3-Bucket zurück.

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

Wie konvertiert man zusammenfassende Statistiken von einem Dask DataFrame in einen Pandas DataFrame

Zusammenfassungsstatistiken von einem Dask DataFrame können in einen Pandas umgewandelt werden, DataFrame indem die compute Methode aufgerufen wird, wie im folgenden Beispielcode gezeigt. Im Beispiel enthält der S3-Bucket einen großen Dask DataFrame , der weder in den Speicher noch in einen Pandas-Datenrahmen passt. Im folgenden Beispiel scannt eine Remote-Funktion den Datensatz und gibt einen Dask zurück, der die Ausgabestatistiken von an einen Pandas DataFrame enthält. describe DataFrame

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

Bewährte Methoden für den Kurs XGBoost DMatric

DMatrixist eine interne Datenstruktur, die von XGBoost zum Laden von Daten verwendet wird. Ein DMatrix Objekt kann nicht ausgewählt werden, um einfach zwischen Rechensitzungen hin- und herzuwechseln. Das direkte Übergeben von DMatrix Instanzen schlägt mit einem SerializationError fehl.

Wie übergibt man ein Datenobjekt an Ihre Remote-Funktion und trainiert damit XGBoost

Um einen Pandas DataFrame in eine DMatrix Instanz umzuwandeln und ihn für das Training in Ihrer Remote-Funktion zu verwenden, übergeben Sie ihn direkt an die Remote-Funktion, wie im folgenden Codebeispiel gezeigt.

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)                        

Bewährte Methoden für TensorFlow Datensätze und Unterklassen

TensorFlow Datensätze und Unterklassen sind interne Objekte, die zum Laden von Daten während des Trainings TensorFlow verwendet werden. TensorFlow Datensätze und Unterklassen können nicht ausgewählt werden, um problemlos zwischen Berechnungssitzungen zu wechseln. Die direkte Übergabe von Tensorflow-Datensätzen oder -Unterklassen schlägt mit einem SerializationError fehl. Verwenden Sie die Tensorflow-I/OAPIs, um Daten aus dem Speicher zu laden, wie im folgenden Codebeispiel gezeigt.

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", {})                        

Bewährte Methoden für Modelle PyTorch

PyTorch Modelle sind serialisierbar und können zwischen Ihrer lokalen Umgebung und der Remote-Funktion weitergegeben werden. Wenn Ihre lokale Umgebung und Ihre Remote-Umgebung unterschiedliche Gerätetypen haben, z. B. (GPUsundCPUs), können Sie ein trainiertes Modell nicht an Ihre lokale Umgebung zurückgeben. Wenn der folgende Code beispielsweise in einer lokalen Umgebung ohne entwickelt, GPUs aber in einer Instanz mit ausgeführt wirdGPUs, führt die direkte Rückgabe des trainierten Modells zu einemDeserializationError.

# 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                        

Um ein in einer GPU Umgebung trainiertes Modell in ein Modell zurückzuversetzen, das nur CPU Funktionen enthält, verwenden Sie die PyTorch Modell-I/O APIs direkt, wie im folgenden Codebeispiel gezeigt.

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 werden Ihre serialisierten Daten gespeichert

Wenn Sie eine Remote-Funktion aufrufen, serialisiert Ihre Funktionsargumente und Rückgabewerte während der Eingabe- und Ausgabephase SageMaker automatisch. 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.

SageMaker Erstellt unter dem Stammverzeichnis 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:

• Speichern Sie Ihr Modell und Ihre Modellartefakte in einem Format, das von Ihrer Python-Version und Ihrem ML-Framework unabhängig ist.

• Wenn Sie Ihr Python- oder ML-Framework aktualisieren, greifen Sie von Ihrem Langzeitspeicher aus auf Ihre Modellergebnisse zu.

Wichtig

Um Ihre serialisierten Daten nach einer bestimmten Zeit zu löschen, legen Sie für Ihren S3-Bucket eine lebenslange Konfiguration fest.

Anmerkung

Dateien, die mit dem Python-Pickle-Modul serialisiert wurden, können weniger portabel sein als andere Datenformate wie Parquet CSV und. JSON Seien Sie vorsichtig, wenn Sie eingelagerte Dateien aus unbekannten Quellen laden.

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

Zugriff auf Ihre serialisierten Daten

Administratoren können Einstellungen für Ihre serialisierten Daten, einschließlich ihres Speicherorts und aller Verschlüsselungseinstellungen, in einer Konfigurationsdatei angeben. Standardmäßig werden die serialisierten Daten mit einem verschlüsselt AWS Key Management Service (AWS KMS) Schlüssel. Administratoren können den Zugriff auf das Stammverzeichnis, das Sie in Ihrer Konfigurationsdatei angeben, auch mit einer Bucket-Richtlinie einschränken. Die Konfigurationsdatei kann gemeinsam genutzt und projekt- und auftragsübergreifend verwendet werden. Weitere Informationen finden Sie unter Konfigurierungsdatei.

Verwenden Sie die RemoteExecutorAPI, um eine Funktion aufzurufen

Sie können die verwenden RemoteExecutorAPI, um eine Funktion aufzurufen. SageMaker Python wandelt SDK den Code innerhalb des RemoteExecutor Aufrufs in einen SageMaker Trainingsjob um. Der Trainingsauftrag ruft dann die Funktion als asynchronen Vorgang auf und gibt ein Future zurück. Wenn Sie den verwenden RemoteExecutorAPI, können Sie mehr als einen Trainingsjob 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 Instanz starten und die verwenden, API um eine Anforderung zur parallel 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.

Zukünftiger Kurs für die RemoteExecutor API

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.