Amazon SageMakerを使用して、ML推論アプリケーションをゼロから構築し、展開する

ゼロから始めて、Amazon SageMakerを使ったML推論アプリケーションの構築と展開

機械学習（ML）が主流となり、広範な採用が行われるにつれ、MLパワーの推論アプリケーションは、さまざまな複雑なビジネス問題を解決するためにますます一般化しています。これらの複雑なビジネス問題の解決策には、しばしば複数のMLモデルとステップを使用する必要があります。この記事では、Amazon SageMaker 上でカスタムコンテナを使用してMLアプリケーションを構築およびホストする方法を説明します。

Amazon SageMakerでは、モデルの展開に組み込みアルゴリズムと事前構築済みSageMakerのDockerイメージが提供されています。しかし、これらが必要に合わない場合は、Amazon SageMakerでカスタムコンテナを使用してホストすることも可能です。

ユーザーがAmazon SageMakerでホストするためにカスタムコンテナを必要とするケースはいくつかあります。

1. カスタムMLフレームワークまたはライブラリ：Amazon SageMakerの組み込みアルゴリズムや事前構築済みコンテナでサポートされていないMLフレームワークやライブラリを使用する予定がある場合、カスタムコンテナを作成する必要があります。
2. 特定のモデル：特定のドメインや業界では、組み込みのAmazon SageMakerオファリングでは利用できない特定のモデルアーキテクチャやカスタマイズされた前処理手法が必要な場合があります。
3. 独自のアルゴリズム：独自の独自のアルゴリズムを開発している場合、Amazon SageMakerにそれらを展開するためにカスタムコンテナが必要です。
4. 複雑な推論パイプライン：ML推論ワークフローには、特定の順序で実行する必要がある複雑なステップの一連のカスタムビジネスロジックが含まれる場合、BYOCを使用するとこれらのステップをより効率的に管理および調整できます。

ソリューションの概要

このソリューションでは、最新のscikit-learnおよびxgboostパッケージを使用して、Amazon SageMaker上でリアルタイムエンドポイントを使用したMLシリアル推論アプリケーションをホストする方法を示します。

• 「18/9から24/9までの週のトップ重要コンピュータビジョン論文」
• このAIニュースレターは、あなたが必要とするすべてです #66
• OpenAIのChatGPTが音声と画像の機能を発表：AI対話における革命的な飛躍

最初のコンテナは、scikit-learnモデルを使用して生データを特徴量化された列に変換します。数値列にはStandardScalerを適用し、カテゴリカルな列にはOneHotEncoderを適用します。

2つ目のコンテナは、事前トレーニングされたXGboostモデル（つまり、予測モデル）をホストします。予測モデルは特徴量化された入力を受け入れ、予測を出力します。

最後に、特徴量生成器と予測モデルをシリアル推論パイプラインとしてAmazon SageMakerリアルタイムエンドポイントに展開します。

推論アプリケーション内で別々のコンテナを持つことの利点について考えてみましょう。

• 分離 – パイプラインのさまざまなステップは、明確に定義された目的を持ち、基になる依存関係のために別々のコンテナで実行する必要があります。これにより、パイプラインを適切に構造化するのにも役立ちます。
• フレームワーク – パイプラインのさまざまなステップは、特定の目的に適したフレームワーク（例：scikitやSpark ML）を使用して実行する必要がありますので、別々のコンテナ上で実行する必要があります。
• リソースの分離 – パイプラインのさまざまなステップは、異なるリソース消費要件を持ち、柔軟性と制御性を高めるために別々のコンテナで実行する必要があります。
• メンテナンスとアップグレード – 運用上の観点から、これにより機能的な分離が促進され、他のモデルに影響を与えることなく個々のステップをより簡単にアップグレードまたは変更することができます。

さらに、個々のコンテナのローカルビルドは、お気に入りのツールや統合開発環境（IDE）を使用した開発とテストの反復プロセスに役立ちます。コンテナが準備できたら、Amazon SageMakerエンドポイントを使用してAWSクラウドに展開し、推論に使用できます。

完全な実装（コードスニペットを含む）は、こちらのGithubリポジトリで利用できます。

<!–

前提条件

ローカル環境でこれらのカスタムコンテナを最初にテストするために、ローカルコンピュータにDocker Desktopをインストールする必要があります。Dockerコンテナのビルドには慣れている必要があります。

さらに、このアプリケーションをエンドツーエンドでテストするためには、Amazon SageMaker、Amazon ECR、およびAmazon S3へのアクセス権を持つAWSアカウントが必要です。

最新バージョンのBoto3とAmazon SageMaker Pythonパッケージがインストールされていることを確認してください:

pip install --upgrade boto3 sagemaker scikit-learn

ソリューションの手順

カスタム特徴量変換コンテナのビルド

最初のコンテナである特徴量変換コンテナをビルドするために、scikit-learnモデルを使用してアワビデータセットの生の特徴量を処理します。前処理スクリプトでは、欠損値の処理のためにSimpleImputer、数値列の正規化のためにStandardScaler、カテゴリカル列の変換のためにOneHotEncoderを使用します。変換器を適合させた後、モデルをjoblib形式で保存します。保存されたモデルアーティファクトをAmazon Simple Storage Service (Amazon S3) バケットに圧縮してアップロードします。

次に、特徴量変換モデルのカスタム推論コンテナを作成するために、nginx、gunicorn、flaskパッケージ、および特徴量変換モデルに必要な他の依存関係を含むDockerイメージをビルドします。

Nginx、gunicorn、およびFlaskアプリはAmazon SageMakerリアルタイムエンドポイントでモデルのサービングスタックとして機能します。

Amazon SageMakerでホストするためのカスタムコンテナを使用する場合、コンテナ内で起動された後、推論スクリプトは以下のタスクを実行することを確認する必要があります:

1. モデルの読み込み: 推論スクリプト(preprocessing.py)は、コンテナ内でモデルを読み込むために/opt/ml/modelディレクトリを参照する必要があります。Amazon S3のモデルアーティファクトは、パス/opt/ml/modelにダウンロードされ、コンテナにマウントされます。
2. 環境変数: カスタム環境変数をコンテナに渡すためには、Model作成ステップまたはEndpoint作成ステップで環境変数を指定する必要があります。
3. APIの要件: 推論スクリプトはFlaskアプリケーションとして、/pingおよび/invocationsの2つのルートを実装する必要があります。 /ping APIはヘルスチェックに使用され、/invocations APIは推論リクエストを処理します。
4. ログ出力: 推論スクリプトの出力ログは、標準出力(stdout)および標準エラー(stderr)ストリームに書き込まれる必要があります。これらのログは、Amazon SageMakerによってAmazon CloudWatchにストリームされます。

次に、preprocessing.pyの一部を示すスニペットです。ここでは、/pingおよび/invocationsの実装が行われています。

完全な実装については、featurizerフォルダ内のpreprocessing.pyを参照してください。

```pythondef load_model():    # featurizerモデルファイルのパスを構築する    ft_model_path = os.path.join(MODEL_PATH, "preprocess.joblib")    featurizer = None    try:        # モデルファイルを開き、joblibを使用してfeaturizerをロードする        with open(ft_model_path, "rb") as f:            featurizer = joblib.load(f)            print("Featurizerモデルをロードしました", flush=True)    except FileNotFoundError:        print(f"エラー：{ft_model_path}でFeaturizerモデルファイルが見つかりませんでした", flush=True)    except Exception as e:        print(f"Featurizerモデルの読み込みエラー：{e}", flush=True)    # エラーがある場合はNone、ロードされたfeaturizerモデルを返す    return featurizerdef transform_fn(request_body, request_content_type):    """    リクエストボディを使用可能なnumpy配列に変換する。    この関数は、リクエストボディとコンテンツタイプを入力とし、    予測モデルの入力として使用できる変換済みのnumpy配列を返します。    Parameters:        request_body (str): 入力データを含むリクエストボディ。        request_content_type (str): リクエストボディのコンテンツタイプ。    Returns:        data (np.ndarray): numpy配列として変換された入力データ。    """    # 入力データの列名を定義する    feature_columns_names = [        "sex",        "length",        "diameter",        "height",        "whole_weight",        "shucked_weight",        "viscera_weight",        "shell_weight",    ]    label_column = "rings"    # サポートされているコンテンツタイプかどうかを確認する（text/csv）    if request_content_type == "text/csv":        # featurizerモデルをロードする        featurizer = load_model()        # featurizerがColumnTransformerであるかどうかをチェックする        if isinstance(            featurizer, sklearn.compose._column_transformer.ColumnTransformer        ):            print(f"Featurizerモデルをロードしました", flush=True)        # リクエストボディから入力データをCSVファイルとして読み込む        df = pd.read_csv(StringIO(request_body), header=None)        # 入力データの列数に基づいて列名を割り当てる        if len(df.columns) == len(feature_columns_names) + 1:            # ラベルつきの例（リングラベルを含む）            df.columns = feature_columns_names + [label_column]        elif len(df.columns) == len(feature_columns_names):            # ラベルなしの例            df.columns = feature_columns_names        # featurizerを使用して入力データを変換する        data = featurizer.transform(df)        # 変換されたデータをnumpy配列として返す        return data    else:        # コンテンツタイプがサポートされていない場合はエラーを発生させる        raise ValueError("サポートされていないコンテンツタイプ：{}".format(request_content_type))@app.route("/ping", methods=["GET"])def ping():    # モデルをロードできるかどうかをチェックし、ステータスを設定する    featurizer = load_model()    status = 200 if featurizer is not None else 500    # 決定されたステータスコードでレスポンスを返す    return flask.Response(response="\n", status=status, mimetype="application/json")@app.route("/invocations", methods=["POST"])def invocations():    # JSONからdictに変換    print(f"Featurizer：content typeを受信しました：{flask.request.content_type}")    if flask.request.content_type == "text/csv":        # 入力データをデコードして変換する        input = flask.request.data.decode("utf-8")        transformed_data = transform_fn(input, flask.request.content_type)        # transformed_dataをCSV文字列にフォーマットする        csv_buffer = io.StringIO()        csv_writer = csv.writer(csv_buffer)        for row in transformed_data:            csv_writer.writerow(row)        csv_buffer.seek(0)        # transformed_dataをCSV文字列としてレスポンスで返す        return flask.Response(response=csv_buffer, status=200, mimetype="text/csv")    else:        print(f"受信：{flask.request.content_type}", flush=True)        return flask.Response(            response="Transformer：この予測モデルはCSVデータのみをサポートしています",            status=415,            mimetype="text/plain",        )```

featurizerとモデルサービングスタックを含むDockerイメージを作成する

カスタムベースイメージを使用してDockerfileを作成し、必要な依存関係をインストールします。

この場合、ベースイメージとしてpython:3.9-slim-busterを使用しています。使用用途に応じて、別のベースイメージに変更することもできます。

その後、Nginxの設定、GunicornのWebサーバーゲートウェイファイル、および推論スクリプトをコンテナにコピーします。また、nginxとgunicornプロセスをバックグラウンドで起動し、推論スクリプト（preprocessing.pyのFlaskアプリケーション）をコンテナのエントリーポイントとして設定するためのserveというPythonスクリプトも作成します。

以下はfeaturizerモデルのホスティング用Dockerfileの一部です。完全な実装については、Dockerfileを参照してください。

```dockerFROM python:3.9-slim-buster…# requirements.txtを/opt/programフォルダにコピーCOPY requirements.txt /opt/program/requirements.txt# requirements.txtにリストされたパッケージをインストールRUN pip3 install --no-cache-dir -r /opt/program/requirements.txt# code/ディレクトリの内容を/opt/programにコピーCOPY code/ /opt/program/# 作業ディレクトリを/opt/programに設定する（serveおよびinference.pyスクリプトが含まれる）WORKDIR /opt/program# サービング用にポート8080を公開EXPOSE 8080ENTRYPOINT ["python"]# code/ディレクトリ内のserveは、nginxとgunicornプロセスを起動するPythonスクリプトCMD [ "serve" ]```

ローカルでフィーチャライザ付きのカスタム推論イメージをテストする

Amazon SageMakerのローカルモードを使用して、フィーチャライザ付きのカスタム推論コンテナをビルドしてテストします。ローカルモードは、Amazon SageMakerでのジョブの起動なしで、処理、トレーニング、推論スクリプトをテストするのに最適です。ローカルテストの結果を確認した後、トレーニングおよび推論スクリプトをAmazon SageMakerにデプロイするために最小限の変更で簡単に適応できます。

フィーチャライザのカスタムイメージをローカルでテストするには、まず前に定義したDockerfileを使用してイメージをビルドします。次に、フィーチャライザモデル（preprocess.joblib）を/opt/ml/modelディレクトリにマウントしてコンテナを起動します。さらに、コンテナからホストへのポート8080のマッピングも行います。

起動した後、http://localhost:8080/invocationsに対して推論リクエストを送信できます。

コンテナをビルドおよび起動するには、ターミナルを開き、以下のコマンドを実行します。

以下のコードに示されているように、<IMAGE_NAME>を、コンテナのイメージ名に置き換える必要があることに注意してください。

以下のコマンドは、トレーニングされたscikit-learnモデル（preprocess.joblib）がmodelsというディレクトリの下に存在することを前提としています。

```shelldocker build -t <IMAGE_NAME> .``````shelldocker run –rm -v $(pwd)/models:/opt/ml/model -p 8080:8080 <IMAGE_NAME>```

コンテナが起動して実行されると、/pingおよび/invocationsの2つのルートをcurlコマンドでテストできます。

以下のコマンドをターミナルから実行します

```shell# ローカルエンドポイントの/pingルートをテストするcurl http://localhost:8080/ping# /invocationsに対して生のCSV文字列を送信し、エンドポイントが変換されたデータを返すcurl --data-raw 'I,0.365,0.295,0.095,0.25,0.1075,0.0545,0.08,9.0' -H 'Content-Type: text/csv' -v http://localhost:8080/invocations```

生の（変換されていない）データをhttp://localhost:8080/invocationsに送信すると、エンドポイントは変換されたデータで応答します。

次のような応答が表示されるはずです：

```shell* 127.0.0.1:8080 の試行中...* localhost (127.0.0.1) のポート 8080 に接続しました (#0)> POST /invocations HTTP/1.1> ホスト: localhost: 8080> ユーザーエージェント: curl/7.87.0> 受け入れ: */*> コンテンツタイプ: text/csv> コンテンツ長: 47>* バンドルをマルチユースをサポートしないようにマーク> HTTP/1.1 200 OK> サーバー: nginx/1.14.2> 日付: Sun, 09 Apr 2023 20:47:48 GMT> コンテンツタイプ: text/csv; charset=utf-8> コンテンツ長: 150> コネクション: keep -alive-1.3317586042173168, -1.1425409076053987, -1.0579488602777858, -1.177706547272754, -1.130662184748842,* ローカルホストへの接続 #0 を維持```

実行中のコンテナを終了し、ローカルのカスタムイメージをプライベートな Amazon Elastic Container Registry（Amazon ECR）リポジトリにタグ付けしてプッシュします。

Amazon ECR にログインするための以下のコマンドを参照してください。ローカルイメージに完全な Amazon ECR イメージパスをつけて、イメージを Amazon ECR にプッシュします。環境に応じて region と account の変数を置き換えてください。

```shell# ログインするaws ecr get-login-password - -region "${region}" |\docker login - -username AWS - -password-stdin ${account}".dkr.ecr."${region}".amazonaws.com# イメージにタグを付けて、プライベートな Amazon ECR にイメージをプッシュするdocker tag ${image} ${fullname}docker push $ {fullname}```

詳細は、リポジトリの作成およびAmazon ECR へのイメージのプッシュに関する AWS コマンドラインインターフェース（AWS CLI）コマンドを参照してください。

オプションのステップ

オプションとして、カスタム Docker イメージを Amazon ECR にデプロイし、リアルタイムのエンドポイントにフィーチャライザモデルを展開してライブテストを実行することもできます。フルなビルド、テスト、およびカスタムイメージを Amazon ECR にプッシュする手順に関しては、featurizer.ipynbのノートブックを参照してください。

Amazon SageMaker は推論エンドポイントを初期化し、モデルアーティファクトをコンテナ内の/opt/ml/modelディレクトリにコピーします。詳細はSageMakerがモデルアーティファクトを読み込む方法を参照してください。

カスタム XGBoost 推論コンテナのビルド

XGBoost 推論コンテナのビルドには、フィーチャライザコンテナのイメージをビルドするときと同じ手順を踏みます：

1. Amazon S3 から事前学習済みのXGBoostモデルをダウンロードします。
2. 事前学習済みのXGBoostモデルをロードし、フィーチャライザから受け取った変換された入力データをXGBoost.DMatrix形式に変換し、ブースタでpredictを実行し、予測結果を JSON 形式で返すinference.pyスクリプトを作成します。
3. モデルサービングスタックを構成するスクリプトと設定ファイル（つまり、nginx.conf、wsgi.py、serve）は同じままで、修正の必要はありません。
4. Dockerfile のベースイメージに Ubuntu:18.04 を使用しますが、これは必須ではありません。Ubuntu ベースイメージを使用しているのは、任意のベースイメージでコンテナをビルドできることをデモンストレーションするためです。
5. カスタム Docker イメージのビルド手順、ローカルでのイメージのテスト、テスト済みイメージの Amazon ECR へのプッシュ手順は、以前と同じです。

手順が前と類似しているため、短縮のために変更されたコーディングのみを以下に示します。

まず、inference.pyスクリプトです。以下は、/pingと/invocationsの実装を示すスニペットです。このファイルの詳細な実装については、inference.pyを、predictorフォルダ内でご覧ください。

```[email protected]("/ping", methods=["GET"])def ping():    """    モデルがロードされているかを確認してモデルサーバーのヘルスをチェックします。    モデルが正常にロードされている場合は、200のステータスコードを返します。エラーがある場合は、500のステータスコードを返します。    Returns:        flask.Response: ステータスコードとmimetypeを含むレスポンスオブジェクト    """    status = 200 if model is not None else 500    return flask.Response(response="\n", status=status, mimetype="application/json")@app.route("/invocations", methods=["POST"])def invocations():    """    入力データを前処理し、予測を行い、予測結果をJSONオブジェクトとして返すことで、予測リクエストを処理します。    この関数は、リクエストのコンテンツタイプがサポートされているかどうかをチェックし（text/csv; charset=utf-8）、    サポートされている場合は入力データをデコードし、前処理し、予測し、予測結果をJSONオブジェクトとして返します。    コンテンツタイプがサポートされていない場合は、415のステータスコードが返されます。    Returns:        flask.Response: 予測結果、ステータスコード、mimetypeを含むレスポンスオブジェクト    """    print(f"Predictor: 受信したコンテンツタイプ: {flask.request.content_type}")    if flask.request.content_type == "text/csv; charset=utf-8":        input = flask.request.data.decode("utf-8")        transformed_data = preprocess(input, flask.request.content_type)        predictions = predict(transformed_data)        # 予測結果をJSONオブジェクトとして返す        return json.dumps({"result": predictions})    else:        print(f"受信したコンテンツタイプ: {flask.request.content_type}", flush=True)        return flask.Response(            response=f"XGBPredictor: この予測器はCSVデータのみをサポートしています。受信したコンテンツタイプ: {flask.request.content_type}",            status=415,            mimetype="text/plain",        )```

以下は、予測モデルをホストするためのDockerfileの一部です。詳細な実装については、predictorフォルダ内のDockerfileを参照してください。

```dockerFROM ubuntu:18.04…# flask、gunicorn、xgboostなど、必要な依存関係をインストールするRUN pip3 --no-cache-dir install  flask  gunicorn  gevent  numpy  pandas  xgboost# code/ディレクトリの内容を/opt/programにコピーするCOPY code /opt/program# サーブとinference.pyのスクリプトがある/opt/programに作業ディレクトリを設定するWORKDIR /opt/program# サーブするためにポート8080を公開するEXPOSE 8080ENTRYPOINT ["python"]# serveは、code/ディレクトリの下にあるpythonスクリプトで、nginxとgunicornプロセスを起動しますCMD ["serve"]```

次に、このカスタム予測器イメージをビルドし、テストし、Amazon ECRのプライベートリポジトリにプッシュします。カスタムイメージをAmazon ECRにビルド、テスト、プッシュする完全な実装については、predictor.ipynbのノートブックを参照してください。

シリアル推論パイプラインのデプロイ

フィーチャライザと予測器のイメージをテストしてAmazon ECRにプッシュした後、モデルアーティファクトをAmazon S3バケットにアップロードします。

その後、2つのモデルオブジェクトを作成します：フィーチャライザ（preprocess.joblib）のための1つと予測器（xgboost-model）のためのもう1つ。作成したカスタムイメージのURIを指定します。

以下はその一部を示したものです。ビルド、テスト、カスタムイメージをAmazon ECRにプッシュするシリアル推論パイプラインの完全な実装については、serial-inference-pipeline.ipynbを参照してください。

```pythonsuffix = f"{str(uuid4())[:5]}-{datetime.now().strftime('%d%b%Y')}"# フィーチャライザモデル（SKLearnモデル）image_name = "<フィーチャライザ_IMAGE_NAME>"sklearn_image_uri = f"{account_id}.dkr.ecr.{region}.amazonaws.com/{image_name}:latest"featurizer_model_name = f""<フィーチャライザ_MODEL_NAME>-{suffix}"print(f"フィーチャライザモデルを作成中: {featurizer_model_name}")sklearn_model = Model(    image_uri=featurizer_ecr_repo_uri,    name=featurizer_model_name,    model_data=featurizer_model_data,    role=role,)# ECRリポジトリの完全な名前predictor_image_name = "<PREDICTOR_IMAGE_NAME>"predictor_ecr_repo_uri= f"{account_id}.dkr.ecr.{region}.amazonaws.com/{predictor_image_name}:latest"# 予測器モデル（XGBoostモデル）predictor_model_name = f"""<PREDICTOR_MODEL_NAME>-{suffix}"print(f"予測器モデルを作成中: {predictor_model_name}")xgboost_model = Model(    image_uri=predictor_ecr_repo_uri,    name=predictor_model_name,    model_data=predictor_model_data,    role=role,)```

これらのコンテナを直列に展開するためには、まずPipelineModelオブジェクトを作成し、featurizerモデルとpredictorモデルを同じ順序でPythonのリストオブジェクトに渡します。

次に、PipelineModel上の.deploy()メソッドを呼び出し、インスタンスのタイプとインスタンスの数を指定します。

```pythonfrom sagemaker.pipeline import PipelineModelpipeline_model_name = f"Abalone-pipeline-{suffix}"pipeline_model = PipelineModel(    name=pipeline_model_name,    role=role,    models=[sklearn_model, xgboost_model],    sagemaker_session=sm_session,)print(f"パイプラインモデル {pipeline_model_name} を展開しています...")predictor = pipeline_model.deploy(    initial_instance_count=1,    instance_type="ml.m5.xlarge",)```

この段階で、Amazon SageMakerはシリアルの推論パイプラインをリアルタイムのエンドポイントに展開します。エンドポイントがInServiceになるまで待機します。

これで、ライブなエンドポイントにいくつかの推論リクエストを送信してエンドポイントをテストできます。

詳細な実装については、serial-inference-pipeline.ipynbを参照してください。

クリーンアップ

テストが完了したら、必要のない料金を避けるために、この投稿でプロビジョニングされたリソースを削除するためのノートブックのクリーンアップセクションの手順に従ってください。推論インスタンスのコストの詳細については、Amazon SageMaker Pricingを参照してください。

```python# エンドポイント、モデルを削除try:    print(f"モデルを削除中: {pipeline_model_name}")    predictor.delete_model()except Exception as e:    print(f"モデルの削除エラー: {pipeline_model_name}\n{e}")    passtry:    print(f"エンドポイントを削除中: {endpoint_name}")    predictor.delete_endpoint()except Exception as e:    print(f"エンドポイントの削除エラー: {endpoint_name}\n{e}")    pass```

結論

この投稿では、Amazon SageMakerのリアルタイムエンドポイント上にカスタム推論コンテナを使用して直列のML推論アプリケーションを構築し、展開する方法を示しました。

このソリューションは、顧客がAmazon SageMakerで独自のカスタムコンテナを効率的にホスティングする方法を示しています。BYOCオプションにより、顧客は迅速にMLアプリケーションを構築し、Amazon SageMakerに展開できます。

ビジネスの重要なパフォーマンス指標（KPI）に関連するデータセットを使用して、このソリューションをお試しください。ソリューション全体を参照するには、このGitHubリポジトリをご覧ください。

参考文献

• Amazon SageMakerでのモデルホスティングパターン
• Amazon SageMakerでのカスタムコンテナの使用
• Amazon SageMakerでのシリアル推論パイプラインとしてのモデルホスティング

We will continue to update VoAGI; if you have any questions or suggestions, please contact us!