「Amazon SageMakerを使用して、Rayベースの機械学習ワークフローをオーケストレーションする」
オーケストレーションするRayベースの機械学習ワークフローをAmazon SageMakerで使用する
機械学習（ML）は、顧客がますます困難な問題を解決しようとするにつれて、ますます複雑になっています。この複雑さは、複数のマシンを使用して単一のモデルをトレーニングする分散MLの必要性につながることが多いです。これにより、複数のノード間でタスクを並列化することが可能になり、トレーニング時間の短縮、拡張性の向上、パフォーマンスの向上が実現されますが、分散ハードウェアの効果的な使用には重要な課題があります。データサイエンティストは、データの分割、負荷分散、障害耐性、拡張性などの課題に取り組まなければなりません。MLエンジニアは、並列化、スケジューリング、障害、再試行などを手動で処理する必要があり、複雑なインフラストラクチャコードが必要です。
この投稿では、分散MLのためのRayとAmazon SageMakerの利点について説明し、これらのフレームワークを使用してスケーラブルなMLワークフローを構築および展開する方法についてステップバイステップでガイドを提供します。
Rayは、オープンソースの分散コンピューティングフレームワークであり、MLモデルの分散トレーニングとサービングのための柔軟なフレームワークを提供します。データの前処理、分散トレーニング、ハイパーパラメータの調整、強化学習、モデルサービングなどの一般的なMLタスクに対するシンプルでスケーラブルなライブラリを介して、低レベルの分散システムの詳細を抽象化します。
SageMakerは、MLモデルの構築、トレーニング、展開のための完全に管理されたサービスです。RayはSageMakerの機能とシームレスに統合し、効率的で信頼性の高い複雑なMLワークロードを構築および展開します。RayとSageMakerの組み合わせにより、スケーラブルなMLワークフローのエンドツーエンドの機能が提供され、次のようなハイライトされた機能があります：
- RayはNVIDIA AIとの協業により、開発者が製品向けのLLMを構築、調整、トレーニング、スケールアップするのを支援します
- アンサンブル学習技術：Pythonでのランダムフォレストを使った手順解説
- 大規模言語モデルは安全性を自己評価できるのか？RAINに会ってください：ファインチューニングなしでAIのアライメントと防御を変革する革新的な推論方法
- Rayの分散アクターと並列性の構造は、分散アプリケーションの開発を簡素化します。
- Ray AIランタイム（AIR）は、開発から本番環境への移行の摩擦を軽減します。RayとAIRを使用すると、同じPythonコードをラップトップから大規模なクラスターまでシームレスにスケーリングできます。
- SageMakerの管理されたインフラストラクチャと、処理ジョブ、トレーニングジョブ、ハイパーパラメータチューニングジョブなどの機能は、分散コンピューティングのためにRayライブラリを使用できます。
- Amazon SageMaker Experimentsを使用すると、試行を迅速に反復し、追跡することができます。
- Amazon SageMaker Feature Storeは、モデルトレーニングのためのMLフィーチャを保存、取得、共有するためのスケーラブルなリポジトリを提供します。
- トレーニングされたモデルは、ガバナンスと管理のためにAmazon SageMaker Model Registryに格納、バージョン管理、追跡できます。
- Amazon SageMaker Pipelinesを使用すると、データの準備とトレーニングからモデルの展開までのエンドツーエンドのMLライフサイクルを自動化することができます。
ソリューションの概要
この投稿では、RayとSageMakerを併用する利点に焦点を当てています。SageMaker Pipelinesを使用して、エンドツーエンドのRayベースのMLワークフローを設定します。ワークフローには、Rayアクターを使用したデータの並列取り込み、Ray Dataを使用したデータの前処理、Ray Trainおよびハイパーパラメータ最適化（HPO）チューニングジョブを使用したモデルのトレーニングとハイパーパラメータの調整、最終的なモデルの評価とモデルレジストリへの登録が含まれます。
データには、8つの特徴（YEAR_BUILT、SQUARE_FEET、NUM_BEDROOM、NUM_BATHROOMS、LOT_ACRES、GARAGE_SPACES、FRONT_PORCH、およびDECK）からなる合成住宅データセットを使用し、モデルはPRICEを予測します。
MLワークフローの各ステージは、独自のスクリプトで区切られ、入力パラメータと出力パラメータを使用します。次のセクションでは、各ステップからの主なコードスニペットをハイライトします。完全なコードは、aws-samples-for-ray GitHubリポジトリで見つけることができます。
前提条件
この投稿に関連するSageMaker Python SDKを使用し、コードを実行するには、以下の前提条件が必要です：
- すべてのAWSリソースを含むAWSアカウント
- SageMaker Studioノートブック、SageMaker Feature Store、SageMaker Model Registry、およびSageMaker Pipelinesへのアクセス権を持つAWS Identity and Access Management（IAM）ロール
SageMaker Feature Storeにデータを取り込む
MLワークフローの最初のステップは、Amazon Simple Storage Service（Amazon S3）からCSV形式のソースデータファイルを読み取り、SageMaker Feature Storeに取り込むことです。SageMaker Feature Storeは、チームがMLフィーチャを作成、共有、管理しやすくするための専用のリポジトリです。特徴の発見、再利用、共有を簡素化し、開発の迅速化、顧客チーム内の協力の増加、コスト削減を実現します。
フィーチャストアへのフィーチャの取り込みは、次の手順で行われます：
- フィーチャグループを定義し、フィーチャストアでフィーチャグループを作成します。
- ソースデータをフィーチャストアに準備し、各行のデータにイベント時刻とレコードIDを追加します。
- Boto3 SDKを使用して、準備されたデータをフィーチャグループに取り込みます。
このセクションでは、Rayを使用してインジェスションタスクの並列処理を行うStep 3のみを強調します。このプロセスの完全なコードはGitHubリポジトリで確認できます。
インジェストフィーチャーメソッドはFeaturestoreというクラス内で定義されています。なお、Featurestoreクラスは@ray.remoteでデコレートされています。これにより、このクラスのインスタンスがRayアクターであることが示されます。Rayアクターは、Rayクラスタ内の異なるノードで実行される複数のタスクから同時にアクセスできる内部状態を持つ分散オブジェクトであり、ミュータブルな状態を管理しカプセル化するための方法を提供します。アクターは複雑な状態を持つアプリケーションを分散環境で構築するために有用です。アクターにはリソース要件も指定できます。この場合、FeatureStoreクラスの各インスタンスには0.5 CPUが必要です。以下のコードを参照してください。
@ray.remote(num_cpus=0.5)
class Featurestore:
def ingest_features(self,feature_group_name, df, region):
"""
フィーチャーストアグループにフィーチャーをインジェストする
Args:
feature_group_name (str): フィーチャーグループ名
data_path (str): CSV形式のトレーニング/検証/テストデータへのパス
"""
...remote演算子を呼び出すことでアクターと対話することができます。次のコードでは、スクリプトに入力引数として望ましいアクターの数を渡します。データはアクターの数に基づいてパーティション分割され、リモート並列プロセスに渡されてフィーチャーストアにインジェストされます。現在のタスクの実行をブロックし、リモート計算が完了し結果が利用可能になるまで待機するために、オブジェクト参照でgetを呼び出すことができます。結果が利用可能になると、ray.getは結果を返し、現在のタスクの実行が継続されます。
import modin.pandas as pd
import ray
df = pd.read_csv(s3_path)
data = prepare_df_for_feature_store(df)
# パーティションに分割する
partitions = [ray.put(part) for part in np.array_split(data, num_actors)]
# アクターを開始し、パーティションをループで割り当てる
actors = [Featurestore.remote() for _ in range(args.num_actors)]
results = []
for actor, partition in zip(actors, input_partitions):
results.append(actor.ingest_features.remote(
args.feature_group_name,
partition, args.region
)
)
ray.get(results)トレーニング、検証、およびテスト用のデータの準備
このステップでは、Ray Datasetを使用してデータセットを効率的に分割、変換、スケーリングして機械学習の準備を行います。Ray Datasetは、さまざまなストレージシステムとファイルフォーマットをサポートし、分散データをRayに読み込むための標準的な方法を提供します。並列変換、シャッフル、グルーピング、集計などの一般的なMLデータ前処理操作のためのAPIを持っています。Ray Datasetは、状態の設定とGPUアクセラレーションを必要とする操作も処理します。また、Spark、Pandas、NumPyなどの他のデータ処理ライブラリやTensorFlow、PyTorchなどのMLフレームワークともスムーズに統合します。これにより、Rayの上にエンドツーエンドのデータパイプラインとMLワークフローを構築することができます。目標は、分散データ処理とMLを実践者や研究者にとってより簡単にすることです。
このデータ前処理を実行するスクリプトのセクションを見てみましょう。まず、フィーチャーストアからデータをロードします。
def load_dataset(feature_group_name, region):
"""
オフラインフィーチャーストアのS3ロケーションからRayデータセットとしてデータをロードする
Args:
feature_group_name (str): フィーチャーグループの名前
Returns:
ds (ray.data.dataset): フィーチャーストアからのリクエストされたデータを含むRayデータセット
"""
session = sagemaker.Session(boto3.Session(region_name=region))
fs_group = FeatureGroup(
name=feature_group_name,
sagemaker_session=session
)
fs_data_loc = fs_group.describe().get("OfflineStoreConfig").get("S3StorageConfig").get("ResolvedOutputS3Uri")
# MLの問題に関連しないフィーチャーストアによって追加された列を削除する
cols_to_drop = ["record_id", "event_time","write_time",
"api_invocation_time", "is_deleted",
"year", "month", "day", "hour"]
ds = ray.data.read_parquet(fs_data_loc)
ds = ds.drop_columns(cols_to_drop)
print(f"{fs_data_loc}のカウントは{ds.count()}です")
return ds次に、ray.dataライブラリから利用可能な上位の抽象化を使用してデータを分割し、スケーリングします：
def split_dataset(dataset, train_size, val_size, test_size, random_state=None):
"""
データセットをトレーニング、検証、テストのサンプルに分割します
Args:
dataset (ray.data.Dataset): 入力データ
train_size (float): トレーニングデータセットとして使用するデータの割合
val_size (float): 検証データセットとして使用するデータの割合
test_size (float): テストデータセットとして使用するデータの割合
random_state (int): 複数の関数呼び出し間で再現可能な出力を得るために、整数を指定します。
Returns:
train_set (ray.data.Dataset): トレーニングデータセット
val_set (ray.data.Dataset): 検証データセット
test_set (ray.data.Dataset): テストデータセット
"""
# このデータセットを固定されたランダムシードでシャッフルします。
shuffled_ds = dataset.random_shuffle(seed=random_state)
# データをトレーニング、検証、テストのデータセットに分割します
train_set, val_set, test_set = shuffled_ds.split_proportionately([train_size, val_size])
return train_set, val_set, test_set
def scale_dataset(train_set, val_set, test_set, target_col):
"""
train_setにStandardScalerを適合させ、val_setとtest_setに適用します
Args:
train_set (ray.data.Dataset): トレーニングデータセット
val_set (ray.data.Dataset): 検証データセット
test_set (ray.data.Dataset): テストデータセット
target_col (str): ターゲットカラム
Returns:
train_transformed (ray.data.Dataset): スケーリングされたトレーニングデータ
val_transformed (ray.data.Dataset): スケーリングされた検証データ
test_transformed (ray.data.Dataset): スケーリングされたテストデータ
"""
tranform_cols = dataset.columns()
# スケーリングされる対象のカラムを除外します
tranform_cols.remove(target_col)
# 標準スケーラーを設定します
standard_scaler = StandardScaler(tranform_cols)
# トレーニングデータセットにスケーラーを適合させます
print("トレーニングデータにスケーリングを適用してデータセットを変換しています...")
train_set_transformed = standard_scaler.fit_transform(train_set)
# 検証データセットとテストデータセットにスケーラーを適用します
print("検証データセットとテストデータセットを変換しています...")
val_set_transformed = standard_scaler.transform(val_set)
test_set_transformed = standard_scaler.transform(test_set)
return train_set_transformed, val_set_transformed, test_set_transformed処理されたトレーニング、検証、およびテストデータセットはAmazon S3に保存され、後続のステップの入力パラメータとして渡されます。
モデルのトレーニングとハイパーパラメータの最適化を実行する
データの前処理が完了し、モデリングの準備が整ったので、予測性能を最大化するためにいくつかの機械学習モデルをトレーニングし、ハイパーパラメータを微調整することができます。XGBoost-Rayを使用して、複数のノードとGPUを使用して大規模なデータセットでXGBoostモデルをトレーニングすることができる、Ray上に構築されたXGBoostの分散バックエンドを使用しています。これにより、XGBoostのトレーニングと予測のAPIを簡単に置き換えるだけでなく、分散データ管理とトレーニングの複雑さを内部で処理することができます。
トレーニングを複数のノードに分散するために、RayHelperというヘルパークラスを利用しています。次のコードに示すように、トレーニングジョブのリソース構成を使用し、最初のホストをヘッドノードとして選択します：
class RayHelper():
def __init__(self, ray_port:str="9339", redis_pass:str="redis_password"):
....
self.resource_config = self.get_resource_config()
self.head_host = self.resource_config["hosts"][0]
self.n_hosts = len(self.resource_config["hosts"])ホスト情報を使用して、各トレーニングジョブインスタンスでRayを初期化する方法を決定することができます：
def start_ray(self):
head_ip = self._get_ip_from_host()
# 現在のホストがヘッドノードとして選択されたホストである場合、
# `ray start`を--headフラグを指定して実行し、これをヘッドノードとして実行します
if self.resource_config["current_host"] == self.head_host:
output = subprocess.run(['ray', 'start', '--head', '-vvv', '--port',
self.ray_port, '--redis-password', self.redis_pass,
'--include-dashboard', 'false'], stdout=subprocess.PIPE)
print(output.stdout.decode("utf-8"))
ray.init(address="auto", include_dashboard=False)
self._wait_for_workers()
print("全てのワーカーが存在しています")
print(ray.cluster_resources())
else:
# 現在のホストがヘッドノードではない場合、
# ipアドレスをhead_hostとして指定して`ray start`を実行します
time.sleep(10)
output = subprocess.run(['ray', 'start',
f"--address={head_ip}:{self.ray_port}",
'--redis-password', self.redis_pass, "--block"], stdout=subprocess.PIPE)
print(output.stdout.decode("utf-8"))
sys.exit(0)トレーニングジョブが開始されると、RayHelperのインスタンスでstart_ray()メソッドを呼び出すことにより、Rayクラスタを初期化できます：
if __name__ == '__main__':
ray_helper = RayHelper()
ray_helper.start_ray()
args = read_parameters()
sess = sagemaker.Session(boto3.Session(region_name=args.region))次に、XGBoost-Rayを使用してXGBoostトレーナーをトレーニングします：
def train_xgboost(ds_train, ds_val, params, num_workers, target_col = "price") -> Result:
"""
XGBoostトレーナーを作成し、トレーニングを行い、結果を返します。
Args:
ds_train (ray.data.dataset): トレーニングデータセット
ds_val (ray.data.dataset): 検証データセット
params (dict): ハイパーパラメータ
num_workers (int): トレーニングを分散させるためのワーカーの数
target_col (str): ターゲットカラム
Returns:
result (ray.air.result.Result): トレーニングジョブの結果
"""
train_set = RayDMatrix(ds_train, 'PRICE')
val_set = RayDMatrix(ds_val, 'PRICE')
evals_result = {}
trainer = train(
params=params,
dtrain=train_set,
evals_result=evals_result,
evals=[(val_set, "validation")],
verbose_eval=False,
num_boost_round=100,
ray_params=RayParams(num_actors=num_workers, cpus_per_actor=1),
)
output_path=os.path.join(args.model_dir, 'model.xgb')
trainer.save_model(output_path)
valMAE = evals_result["validation"]["mae"][-1]
valRMSE = evals_result["validation"]["rmse"][-1]
print('[3] #011validation-mae:{}'.format(valMAE))
print('[4] #011validation-rmse:{}'.format(valRMSE))
local_testing = False
try:
load_run(sagemaker_session=sess)
except:
local_testing = True
if not local_testing: # SageMakerトレーニングを使用する場合は実験を追跡する
with load_run(sagemaker_session=sess) as run:
run.log_metric('validation-mae', valMAE)
run.log_metric('validation-rmse', valRMSE)trainerをインスタンス化する際にRayParamsを渡すことに注意してください。これは、アクターの数とアクターごとのCPUの数を指定します。XGBoost-Rayはこの情報を使用して、Rayクラスタに接続されたすべてのノードにトレーニングを分散します。
SageMaker Python SDKに基づいてXGBoostエスティメーターオブジェクトを作成し、HPOジョブに使用します。
SageMaker Pipelinesを使用して前述のステップをオーケストレーションする
エンドツーエンドのスケーラブルで再利用可能なMLワークフローを構築するには、CI/CDツールを使用して前述のステップをパイプラインに組み込む必要があります。SageMaker Pipelinesは、SageMaker、SageMaker Python SDK、およびSageMaker Studioとの直接的な統合を持っています。この統合により、簡単に使用できるPython SDKを使用してMLワークフローを作成し、SageMaker Studioを使用してワークフローを視覚化および管理することができます。また、パイプラインの実行中にデータの履歴を追跡し、キャッシュのためのステップを指定することもできます。
SageMaker Pipelinesは、MLワークフローの構築に必要なステップを含む有向非巡回グラフ（DAG）を作成します。各パイプラインは、ステップ間のデータ依存関係によって結びつけられた一連のステップであり、パラメータ化されており、パイプラインの各実行に対して入力変数をパラメータとして提供できます。SageMaker Pipelinesには、ParameterString、ParameterInteger、ParameterFloat、およびParameterBooleanの4種類のパイプラインパラメータがあります。このセクションでは、一部の入力変数をパラメータ化し、ステップのキャッシュ設定を行います：
processing_instance_count = ParameterInteger(
name='ProcessingInstanceCount',
default_value=1
)
feature_group_name = ParameterString(
name='FeatureGroupName',
default_value='fs-ray-synthetic-housing-data'
)
bucket_prefix = ParameterString(
name='Bucket_Prefix',
default_value='aws-ray-mlops-workshop/feature-store'
)
rmse_threshold = ParameterFloat(name="RMSEThreshold", default_value=15000.0)
train_size = ParameterString(
name='TrainSize',
default_value="0.6"
)
val_size = ParameterString(
name='ValidationSize',
default_value="0.2"
)
test_size = ParameterString(
name='TestSize',
default_value="0.2"
)
cache_config = CacheConfig(enable_caching=True, expire_after="PT12H")2つの処理ステップを定義します：1つはSageMaker Feature Storeの取り込み、もう1つはデータの準備です。これらは以前に説明したステップと非常に似ているはずです。新しいコード行はステップの定義の後にあるProcessingStepで、処理ジョブの設定を取り込み、パイプラインのステップとして含めることができます。また、データの準備ステップがSageMaker Feature Storeの取り込みステップに依存するよう指定します。以下のコードを参照してください：
feature_store_ingestion_step = ProcessingStep(
name='FeatureStoreIngestion',
step_args=fs_processor_args,
cache_config=cache_config
)
preprocess_dataset_step = ProcessingStep(
name='PreprocessData',
step_args=processor_args,
cache_config=cache_config
)
preprocess_dataset_step.add_depends_on([feature_store_ingestion_step])同様に、モデルのトレーニングとチューニングのステップを構築するためには、モデルのトレーニングステップのコードの後にTuningStepの定義を追加する必要があります。これにより、SageMakerハイパーパラメータのチューニングをパイプラインのステップとして実行できるようになります：
tuning_step = TuningStep(
name="HPTuning",
tuner=tuner,
inputs={
"train": TrainingInput(
s3_data=preprocess_dataset_step.properties.ProcessingOutputConfig.Outputs[
"train"
].S3Output.S3Uri,
content_type="text/csv"
),
"validation": TrainingInput(
s3_data=preprocess_dataset_step.properties.ProcessingOutputConfig.Outputs[
"validation"
].S3Output.S3Uri,
content_type="text/csv"
)
},
cache_config=cache_config,
)
tuning_step.add_depends_on([preprocess_dataset_step])チューニングステップの後、最良のモデルをSageMakerモデルレジストリに登録することを選択します。モデルの品質を制御するために、パイプラインの入力パラメータrmse_thresholdで定義された閾値と最良のモデルの目標メトリック（RMSE）を比較する最小品質ゲートを実装します。この評価を行うために、評価スクリプトを実行する別の処理ステップを作成します。モデルの評価結果はプロパティファイルとして保存されます。プロパティファイルは、他のステップを実行する方法を決定するために処理ステップの結果を分析する際に特に便利です。以下のコードを参照してください：
# 他のステップがこれらの結果にアクセスできるようにモデル評価結果を保存する場所を指定します
evaluation_report = PropertyFile(
name='EvaluationReport',
output_name='evaluation',
path='evaluation.json',
)
# ProcessingStepはHPOステップから選択されたモデルのパフォーマンスを評価するために使用されます。
# この場合、トップのパフォーマンスを示すモデルが評価されます。
evaluation_step = ProcessingStep(
name='EvaluateModel',
processor=evaluation_processor,
inputs=[
ProcessingInput(
source=tuning_step.get_top_model_s3_uri(
top_k=0, s3_bucket=bucket, prefix=s3_prefix
),
destination='/opt/ml/processing/model',
),
ProcessingInput(
source=preprocess_dataset_step.properties.ProcessingOutputConfig.Outputs['test'].S3Output.S3Uri,
destination='/opt/ml/processing/test',
),
],
outputs=[
ProcessingOutput(
output_name='evaluation', source='/opt/ml/processing/evaluation'
),
],
code='./pipeline_scripts/evaluate/script.py',
property_files=[evaluation_report],
)パイプラインに最良のモデルをSageMakerモデルレジストリに登録するためにModelStepを定義します。最良のモデルが予め定義された品質チェックをパスしない場合、エラーメッセージを出力するためにFailStepを追加します：
register_step = ModelStep(
name='RegisterTrainedModel',
step_args=model_registry_args
)
metrics_fail_step = FailStep(
name="RMSEFail",
error_message=Join(on=" ", values=["RMSEが閾値を超えたため、実行に失敗しました：", rmse_threshold]),
)次に、ConditionStepを使用して、モデル登録ステップまたは失敗ステップがパイプラインで次に実行されるかどうかを評価します。この場合、最良のモデルのRMSEスコアが閾値よりも低い場合、最良のモデルが登録されます。
# モデルの品質を評価し、実行を分岐させるためのCondition step
cond_lte = ConditionLessThanOrEqualTo(
left=JsonGet(
step_name=evaluation_step.name,
property_file=evaluation_report,
json_path='regression_metrics.rmse.value',
),
right=rmse_threshold,
)
condition_step = ConditionStep(
name='CheckEvaluation',
conditions=[cond_lte],
if_steps=[register_step],
else_steps=[metrics_fail_step],
)最後に、すべての定義されたステップをパイプラインに組み合わせます：
pipeline_name = 'synthetic-housing-training-sm-pipeline-ray'
step_list = [
feature_store_ingestion_step,
preprocess_dataset_step,
tuning_step,
evaluation_step,
condition_step
]
training_pipeline = Pipeline(
name=pipeline_name,
parameters=[
processing_instance_count,
feature_group_name,
train_size,
val_size,
test_size,
bucket_prefix,
rmse_threshold
],
steps=step_list
)
# 注意：同じ名前の既存のパイプラインがある場合、上書きされます。
training_pipeline.upsert(role_arn=role_arn)前述のパイプラインは、SageMaker Studioで直接視覚化および実行することも、execution = training_pipeline.start()を呼び出すことで実行することもできます。次の図はパイプラインのフローを示しています。
さらに、パイプラインの実行によって生成されたアーティファクトの系譜を確認することもできます。
from sagemaker.lineage.visualizer import LineageTableVisualizer
viz = LineageTableVisualizer(sagemaker.session.Session())
for execution_step in reversed(execution.list_steps()):
print(execution_step)
display(viz.show(pipeline_execution_step=execution_step))
time.sleep(5)モデルの展開
パイプラインランを介してSageMakerモデルレジストリに最適なモデルが登録された後、SageMakerの完全に管理されたモデル展開機能を使用して、モデルをリアルタイムエンドポイントに展開します。SageMakerには、さまざまなユースケースのニーズに対応するための他のモデル展開オプションもあります。詳細については、適切なオプションを選択する際に推論のためのモデルを展開するを参照してください。まず、モデルをSageMakerモデルレジストリに登録しましょう：
xgb_regressor_model = ModelPackage(
role_arn,
model_package_arn=model_package_arn,
name=model_name
)モデルの現在のステータスはPendingApprovalです。展開前にステータスをApprovedに設定する必要があります：
sagemaker_client.update_model_package(
ModelPackageArn=xgb_regressor_model.model_package_arn,
ModelApprovalStatus='Approved'
)
xgb_regressor_model.deploy(
initial_instance_count=1,
instance_type='ml.m5.xlarge',
endpoint_name=endpoint_name
)クリーンアップ
実験が終了したら、不要な料金を避けるためにリソースをクリーンアップすることを忘れないでください。リアルタイムエンドポイント、モデルグループ、パイプライン、およびフィーチャーグループを削除するには、DeleteEndpoint、DeleteModelPackageGroup、DeletePipeline、およびDeleteFeatureGroupのAPIを呼び出し、すべてのSageMaker Studioノートブックインスタンスをシャットダウンします。
結論
この記事では、SageMakerパイプラインを使用してRayベースのMLワークフローを段階的に実行する方法について説明しました。また、SageMakerパイプラインがサードパーティのMLツールと統合する能力も示しました。AWSには、Rayワークロードをスケーラブルかつ安全にサポートするさまざまなサービスがあり、パフォーマンスの優れた運用効率を確保します。今日行動を起こして、Amazon SageMakerパイプラインとRayで機械学習のワークフローを最適化するためのすべての機能を活用してください！
We will continue to update VoAGI; if you have any questions or suggestions, please contact us!
Was this article helpful?
93 out of 132 found this helpful
Related articles
