Amazon SageMakerとDataRobotで実現するモデル監視ダッシュボード

AIモデルの統合監視と可視化を実現するDataRobot MLOps

DataRobotでお客様におけるモデルのインテグレーションからMLOpsの実装までを支援しているAIサクセスエンジニアのブルーム・グレッグです。

ビジネスにおける機械学習モデルの活用が加速する中、「AIガバナンス」という言葉が注目を集めています。AIガバナンスとは、AIモデルの開発から運用にわたるライフサイクル全体を通して責任ある管理を行うことを指しますが、運用面ではモデルの精度やドリフトを継続的に監視し、適切な対策を講じることが重要な要素の一つです。

しかしモデル監視の実装には多くの課題が伴います。従来のモデル監視ではエキスパート向けの複雑な製品を活用する必要があり、技術的なハードルが高いことが問題でした。また、監視によって得られた膨大なログデータを、誰にでも理解しやすいダッシュボードにまとめる作業も、多大な労力を要します。

DataRobot MLOpsではモデル監視ダッシュボードをノーコードで簡単に構築できるという特徴があります（図1）。予測の利用回数や予測にかかった時間といった基本的な情報に加え、学習データと予測データとの間の特徴量の分布の違い(データドリフト)や精度などを示す事前定義済みのダッシュボードを利用することができ、エキスパートから非技術チームの監督者まで利用可能な監視画面をクリック作業だけでも構成することができます。

Amazon SageMakerで開発・デプロイしたモデルの統合監視

DataRobot MLOpsではDataRobotで開発したモデルだけでなく、Amazon SageMakerなど他のプラットフォームで開発、デプロイしたモデルにも利用することができます。本ブログではAmazon SageMakerで開発、デプロイしたモデルに対し、DataRobot MLOpsを活用してモデルの精度やドリフトを可視化する監視ダッシュボードを構成する方法をステップバイステップで解説します。

これらの手順は、DataRobot UIから手動で行うことも、DataRobotのAPIを用いてスクリプトから行うこともできます。本ブログではAPIを用いた操作を中心にご紹介します。本ブログでご紹介するスクリプト化は本ブログの添付ファイルからダウンロードいただけます。早速動かしてみたい方はぜひご利用ください。(SageMaker Notebookで実行する前提で作成されています)

アーキテクチャ

本ブログで構成するモデル監視の概要アーキテクチャは以下の通りです（図２）。

Amazon S3内に格納されたデータを元にSageMakerでモデルを構築します。推論用のコードにはDataRobot MLOpsライブラリの機能を使って監視データを収集するロジックを実装します。モデルはSageMakerの推論エンドポイントにデプロイします。推論が行われると、自動的に監視データが収集され、SQSチャンネルに送信されます。DataRobotはSQSから監視データを収集し、監視ダッシュボードを自動的に作成します。

準備

アカウントの用意

Amazon SageMakerとDataRobotのアカウントが必要です。DataRobotアカウントをお持ちではないかたはこちらでトライアルアカウントを作成することができます。

DataRobot AI Platform フリートライアル：https://www.datarobot.com/jp/trial/ 

構築の大まかな流れ：

1. Amazon SageMaker用カスタムコンテナの構築
2. モデルの学習
3. DataRobot MLOpsの設定
4. Amazon SageMakerで推論を実施
5. DataRobot MLOpsダッシュボードの確認

構築用のノートブックとアーティファクトの関係は以下の通りです（図３）。

Amazon SageMaker用カスタムコンテナの構築

Amazon SageMaker用のカスタムコンテナを構築するにあたって、このブログではAmazon SageMaker Examplesで公開されているワークショップ Building your own algorithm container を利用します。

本ブログの執筆時点では、上記ワークショップを動作させるために複数のアップデートが必要なため、上記ワークショップをもとに一連のアップデートを実施しビルド済みのコンテナをAmazon ECR Public Galleryで公開しています。こちらを取得し、お使いのECRへ登録します。

事前準備済みコンテナーイメージの取得とECRへの登録

docker pull public.ecr.aws/datarobot-support/sagemaker-mlops:latest

aws ecr get-login-password --region <リージョン> | docker login --username AWS 
--password-stdin <ECR URL>

# 実行例：
# aws ecr get-login-password --region us-east-1 | docker login --username AWS \
--password-stdin 123456789012.dkr.ecr.us-east-1.amazonaws.com

docker tag public.ecr.aws/datarobot-support/sagemaker-mlops:latest <ECR 
URL>/sagemaker-blog:latest

docker push <ECR URL>/sagemaker-blog:latest

[参考] モデル監視用コードの実装

本ステップは前掲の事前準備済みコンテナーイメージにすでに含まれているため、別途実施する必要はありません。ご自身でモデル監視コードを実装したい場合に参考にしてください。

datarobot-mlopsパッケージのインストール

Dockerfile内で、pythonをインストールするステップの直後に次の1行を追加し、datarobot-mlopsライブラリをインストールするようにします。これにより、SageMakerコンテナが監視データをDataRobot MLOpsへ送信できるようになります。

RUN pip --no-cache-dir install datarobot-mlops[aws]

datarobot-mlopsパッケージを使ったモデル監視の実装

predictor.py内のScoringServiceクラスを編集します。ここで、datarobot-mlopsライブラリを呼び出してメトリクスを収集し、SQSスプールチャネルに送信するようにします。また精度監視を実現するために、関連付けIDも送信データの中に含めるようにしています。変更の全体をご覧になりたい場合は、本ブログに添付のpredictor.pyをご覧ください。

from datarobot.mlops.mlops import MLOps

class ScoringService(object):
    model = None  
    mlops = None  

    @classmethod
    def get_mlops(cls):
        """MLOPS: initialize mlops library"""
        # Get environment parameters
        MLOPS_DEPLOYMENT_ID = os.environ.get('MLOPS_DEPLOYMENT_ID')
        MLOPS_MODEL_ID = os.environ.get('MLOPS_MODEL_ID')
        MLOPS_SQS_QUEUE = os.environ.get('MLOPS_SQS_QUEUE')
        MLOPS_ASSOC_ID_COLUMN = os.environ.get('MLOPS_ASSOC_ID_COLUMN')

        if cls.mlops == None:
            cls.mlops = MLOps() \
                .set_async_reporting(False) \
                .set_deployment_id(MLOPS_DEPLOYMENT_ID) \
                .set_model_id(MLOPS_MODEL_ID) \
                .set_sqs_spooler(MLOPS_SQS_QUEUE) \
                .set_association_id_column_name(MLOPS_ASSOC_ID_COLUMN) \
                .init()

        return cls.mlops

    @classmethod
    def get_model(cls):
        if cls.model == None:
            with open(os.path.join(model_path, "decision-tree-model.pkl"), \
"rb") as inp:
                cls.model = pickle.load(inp)
        return cls.model

    @classmethod
    def predict(cls, input:pd.DataFrame, assoc_col:list[str], \ pred_timestamps:list[str]):
        clf = cls.get_model() 
        class_names = json.loads(os.environ.get('CLASS_NAMES')) 
        start_time = time.time()         
        predictions_array = clf.predict_proba(input.values) 
        prediction = np.take(class_names, np.argmax(predictions_array, axis=1)) 
        execution_time = time.time() - start_time 
        ml_ops = cls.get_mlops()    
        ml_ops.report_deployment_stats(predictions_array.shape[0], \
execution_time * 1000)
        
        ml_ops.report_predictions_data(
            features_df=input,
            predictions=predictions_array.tolist(),
            class_names=class_names,
            association_ids=assoc_col
        )

        return prediction

モデルの学習

準備

必要なパッケージと変数をセットします。

import boto3
import numpy as np
import pandas as pd
from sagemaker import get_execution_role

# IAMロール名を取得
role = get_execution_role()
print(role)

IMAGE = <ECR URL>/sagemaker-blog:latest

SageMaker用のセッションを開始

from time import gmtime, strftime
import sagemaker as sage
sess = sage.Session()

データをS3にアップロード

今回は、上記のワークショップに含まれているデータセット(data/iris.csv)と同様のデータを本ブログの最下部に添付しています。

# S3用プレフィックス
prefix = "DEMO-scikit-byo-iris-v3"

TRAINING_DATASET = "iris.csv"
data_location = sess.upload_data(TRAINING_DATASET, key_prefix=prefix)

Estimatorを初期化してモデル学習を実施

print("ECR URL: ")
print(IMAGE)
print("Data Location: ")
print(data_location)

tree = sage.estimator.Estimator(
    IMAGE,
    role,
    1,
    "ml.c4.2xlarge",
    output_path="s3://{}/output".format(sess.default_bucket()),
    sagemaker_session=sess,
)

tree.fit(data_location)

DataRobot MLOpsの設定

準備

DataRobotの操作を効率化するためにDataRobot Pythonクライアントをインストールします。

!pip install datarobot

import json
import datarobot as dr

# カスタムモデル、外部環境、デプロイ等の名称に共通で使用する文字列を指定
PROJ_NAME = "SageMaker_MLOps_Demo"

DataRobotのUI内、「開発者ツール」のページでAPIキーを取得します。

#DataRobotのAPIエンドポイント。米国リージョンの場合は app.datarobot.com。お使いの環境をご確認ください。
endpoint="https://app.jp.datarobot.com/api/v2"

#DataRobotのUI内、「開発者ツール」ページへのリンク
print(f"{endpoint[:-7]}/account/developer-tools")

上記ページで取得したAPIキーを用いてDataRobot Pythonクライアントを設定します。

#APIキーをtoken変数に設定
client = dr.Client(
    token="<上記ページで取得したAPIキー>",
    endpoint=endpoint
)

#正常に終了するとTrueが返されます
client.verify

DataRobotがSQSから監視データを取得できるように、AWSの資格証明を作成します。

CREDS = {
  "name": PROJ_NAME,
  "awsAccessKeyId": "<アクセスキー>",
  "awsSecretAccessKey":"<シークレットキー>",
  "awsSessionToken": "<セッショントークン>", #一時認証ご利用の場合
  "credentialType": "s3"
}
CREDENTIAL = client.post("credentials/", data=CREDS).json()
CREDENTIAL_ID = CREDENTIAL["credentialId"]
CREDENTIAL_ID

学習データセットをDataRobotにアップロード

ドリフト監視のベースラインとなるデータ(学習データセット)をDataRobotにアップロードします。UIから実施する方法はこちらをご覧ください：データのロード

features=["sepal_length", "sepal_width", "petal_length", "petal_width"]
df = pd.read_csv(TRAINING_DATASET, names=["variety"] + features, header=None)
dataset = dr.Dataset.create_from_in_memory_data(df, fname=TRAINING_DATASET, \　max_wait=900)
dataset.id

外部モデルを定義

外部モデル機能を利用して、Amazon SageMakerにデプロイされるモデルの情報をDataRobotに登録します。

class_names = ["setosa", "versicolor", "virginica"]
prediction_type = "Multiclass"

MODEL_PACKAGE = {
    "name": PROJ_NAME,
    "modelDescription": {
        "modelName": "Iris classification model",
        "description": "Irisデータセットの分類",
    },
    "target": {"type": prediction_type, "name": "variety", "classNames": class_names},
    "datasets": {"trainingDataCatalogId": dataset.id},
    "registeredModelName": PROJ_NAME,
}

custom_model = client.post("modelPackages/fromJSON", data=MODEL_PACKAGE).json()
custom_model["id"]

AWS SQSキューを作成

SageMakerの推論サーバーとDataRobotとの間で監視データを受け渡しするためのSQSキューを作成します。

#SQSキュー名を設定
AWS_SQS_QUEUE = f"datarobot-sqs-{PROJ_NAME}"

sqs = boto3.resource("sqs")
sqs_resp = sqs.create_queue(QueueName=AWS_SQS_QUEUE)
sqs_resp.url

予測環境を作成

外部予測環境機能を利用して、モデルがDataRobotの外部にデプロイされていることを示す情報をDataRobotに登録します。監視データの取得先として、上記で作成したSQSキューを指定します。

PRED_ENV = {
  "name": PROJ_NAME,
  "description": "SageMakerとDataRobot連携のデモ",
  "platform": "aws",
  "supportedModelFormats": ["externalModel"],
  "spoolerType": "sqs",
  "credentialId": CREDENTIAL_ID,
  "managementMeta": {
    "additionalMetadata": [
      {
        "key": "AWS_REGION",
        "value": sess.boto_session.region_name
      },
      {
        "key": "MLOPS_SQS_QUEUE_URL",
        "value": sqs_resp.url
      }
    ]
  }
}

prediction_environment = client.post("predictionEnvironments",data=PRED_ENV).json()
prediction_environment["id"]

外部デプロイを作成

DataRobotでデプロイを作成することにより、DataRobot MLOpsによる各種監視機能、ダッシュボード機能の利用が可能になります。ここでは外部モデルのデプロイ機能を利用します。

deployment = dr.Deployment.create_from_registered_model_version(
    model_package_id= custom_model["id"],
    label=PROJ_NAME,
    prediction_environment_id= prediction_environment["id"],
)
deployment.id

#ドリフト監視を有効化。AsyncTimeoutErrorが発生する場合がありますが、次のステップでドリフト設定がTrueになっていることが確認できれば問題ありません。
deployment.update_drift_tracking_settings(
    target_drift_enabled=True,
    feature_drift_enabled=True
)

deployment.get_drift_tracking_settings()

# 出力例：{'target_drift': {'enabled': True}, 'feature_drift': {'enabled': True}}

#精度監視を有効化。同様にAsyncTimeoutErrorが発生する場合がありますが、それでも正常に設定されています。
ASSOC_ID = "association_id"
deployment.update_association_id_settings(
    column_names=[ASSOC_ID],
    max_wait=40,
)

deployment.get_association_id_settings()
#出力例：{'column_names': ['assoc_id'],
# 'required_in_prediction_requests': True,
# 'auto_generate_id': False}

Amazon SageMakerで推論を実施

Amazon SageMakerで推論サーバーのインスタンスをデプロイします。その際に、MLOpsライブラリに関わる環境変数を設定します。

from sagemaker.serializers import CSVSerializer

# MLOpsライブラリに関わる環境変数を設定
env_vars = {
    "MLOPS_DEPLOYMENT_ID": deployment.id,
    "MLOPS_MODEL_ID": custom_model["id"],
    "MLOPS_SQS_QUEUE": sqs_resp.url,
    "MLOPS_ASSOC_ID_COLUMN": ASSOC_ID,
    "MLOPS_DISABLE_FAST_JSON": "1",
    "prediction_type": prediction_type,
    "CLASS_NAMES": json.dumps(class_names),
}

print(env_vars)
predictor = tree.deploy(1, "ml.m4.xlarge", serializer=CSVSerializer(), env=env_vars)

推論を実施

ここでは本ブログに添付のiris_predictions.csvを使用して推論を実施します。

import io

# 予測データを読み込む
pred_df = pd.read_csv("iris_predictions.csv", index=False, header=False)

out = io.StringIO()
pred_df.to_csv(out, header=True, index=False)
results = predictor.predict(out.getvalue() ).decode("utf-8")
print(results)

[参考] 精度監視のための実績値アップロード

予測に対する実績値(今回の文脈では「予測されたIrisの種類に対して、実際のIrisの種類がなんであったか」)をアップロードすることで、モデルの精度を算出し監視することもできます。ここでは、サンプル実績値データ (actuals.csv)をアップロードする方法をご紹介します。

actuals = pd.read_csv("actuals.csv", index=False)
deployment_future = dr.Deployment.get("<depl_id>")
deployment_future.submit_actuals(actuals)

DataRobot MLOpsダッシュボードの確認

全てのプロセスが完了した後、Amazon SageMakerで作成したモデルの結果をDataRobot MLOpsダッシュボードで確認することができます。パフォーマンスとは、モデルが正常に稼働しているかを表す「サービスの正常性」、「データドリフト」と「精度」です。

1. サービスの正常性

サービスの正常性では、予測の利用回数（行数やリクエスト数）や予測にかかった時間をトラッキングし、機械学習モデルが正しく予測が実行されているか監視します（図４）。

2. データドリフト

データドリフトは学習データとその後の予測処理時にモデルに入力されるデータを比較して、各特徴量の分布の違いを意味しており、入力データパイプラインの何らかの変化やトレンド変化の検知に役立つ指標です。DataRobotのデータドリフト監視では、各特徴量ごとに元の学習データからどのように乖離し、時系列でどのように乖離していったかを確認しています（図５）。

左側のグラフは特徴量の有用性（横軸）とドリフト（縦軸）を表しており、有用性が高い特徴量にドリフトが発生していた場合に赤信号が点灯します。有用性が高い特徴量にドリフトが発生するということは、学習時にはない重要なパターンがモデルに入力されたことになるので、予測値を注意深く観察することが必要です。精度への影響が懸念されるときは、モデルを再学習することも視野に入れる必要があります。

右側のグラフは各特徴量について学習データと予測データの分布を確認することができ、データドリフトの発生原因を調査し、対策を講じる必要性について検討するときに役立ちます。

3. 精度

実測値を送信することでモデルの精度をトラッキングすることも可能です。任意に設定した予測IDと予測結果をメトリックとして送信し、予測IDと一致する実測の値が取得できたタイミングでDataRobot  MLOpsにアップロードすることによって、DataRobot内部で精度のトラッキングを行うことができます。

データ・コンテンツのダウンロード

DLはこちらから

• AWS_SageMaker_DataRobot_MLOps.ipynb
• iris.csv
• iris_predictions.csv
• predictor.py
• actuals.csv
• Dockerfile