Databricksにおけるモデルサービング

[MLflow Model Serving on Databricks \| Databricks on AWS](https://docs.databricks.com/applications/mlflow/model-serving.html) [2021/10/14時点]の翻訳です。

> **プレビュー**
本機能は[パブリックプレビュー](https://docs.databricks.com/release-notes/release-types.html)です。

MLflowのモデルサービングを用いることで、モデルレジストリからモデルを、モデルのバージョン、ステージの基づいて自動でアップデートされるRESTエンドポイントをホストすることができます。

特定の登録モデルに対するモデルサービングを有効化する際、Databricksはモデル固有のクラスターを自動で作成し、非アーカイブの状態のモデル全てをそのクラスター上にデプロイします。エラーが発生した際にはDatabricksをクラスターを再起動し、モデルサービングを無効にした際にはクラスターを停止します。モデルサービングは自動でモデルレジストリと同期し、新たに登録されたモデルバージョンをデプロイします。デプロイされたモデルバージョンは、標準的なREST APIリクエストで問い合わせすることができます。Databricksは標準的な認証方法を用いてリクエストを認証します。

このサービスはプレビューですが、Databricksは低スループット、クリティカルではないアプリケーションでの利用をお勧めします。保証はしませんが、ターゲットのスループットは200qps(秒あたりクエリー数)であり、ターゲットの可用性は99.5%です。さらに、リクエストあたり16MBのペイロードサイズの制限があります。

それぞれのモデルバージョンは[MLflowのモデルデプロイメント](https://www.mlflow.org/docs/latest/models.html#deploy-mlflow-models)を用いてデプロイされ、依存関係によって指定されるConda環境で実行されます。

> **注意**

> - サービングが有効化されている間は、アクティブなモデルバージョンがなくてもクラスターは維持されます。サービングクラスターを停止するには、登録モデルに対するモデルサービングを無効にしてください。
- クラスターはall-purposeクラスターとみなされ、all-purposeワークロードの課金が発生します。
- [Global init scripts](https://docs.databricks.com/clusters/init-scripts.html#init-script-types)はサービングクラスターでは実行されません。

# 要件

- MLflowのモデルサービングはPythonのMLflowモデルで利用できます。全てのモデルの依存関係をconda環境で宣言する必要があります。
- モデルサービングを有効化するには、[クラスター作成権限](https://docs.databricks.com/administration-guide/access-control/cluster-acl.html#cluster-create-permission)を持っている必要があります。

# モデルレジストリからのモデルサービング

Databricksのモデルサービングはモデルレジストリから利用できます。

## モデルサービングの有効化、無効化

[登録モデルページ](https://qiita.com/taka_yayoi/items/e7a4bec6420eb7069995#%E7%99%BB%E9%8C%B2%E3%83%A2%E3%83%87%E3%83%AB%E3%83%9A%E3%83%BC%E3%82%B8)からモデルサービングを有効化できます。

1. **Serving**タブをクリックします。モデルのサービングが有効化されていない場合には、**Enable Serving**ボタンが表示されます。
![](https://docs.databricks.com/_images/enable-serving.png)
1. **Enable Serving**をクリックします。Servingタブが表示され、**Status**がPendingとなります。数分後に**Status**がReadyになります。

モデルサービングを無効化するには**Stop**をクリックします。

## モデルサービングの検証

**Serving**タブから、サーブされるモデルにリクエストを送信したり、レスポンスを参照できます。
![](https://docs.databricks.com/_images/enable-serving.gif)

## モデルバージョンのURI

それぞれのモデルバージョンに1つ以上のユニークなURIが割り当てられます。少なくとも、それぞれのモデルバージョンには以下の構成のURLが割り当てられます。

`<databricks-instance>/model/<registered-model-name>/<model-version>/invocations`

例えば、`iris-classifier`として登録されたモデルのバージョン1を呼び出すには、以下のURIを用います。

`https://<databricks-instance>/model/iris-classifier/1/invocations`

ステージを指定してモデルバージョンを呼び出すこともできます。例えば、バージョン1が**Production**ステージにあるのであれば、以下のURIで呼び出すこともできます。

`https://<databricks-instance>/model/iris-classifier/Production/invocations`

利用できるモデルURIは、サービングページのModel Versionsタブの上に表示されます。

# サーブされるバージョンの管理

全てのアクティブな(Archiveではない)モデルバージョンはデプロイされ、URIを用いて問い合わせを行うことができます。新規モデルバージョンが登録された時に、Databricksは自動でデプロイを行い、アーカイブされた際には古いバージョンを自動で削除します。

> **注意**
登録モデルのデプロイされた全てのバージョンは同じクラスターを共有します。

# モデルアクセス権の管理

モデルのアクセス権はモデルレジストリから継承されます。サービング機能の有効化、無効化には登録されたモデルに対する`Manage`権限が必要となります。読み取り権限を持つユーザーは誰でもデプロイされたバージョンのスコアリングを実行できます。

# デプロイされたモデルバージョンのスコアリング

デプロイされたモデルのスコアリングを実行するには、UIを使うか、モデルURIに対してREST APIのリクエストを送信します。

## UIによるスコアリング

これはモデルをテストするための最も簡単でクイックな方法です。モデルの入力データをJSONフォーマットで挿入し、**Send Request**をクリックします。入力サンプルとともにモデルが記録されている場合(上の図で示されています)には、**Load Example**をクリックして、入力サンプルを呼び出すことができます。

## REST APIリクエストによるスコアリング

[標準的なDatabricksの認証方法](https://docs.databricks.com/dev-tools/api/latest/authentication.html)を用いて、REST API経由でスコアリングのリクエストを送信することができます。以下では、パーソナルアクセストークンを用いた例を示します。

`https://<databricks-instance>/model/iris-classifier/Production/invocations`(ここで`<databricks-instance>`は[お使いのDatabricksインスタンス名](https://docs.databricks.com/workspace/workspace-details.html#workspace-instance-names-urls-and-ids)となります)のような`MODEL_VERSION_URI`と`DATABRICKS_API_TOKEN`と呼ばれるDatabricks REST APIトークンがある場合、サーブされるモデルに対してクエリーをどのように行うのかを示すサンプルスニペットを以下に示します。

### Bash

データフレームの入力を受け付けるモデルに対するクエリーのスニペットです。

```bash:Bash
curl -X POST -u token:$DATABRICKS_API_TOKEN $MODEL_VERSION_URI \
  -H 'Content-Type: application/json' \
  -d '[
    {
      "sepal_length": 5.1,
      "sepal_width": 3.5,
      "petal_length": 1.4,
      "petal_width": 0.2
    }
  ]'
```

テンソルの入力を受け付けるモデルに対するクエリーのスニペットです。テンソルの入力は[TensorFlow Serving’s API docs](https://www.tensorflow.org/tfx/serving/api_rest#request_format_2)で説明されているようにフォーマットされる必要があります。

```bash:Bash
curl -X POST -u token:$DATABRICKS_API_TOKEN $MODEL_VERSION_URI \
   -H 'Content-Type: application/json' \
   -d '{"inputs": [[5.1, 3.5, 1.4, 0.2]]}'
```

### Python

```py:Python
import numpy as np
import pandas as pd
import requests

def create_tf_serving_json(data):
  return {'inputs': {name: data[name].tolist() for name in data.keys()} if isinstance(data, dict) else data.tolist()}

def score_model(model_uri, databricks_token, data):
  headers = {
    "Authorization": f"Bearer {databricks_token}",
    "Content-Type": "application/json",
  }
  data_json = data.to_dict(orient='records') if isinstance(data, pd.DataFrame) else create_tf_serving_json(data)
  response = requests.request(method='POST', headers=headers, url=model_uri, json=data_json)
  if response.status_code != 200:
      raise Exception(f"Request failed with status {response.status_code}, {response.text}")
  return response.json()


# Scoring a model that accepts pandas DataFrames
data =  pd.DataFrame([{
  "sepal_length": 5.1,
  "sepal_width": 3.5,
  "petal_length": 1.4,
  "petal_width": 0.2
}])
score_model(MODEL_VERSION_URI, DATABRICKS_API_TOKEN, data)

# Scoring a model that accepts tensors
data = np.asarray([[5.1, 3.5, 1.4, 0.2]])
score_model(MODEL_VERSION_URI, DATABRICKS_API_TOKEN, data)
```

### PowerBI

以下のステップを踏むことで、Power BIデスクトップでデータセットのスコアリングを行うことができます。

1. スコアリングしたいデータセットを開く
1. Transform Dataに移動
1. 左のパネルを右クリックし、**Create New Query**を選択
1. **View > Advanced Editor**に移動
1. 適切な`DATABRICKS_API_TOKEN`、`MODEL_VERSION_URI `で置換した後で、以下のコードスニペットでクエリーのボディを置換

```
(dataset as table ) as table =>
let
  call_predict = (dataset as table ) as list =>
  let
    apiToken = DATABRICKS_API_TOKEN,
    modelUri = MODEL_VERSION_URI,
    responseList = Json.Document(Web.Contents(modelUri,
      [
        Headers = [
          #"Content-Type" = "application/json",
          #"Authorization" = Text.Format("Bearer #{0}", {apiToken})
        ],
        Content = Json.FromValue(dataset)
      ]
    ))
  in
    responseList,
  predictionList = List.Combine(List.Transform(Table.Split(dataset, 256), (x) => call_predict(x))),
  predictionsTable = Table.FromList(predictionList, (x) => {x}, {"Prediction"}),
  datasetWithPrediction = Table.Join(
    Table.AddIndexColumn(predictionsTable, "index"), "index",
    Table.AddIndexColumn(dataset, "index"), "index")
in
  datasetWithPrediction
```

1. 好きなモデル名でクエリーの名前をつける
1. データセットに対してアドバンスドクエリーエディタを開き、モデル関数を適用

サーバーによって受け付けられる入力データのフォーマットに関しては、[MLflow documentation](https://www.mlflow.org/docs/latest/models.html#deploy-mlflow-models)をご覧ください。

# サーブされたモデルのモニタリング

サービングページでは、サービングクラスターのステータスとそれぞれのモデルバージョンが表示されます。

- サービングクラスターの状態を調査するには、当該モデルの全てのサービングイベントを表示する**Model Events**タブを使用します。
- 個々のモデルバージョンの状態を調査するには、**Model Versions**タブをクリックし、**Logs**あるいは**Version Events**タブをスクロールします。
![](https://docs.databricks.com/_images/serving-tab.png)

# サービングクラスターのカスタマイズ

サービングクラスターをカスタマイズするには、**Serving**タブの**Cluster Settings**タブを使用します。
![](https://docs.databricks.com/_images/cluster-settings.png)

- サービングクラスターのメモリーサイズとコア数を変更するには、希望するクラスター設定を選択するために**Instance Type**ドロップダウンメニューを選択します。**Save**をクリックすると、既存のクラスターは停止され、指定された設定による新規クラスターが作成されます。
- タグを追加するには、**Add Tag**フィールドに名前と値を入力し、**Add**をクリックします。
- 既存のタグを編集、削除するには、**Tags**テーブルの**Actions**カラムのアイコンの一つをクリックします。

# 既知のエラー

`ResolvePackageNotFound: pyspark=3.1.0`

このエラは、モデルが`pyspark`に依存しており、Databricksランタイム 8.xで記録された際に発生します。このエラーに遭遇した場合には、モデルを記録する際に[conda\_envパラメーター](https://www.mlflow.org/docs/latest/projects.html#specifying-an-environment)を用いて、明示的に`pyspark`のバージョンを指定してください。

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)