G-genの佐伯です。当記事では Vertex AI の AutoML 及びバッチ予測の基本的な操作方法や、簡易で安価に予測データを収集する手法を解説します。後編では Vertex AI AutoML で作成した機械学習モデルをローカルの docker で動作させ、安価に予測値を取得する方法をご紹介します。

はじめに

検証内容

Vertex AI AutoML 表形式のモデルはエクスポートしてローカル環境で動作させることができます。

当記事では以下のドキュメントを参考にして、検証してみます。

参考 : AutoML 表形式のモデルをエクスポートする

当記事で活用するモデル

当記事では、以下の記事で使用した米国の国勢調査局のデータベースから抽出した、ある個人に関する様々な情報（年齢、職業、結婚歴など）と「年収（income）が$50,000よりも高いか、$50,000以下か」が記録されたデータセットから作成されたモデルを活用します。

blog.g-gen.co.jp

Vertex AI Model Registry からモデルをエクスポート

Vertex AI の Model Registry から、Cloud Storage にモデルをエクスポートします。

今回は saikio-vertexai-test という名前の Cloud Storage バケットを使用します。

Cloud Storage へのエクスポートが完了したら、ローカル環境から以下のコマンドを実行し、モデルを Cloud Storage からローカルの Linux 環境にコピーします。

$ gsutil cp -r gs://vertex_ai_test2 .

コピーした Cloud Storage バケット名のディレクトリに移動します。

以降、ローカル環境で実行するコマンドは、ここから実行します。

cd vertex_ai_test2

$ ls
model-8435048588018450432

ローカル Docker コンテナで推論を実行

エクスポートしたモデルをローカルの Docker 環境にデプロイし、予測リクエストを処理してみます。基本的にはドキュメントに記載されている手順を参考にします。

モデルは以下のようなディレクトリの配下にコピーされています。

./model-<model-id>/tf-saved-model/<export-timestamp>

Docker ではタイムスタンプが含まれるディレクトリが無効にされてしまうので、ディレクトリ名を変更し、model というディレクトリ名に変更しました。以上のことから、今回モデルが存在するディレクトリは ~/vertex_ai_test2/model-8435048588018450432/tf-saved-model/model としています。

次に、Google Cloud が提供する、 Vertex AI のモデルを実行するためのモデルサーバーのコンテナイメージを使用して、モデルをデプロイします。

先述のドキュメント通りにコマンドを実行した場合、 asia-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server-v1 からモデルサーバーの Docker イメージを pull しますが、このイメージを使用してデプロイした場合、予測リクエストに対してエラーが返り、リクエストを正常に処理することができませんでした。

そこで、この Issue を参考に asia-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server:20230123_0325_RC00 のイメージを pull し、コンテナを実行します。ここでは、コンテナに serene_cerf という名前をつけています。

$ docker run -dit --name serene_cerf \
 -v `pwd`/model-8435048588018450432/tf-saved-model/model:/models/default \
 -p 8080:8080 \
 asia-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server:20230123_0325_RC00

docker稼働状況

以下のコマンドを実行して、1つだけのテストデータを作成します。

cd ~/vertex_ai_test2;
sudo vim request.json

作成したデータをvertex_ai_test2フォルダ内に配置します。

vertex_ai_test2フォルダ内で以下のコマンドを実行し、予測値を確認します。ローカルで動作する Docker コンテナに対して推論を実行させています。

curl -X POST --data @request.json http://localhost:8080/predict

{"predictions": [{"scores": [0.0018795928917825222, 0.9981204271316528], "classes": ["<=50K", ">50K"]}]}

無事、予測値が算出され99.8%以上の確率で、年収50K以上であることが確認できました。

トレーニングデータを大幅に増やし推論を実行

上記では、データが1つのみでしたがこれを更に300行と増加させて、予測させてみます。

前編のトレーニングデータで使用した『pred_data.csv』を用意します。

以下のpythonプログラム(AutoML_pred.py)にて、pred_data.csvを読み込ませ、予測値を算出してみます。

AutoML_pred.pyとpred_data.csvは同じ階層で実行。

import os
import csv
import subprocess, json
import sys, traceback
from loguru import logger
# out_put for log
logger.add("./AutoML_pred_log/my_error.log", rotation="100MB", retention="1 year")  
  
# key_columns  
data_key = ['age', 'workclass', 'fnlwgt', 'education', 'education.num', 'marital.status', 'occupation',
            'relationship','race', 'sex', 'capital.gain', 'capital.loss', 'hours.per.week', 'native.country']  
  
#データから予測値を出力し、得られたデータをcsv形式で出力  
def Data_Predict():  
  
    # predict data(data for csv_type)  
    MyPath = 'pred_data.csv'  
  
    # read pred_data.csv  
    with open(MyPath, "r", encoding="utf-8") as f:  
        reader = csv.reader(f)  
        rows = [dict(zip(data_key, row)) for row in reader]  
  
    try:
        # delete 1 row
        del rows[0]
  
        # all predictionaled data on making model
        out_put = list()
        for i in range(len(rows)):
            # setting data for curl command
            curl_data = {"instances": [rows[i]]}
  
            # curl command for request
            cmd = "cd ~/vertex_ai_test2;curl -H {} -X POST --data '{}' http://localhost:8080/predict".format(
                '"content-type: application/json"', json.dumps(curl_data))
  
            # shell=True,setting string from standard output
            result = subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
  
            my_data = json.loads(result)
  
            # dictionary data set
            mm = dict()
  
            # key of [income_<=50K_value]
            key1 = 'income_{}_value'.format(my_data["predictions"][0]["classes"][0])
            # value of [income_<=50K_value]
            val1 = my_data["predictions"][0]["scores"][0]
            # key of [income_>50K_value]
            key2 = 'income_{}_value'.format(my_data["predictions"][0]["classes"][1])
            # value of [income_>50K_value]
            val2 = my_data["predictions"][0]["scores"][1]
  
            mm[key1] = val1
            mm[key2] = val2
  
            # 予測値を蓄積
            out_put.append(mm)
  
    except Exception as e:
        logger.warning("exceptions2 : {} : {}".format(e, traceback.format_exc()))
  
    # csv出力
    with open('out_put.csv', 'w', encoding='utf-8', newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[key1,key2])
        writer.writeheader()
        writer.writerows(out_put)
  
  
if __name__ == '__main__':
  
    Data_Predict()