Cloud Pak for Dataで関数をデプロイする

更新履歴：
（2020/6/9) 関数内のexception時の戻り値にはキー"predictions"を含める必要がある事がわかったため、該当部分を修正。
(2020/6/26) v3.0.1GAで確認した結果に差し替え。モデルと関数の保存時にINPUT_DATA_SCHEMAを追加して、デプロイメントのテスト画面でフォーム形式のデータ入力が可能になるように変更。
(2020/9/8) WML client初期化でのversion番号を自動取得するように変更

---

Cloud Pak for Data (以下CP4D)で、Pythonの関数をデプロイすることが可能です。用途は主にモデル実行の前後処理で、try&exceptのエラー処理も組み込めますし、なにより複数のモデルを呼び出す（使い分ける）ことが可能になります。

CP4D v2.5 製品マニュアルより抜粋
https://www.ibm.com/support/knowledgecenter/ja/SSQNUZ_2.5.0/wsj/analyze-data/ml-deploy-functions_local.html

Python 関数は、モデルのデプロイと同じ方法で Watson Machine Learning にデプロイできます。ご使用のツールとアプリケーションは、Watson Machine Learning Python クライアントまたは REST API を使用して、デプロイ済みのモデルにデータを送信するのと同じ方法で、デプロイ済みの関数にデータを送信できます。関数をデプロイすることで、詳細 (資格情報など) を非表示にしたり、データをモデルに渡す前に前処理したり、エラー処理を実行したり、呼び出しを複数のモデルに組み込んだりできます。これらの機能はすべて、アプリケーションではなくデプロイされた関数内に組み込まれています。

WMLでの関数のデプロイは先行記事がありますが、IBM CloudサービスのWatson Studioでのやり方であり、CP4Dでは少しやり方が異なっていてハマったので、うまく行った例を記事として残します。
シンプルな関数を作ってOnline型でデプロイし、動作を確認してみた結果です。

構成イメージ

Notebookで作成したモデルと関数は、デプロイメントスペースへ格納し、各々をOnline型でデプロイします。関数にはモデルのデプロイメントを呼び出すように設定しておくことで、関数のデプロイメントの呼び出し → モデルのデプロイメントの呼び出しという構成にします。Notebookの最後で、関数のデプロイメントをスコアリング実行して動作を確認します。

Notebookで実際に作ってみる

CP4Dにログインし、分析プロジェクトを開き、新規のNotebookを起動します。確認したCP4Dのバージョンはv3.0.1 GAです。 (2020/6/26変更

準備：モデルの作成とデプロイ

先にモデルを作成してデプロイしておきます。このデプロイメントIDを、関数の中で呼び出すために後ほど使用します。

(1) モデルを作成する

Irisデータを使ったランダムフォレストのモデルを作ります。

# Irisサンプルデータをロード
import pandas as pd
from sklearn.datasets import load_iris
iris = load_iris()
df = pd.DataFrame(iris.data, columns=iris.feature_names)
df['iris_type'] = iris.target_names[iris.target]

# ランダムフォレストでモデルを作成
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier

X = df.drop('iris_type', axis=1)
y = df['iris_type']
X_train, X_test, y_train, y_test = train_test_split(X,y,random_state=0)

clf = RandomForestClassifier(max_depth=2, random_state=0, n_estimators=10)
model = clf.fit(X_train, y_train)

# モデルの精度を確認
from sklearn.metrics import confusion_matrix, accuracy_score
y_test_predicted = model.predict(X_test)    
print("confusion_matrix:")
print(confusion_matrix(y_test,y_test_predicted))
print("accuracy:", accuracy_score(y_test,y_test_predicted))

上記のmodelが学習済モデルです。

(2) モデルを保存しデプロイする

モデルをデプロイメントスペースに保存し、Online型のデプロイメントを作成します。
操作はWML clientを使います。（参考記事)

# WML clientの初期化と認証
from watson_machine_learning_client import WatsonMachineLearningAPIClient
import requests, os
token = os.environ['USER_ACCESS_TOKEN']
url = "https://cp4d.host.name.com"
version = requests.get(os.environ['RUNTIME_ENV_APSX_URL'] + "/diag").text[0:5]
wml_credentials = {
    "token" : token,
    "instance_id" : "openshift",
    "url": url,
    "version": version
}
client = WatsonMachineLearningAPIClient(wml_credentials)

# デプロイメントスペースID一覧の表示
# client.repository.list_spaces()

# デプロイメントスペースへ切り替え (IDは上記のlist_spaces()で調べる)
space_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
client.set.default_space(space_id)

# モデルのメタ情報を記述 (2020/6/26 meta_propsのINPUT_DATA_SCHEMAをlist型に変更。WML client v1.0.95での変更点
model_name = "sample_iris_model"
meta_props={
    client.repository.ModelMetaNames.NAME: model_name,
    client.repository.ModelMetaNames.RUNTIME_UID: "scikit-learn_0.22-py3.6",
    client.repository.ModelMetaNames.TYPE: "scikit-learn_0.22",
    client.repository.ModelMetaNames.INPUT_DATA_SCHEMA: [{
        "id":"iris model",
        "type":"list",
        "fields":[
            {'name': 'sepal length (cm)', 'type': 'double'},
            {'name': 'sepal width (cm)', 'type': 'double'},
            {'name': 'petal length (cm)', 'type': 'double'},
            {'name': 'petal width (cm)', 'type': 'double'}
        ]
    }],
    client.repository.ModelMetaNames.OUTPUT_DATA_SCHEMA: {
        "id":"iris model",
        "fields": [
            {'name': 'iris_type', 'type': 'string','metadata': {'modeling_role': 'prediction'}}
        ]
    }
}

# モデルを保存
model_artifact = client.repository.store_model(model, meta_props=meta_props, training_data=X, training_target=y)
model_id = model_artifact['metadata']['guid']

# Online型でデプロイメントを作成
dep_name = "sample_iris_online"
meta_props = {
    client.deployments.ConfigurationMetaNames.NAME: dep_name,
    client.deployments.ConfigurationMetaNames.ONLINE: {}
}
deployment_details = client.deployments.create(model_id, meta_props=meta_props)
dep_id = deployment_details['metadata']['guid']

(3) モデルの動作確認

作ったモデル(デプロイメント)がちゃんと動くか、一応確認しておきます。WML clientを使って動作確認が可能です。

# サンプルデータを生成しJSON化
scoring_x = pd.DataFrame(
    data = [[5.1,3.5,1.4,0.2]],
    columns=['sepal length (cm)','sepal width (cm)','petal length (cm)','petal width (cm)']
)
values = scoring_x.values.tolist()
fields = scoring_x.columns.values.tolist()
scoring_payload = {client.deployments.ScoringMetaNames.INPUT_DATA: [{'fields': fields, 'values': values}]}

# スコアリング実行
prediction = client.deployments.score(dep_id, scoring_payload)
prediction

以下の結果が返ってくれば、動作確認は完了です。

output

{'predictions': [{'fields': ['prediction', 'probability'],
   'values': [[0, [0.8131726303900102, 0.18682736960998966]]]}]}

関数の作成

作成する関数に対しては、いろいろとお作法があります。

• 2層にネストした関数とする
• 外側の関数では、ライブラリのインポートなどを行う
• 内側の関数名はscoreとし、入力データpayloadを受け取って処理を行い結果をreturnする
• returnする出力データは、JSONシリアル化可能な辞書またはリスト

こちらが実際に作った関数です。WML clientを使ったデプロイメントの実行の仕方は参考記事を参照してください。

関数の作成

# variables for function
func_variables = {
    "url" : url,
    "space_id" : space_id,
    "dep_id" : dep_id,
    "token"  : token
}

# Function for scoring model only
def iris_scoring(func_vars=func_variables):
    from watson_machine_learning_client import WatsonMachineLearningAPIClient
    # 注意点：wml_credentialsはここに定義する
    wml_credentials = {
                   "token" : func_vars['token'],
                   "instance_id": "openshift",
                   "url": func_vars['url'],
                   "version" : "3.0.0"
                  }
    client = WatsonMachineLearningAPIClient(wml_credentials)

    def score(scoring_payload):
        try:
            client.set.default_space(func_vars['space_id'])
            prediction = client.deployments.score(func_vars['dep_id'], scoring_payload)
            return prediction

        # 2020/6/9修正。戻り値にはキー"predictions"を含める必要がある。
        except Exception as e:
            return { "predictions": [{"fields": "error_response", "values": [str(repr(e))]}] }

    return score

関数内で使う変数群は、関数の外で定義しておき関数に渡します。urlはCP4DのURL(FQDNまで。最後のスラッシュ無し)、space_idは先ほど作成したモデルをデプロイしたデプロイメントスペースのID、dep_idはそのモデルのデプロイメントIDです。
関数の中身は、シンプルに先ほどのOnlineデプロイメントを呼び出すのみです。

なおコメントにも書いていますが、WML clientの初期化を外側関数でやること、またそこでつかうwml_credentialsを外側関数の中で定義することがポイントです。関数外で定義したwml_credentialsを引数で受け取るようにするとうまく動きません。ここがIBM CloudのWatson Studioと違う点のようで、とてもハマりました。

作成した関数の動作確認をしておきます。
Notebookで続けて以下のように実行し、関数からデプロイメントの呼び出しが成功するかどうかを確認します。入力データscoring_payloadは先ほど作成したサンプルデータを使います。

関数の動作確認

# 関数を実行(内側関数にscoring_payloadを渡す)
iris_scoring()(scoring_payload)

以下の結果が返ってくればOKです。

output

{'predictions': [{'fields': ['prediction', 'probability'],
   'values': [['setosa', [0.9939393939393939, 0.006060606060606061, 0.0]]]}]}

関数の保存

関数をデプロイメントスペースに保存します。

# メタ情報を作成 (2020/6/26 meta_propsにINPUT_DATA_SCHEMASを追加。modelとは形式が若干違うので注意)
meta_props = {
    client.repository.FunctionMetaNames.NAME: 'iris_scoring_func',
    client.repository.FunctionMetaNames.RUNTIME_UID: "ai-function_0.1-py3.6",
    client.repository.FunctionMetaNames.SPACE_UID: space_id,
    client.repository.FunctionMetaNames.INPUT_DATA_SCHEMAS: [{
        "id":"iris function",
        "fields": [
            {"metadata": {}, "type": "double", "name": "sepal length (cm)", "nullable": False},
            {"metadata": {}, "type": "double", "name": "sepal width (cm)", "nullable": False},
            {"metadata": {}, "type": "double", "name": "petal length (cm)", "nullable": False},
            {"metadata": {}, "type": "double", "name": "petal width (cm)", "nullable": False}
        ]
    }],
    client.repository.FunctionMetaNames.OUTPUT_DATA_SCHEMAS: [{
        "id":"iris function",
        "fields": [
            {"metadata": {'modeling_role': 'prediction'}, "type": "string", "name": "iris_type", "nullable": False}
        ]
    }]
}

# 関数を保存
function_details = client.repository.store_function(meta_props=meta_props, function=iris_scoring)
function_id = function_details['metadata']['guid']

デプロイメントスペースでは、資産の下の「機能」というセクションに登録されている事がわかります。

関数のデプロイ

デプロイメントスペースに保存した関数を、Online型でデプロイします。

# メタ情報を作成
meta_props = {
    client.deployments.ConfigurationMetaNames.NAME: "iris_scoring_online",
    client.deployments.ConfigurationMetaNames.ONLINE: {}
}

# デプロイメントを作成
function_deployment_details = client.deployments.create(function_id, meta_props=meta_props)
func_dep_id = function_deployment_details['metadata']['guid']

これで関数をデプロイできました。実行準備完了です。

関数の動作確認

さっそく動作確認してみます。WML clientを使ってデプロイメントをスコアリング呼び出しします。

prediction = client.deployments.score(func_dep_id, scoring_payload)
prediction

先ほどと同じ、以下の結果が返ってくれば成功です。

output

{'predictions': [{'fields': ['prediction', 'probability'],
   'values': [['setosa', [0.9939393939393939, 0.006060606060606061, 0.0]]]}]}

これで、関数を使ってモデル呼び出しを行うシンプルな方法が確立できました。
後ほどより応用的な複数モデルの呼び出しパターンを投稿する予定です。