Memanggil fungsi jarak jauh

Untuk memanggil fungsi di dalam dekorator @remote, gunakan salah satu metode berikut:

Jika Anda menggunakan metode dekorator @remote untuk memanggil fungsi, pekerjaan pelatihan akan menunggu fungsi selesai sebelum memulai tugas baru. Namun, jika Anda menggunakan RemoteExecutor API, Anda dapat menjalankan lebih dari satu pekerjaan secara paralel. Bagian berikut menunjukkan kedua cara memanggil fungsi.

Gunakan dekorator @remote untuk memanggil fungsi

Anda dapat menggunakan dekorator @remote untuk membuat anotasi suatu fungsi. SageMaker AI akan mengubah kode di dalam dekorator menjadi pekerjaan SageMaker pelatihan. Pekerjaan pelatihan kemudian akan memanggil fungsi di dalam dekorator dan menunggu pekerjaan selesai. Contoh kode berikut menunjukkan cara mengimpor pustaka yang diperlukan, memulai instance SageMaker AI, dan membuat anotasi perkalian matriks dengan dekorator @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()

Dekorator didefinisikan sebagai berikut.

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

Saat Anda memanggil fungsi yang didekorasi, SageMaker Python SDK memuat pengecualian apa pun yang ditimbulkan oleh kesalahan ke dalam memori lokal. Dalam contoh kode berikut, panggilan pertama ke fungsi divide selesai dengan sukses dan hasilnya dimuat ke memori lokal. Pada panggilan kedua ke fungsi membagi, kode mengembalikan kesalahan dan kesalahan ini dimuat ke memori lokal.

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)

Cara mengubah nilai variabel lokal

Fungsi dekorator dijalankan pada mesin jarak jauh. Mengubah variabel non-lokal atau argumen input di dalam fungsi yang didekorasi tidak akan mengubah nilai lokal.

Dalam contoh kode berikut, daftar dan dict ditambahkan di dalam fungsi dekorator. Ini tidak berubah ketika fungsi dekorator dipanggil.

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

Untuk mengubah nilai variabel lokal yang dideklarasikan di dalam fungsi dekorator, kembalikan variabel dari fungsi. Contoh kode berikut menunjukkan bahwa nilai variabel lokal diubah ketika dikembalikan dari fungsi.

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

Serialisasi dan deserialisasi data

Saat Anda memanggil fungsi jarak jauh, SageMaker AI secara otomatis membuat serial argumen fungsi Anda selama tahap input dan output. Argumen fungsi dan pengembalian diserialisasi menggunakan cloud pickle. SageMaker AI mendukung serialisasi objek dan fungsi Python berikut.

Berikut ini dapat digunakan dengan beberapa batasan.

Bagian berikut berisi praktik terbaik untuk menggunakan kelas Python sebelumnya dengan beberapa batasan dalam fungsi jarak jauh Anda, informasi tentang di mana SageMaker AI menyimpan data serial Anda dan cara mengelola akses ke sana.

Praktik terbaik untuk kelas Python dengan dukungan terbatas untuk serialisasi data jarak jauh

Anda dapat menggunakan kelas Python yang tercantum di bagian ini dengan batasan. Bagian selanjutnya membahas praktik terbaik untuk cara menggunakan kelas Python berikut.

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

Tempat SageMaker AI menyimpan data serial Anda

Saat Anda memanggil fungsi jarak jauh, SageMaker AI secara otomatis membuat serialisasi argumen fungsi Anda dan mengembalikan nilai selama tahap input dan output. Data serial ini disimpan di bawah direktori root di bucket S3 Anda. Anda menentukan direktori root,<s3_root_uri>, dalam file konfigurasi. Parameter job_name dibuat secara otomatis untuk Anda.

Di bawah direktori root, SageMaker AI membuat <job_name> folder, yang menyimpan direktori kerja Anda saat ini, fungsi serial, argumen untuk fungsi serial Anda, hasil, dan pengecualian apa pun yang muncul dari memanggil fungsi serial.

Di bawah<job_name>, direktori workdir berisi arsip zip dari direktori kerja Anda saat ini. Arsip zip mencakup file Python apa pun di direktori kerja Anda dan requirements.txt file, yang menentukan dependensi apa pun yang diperlukan untuk menjalankan fungsi jarak jauh Anda.

Berikut ini adalah contoh struktur folder di bawah bucket S3 yang Anda tentukan dalam file konfigurasi Anda.

<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

Direktori root yang Anda tentukan di bucket S3 Anda tidak dimaksudkan untuk penyimpanan jangka panjang. Data serial terkait erat dengan versi Python dan versi kerangka kerja pembelajaran mesin (ML) yang digunakan selama serialisasi. Jika Anda memutakhirkan versi Python atau kerangka kerja ML, Anda mungkin tidak dapat menggunakan data serial Anda. Sebagai gantinya, lakukan hal berikut.

Untuk informasi selengkapnya tentang apa yang harus disertakan dalam file konfigurasi untuk fungsi jarak jauh, lihat File Konfigurasi.

Akses ke data serial Anda

Secara default, SageMaker AI menyimpan data serial di bucket SageMaker AI default, yang mungkin dibagikan di seluruh proyek. Untuk mencegah modifikasi yang tidak sah, gunakan s3_root_uri parameter @remote dekorator atauRemoteExecutor. Tentukan bucket atau awalan S3 khusus untuk argumen fungsi serial Anda, hasil, dan arsip kode.

Contoh berikut menunjukkan cara menentukan lokasi S3 khusus dengan @remote dekorator:

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

Secara default, AWS Key Management Service (AWS KMS) mengenkripsi data serial. Anda dapat mengonfigurasi lokasi penyimpanan dan pengaturan enkripsi dalam file konfigurasi. Anda dapat berbagi file konfigurasi di seluruh proyek dan pekerjaan. Untuk informasi selengkapnya tentang file konfigurasi, lihat File konfigurasi.

Gunakan RemoteExecutor API untuk memanggil fungsi

Anda dapat menggunakan RemoteExecutor API untuk memanggil fungsi. SageMaker AI Python SDK akan mengubah kode di dalam RemoteExecutor panggilan menjadi pekerjaan pelatihan SageMaker AI. Pekerjaan pelatihan kemudian akan memanggil fungsi sebagai operasi asinkron dan mengembalikan masa depan. Jika Anda menggunakan RemoteExecutor API, Anda dapat menjalankan lebih dari satu pekerjaan pelatihan secara paralel. Untuk informasi lebih lanjut tentang futures di Python, lihat Futures.

Contoh kode berikut menunjukkan cara mengimpor pustaka yang diperlukan, menentukan fungsi, memulai instance SageMaker AI, dan menggunakan API untuk mengirimkan permintaan untuk menjalankan 2 pekerjaan secara paralel.

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

RemoteExecutorKelas adalah implementasi dari perpustakaan Concurrent .futures.executor.

Contoh kode berikut menunjukkan bagaimana mendefinisikan fungsi dan memanggilnya menggunakanRemoteExecutorAPI. Dalam contoh ini, RemoteExecutor akan mengirimkan 4 pekerjaan secara total, tetapi hanya 2 secara paralel. Dua pekerjaan terakhir akan menggunakan kembali cluster dengan overhead minimal.

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

max_parallel_jobParameter hanya berfungsi sebagai mekanisme pembatas laju tanpa mengoptimalkan alokasi sumber daya komputasi. Dalam contoh kode sebelumnya, RemoteExecutor tidak mencadangkan sumber daya komputasi untuk dua pekerjaan paralel sebelum pekerjaan apa pun dikirimkan. Untuk informasi selengkapnya tentang max_parallel_job atau parameter lain untuk dekorator @remote, lihat spesifikasi kelas fungsi dan metode jarak jauh.

Kelas masa depan untuk RemoteExecutor API

Kelas masa depan adalah kelas publik yang mewakili fungsi pengembalian dari pekerjaan pelatihan ketika dipanggil secara asinkron. Kelas masa depan mengimplementasikan kelas Concurrent .futures.future. Kelas ini dapat digunakan untuk melakukan operasi pada pekerjaan yang mendasarinya dan memuat data ke dalam memori.