Legacy MLflow モデルサービングを Databricks

備考

プレビュー

important

このドキュメントは廃止されており、更新されない可能性があります。このコンテンツに記載されている製品、サービス、またはテクノロジはサポートされなくなりました。
この記事のガイダンスは、レガシ MLflow モデルサービング用です。 Databricks では、モデルサービングワークフローをモデルサービングに移行して、拡張モデルエンドポイントのデプロイとスケーラビリティを実現することをお勧めします。詳細については、「 Mosaic AI Model Servingを使用したモデルのデプロイ」を参照してください。

Legacy MLflow モデルサービングを使用すると、 Model Registry as REST エンドポイントから機械学習モデルをホストでき、モデルバージョンとそのステージの可用性に基づいて自動的に更新されます。これは、現在クラシックコンピュートプレーンと呼ばれているもの内で独自のアカウントで実行される単一ノードクラスターを使用します。このコンピュートプレーンには、仮想ネットワークとそれに関連付けられているコンピュートリソース (ノートブックとジョブのクラスター、プロウェアハウスとクラシック SQLウェアハウス、レガシモデルサービングエンドポイントなど) が含まれています。

特定の登録済みモデルに対してモデルサービングを有効にすると、 Databricks によってモデルの一意のクラスターが自動的に作成され、そのクラスターにモデルのアーカイブされていないすべてのバージョンがデプロイされます。 Databricks は、エラーが発生した場合にクラスターを再開し、モデルのモデルサービングを無効にするとクラスターを終了します。モデルサービングは、 Model Registry と自動的に同期し、新しく登録されたモデルバージョンをデプロイします。デプロイされたモデルバージョンは、標準の REST API 要求を使用してクエリを実行できます。 Databricks は、標準認証を使用してモデルへの要求を認証します。

このサービスはプレビュー段階ですが、Databricks では、低スループットで重要でないアプリケーションに使用することをお勧めします。ターゲットスループットは200 qps、ターゲット可用性は99.5%ですが、どちらも保証されていません。さらに、要求ごとに 16 MB のペイロードサイズ制限があります。

各モデルバージョンは、MLflow モデルデプロイを使用してデプロイされ、その依存関係で指定された Conda 環境で実行されます。

注記

クラスターは、アクティブなモデルバージョンが存在しない場合でも、サービスが有効になっている限り維持されます。サービングクラスターを終了するには、登録済みモデルのモデルサービングを無効にします。
クラスターはAll-Purposeクラスターと見なされ、All-Purpose Workload Priceが適用されます。
Global initスクリプトはモデルサービングクラスターでは実行されません。

important

Anaconda Incは、anaconda.org チャンネルの利用規約を更新しました。新しい利用規約に基づいて、Anacondaのパッケージと配布に依存している場合は、商用ライセンスが必要になる場合があります。詳細については、 Anaconda Commercial Edition FAQ を参照してください。 Anacondaチャンネルの使用は、そのサービス利用規約に準拠します。

MLflowv1.18 より前 (Databricks Runtime 8.3ML 以前) で記録されたモデルは、デフォルトによって condadefaults チャンネル (https://repo.anaconda.com/pkgs/) で記録されていました依存関係として。このライセンス変更に伴い、 Databricks は MLflow v1.18 以降を使用してログインしたモデルでの defaults チャンネルの使用を停止しました。ログに記録されたデフォルトチャンネルは、コミュニティが管理する https://conda-forge.org/ を指す conda-forgeになりました。

MLflow v1.18より前のモデルをログに記録し、そのモデルのconda環境からdefaultsチャンネルを除外した場合、そのモデルは意図していないdefaultsチャンネルに依存している可能性があります。モデルにこの依存関係があるかどうかを手動で確認するには、記録済みモデルにパッケージ化されているconda.yaml ファイル内のchannel値を調べます。たとえば、defaults チャンネルの依存関係を持つモデルのconda.yamlは、次のようになります。

YAML
channels:
- defaults
dependencies:
- python=3.8.8
- pip
- pip:
    - mlflow
    - scikit-learn==0.23.2
    - cloudpickle==1.6.0
      name: mlflow-env

Databricks は、Anaconda との関係の下でモデルと対話するための Anaconda リポジトリの使用が許可されているかどうかを判断できないため、Databricks は顧客に変更を強制しません。 Databricks の使用による Anaconda.com リポジトリの使用が Anaconda の条件で許可されている場合は、何もする必要はありません。

モデルの環境で使用するチャンネルを変更したい場合は、新しい conda.yamlでモデルをモデルレジストリに再登録することができます。これを行うには、log_model()の conda_env パラメーターでチャンネルを指定します。

log_model() APIの詳細については、使用しているモデルフレーバーのMLflowドキュメンテーション(log_model for a scikit-learnなど)を参照してください。

conda.yamlファイルの詳細については、 MLflow のドキュメントを参照してください。

必要条件

Legacy MLflow モデルサービングは Python MLflow モデルでご利用いただけます。すべてのモデルの依存関係は、conda環境で宣言する必要があります。記録済みモデルの依存関係を参照してください。
モデルサービングを有効にするには、クラスター作成権限が必要です。

モデルサービング from Model Registry

モデルサービングはDatabricks からModel Registry で入手できます。

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

モデルの配信は、登録済みのモデルページから有効にします。

「Serving 」タブをクリックします。モデルでサービングが有効になっていない場合は、[ サービングを有効にする] ボタンが表示されます。
「 Enable Serving」 をクリックします。 [サービス] タブが表示され、[ ステータス ] が [保留中] と表示されます。数分後、[ ステータス ] が [準備完了] に変わります。

モデルの配信を無効にするには、[ 停止] をクリックします。

Validate モデルサービング

[ Serving ] タブでは、提供されたモデルにリクエストを送信し、レスポンスを表示できます。

サービングを有効にする

モデルバージョン URI

デプロイされた各モデルバージョンには、1 つまたは複数の一意の URI が割り当てられます。少なくとも、各モデルバージョンには、次のように構築されたURIが割り当てられます。

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

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

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

モデルバージョンをステージごとに呼び出すこともできます。たとえば、バージョン 1 が 本番運用 ステージにある場合は、次の URI を使用してスコアリングすることもできます。

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

使用可能なモデル URI のリストは、配信ページの [モデルバージョン] タブの上部に表示されます。

提供済みバージョンの管理

すべてのアクティブな (アーカイブされていない) モデルバージョンがデプロイされ、URI を使用してクエリを実行できます。 Databricks は、新しいモデルバージョンが登録されると自動的にデプロイされ、古いバージョンがアーカイブされると自動的に削除されます。

注記

登録済みモデルのデプロイされたすべてのバージョンは、同じクラスターを共有します。

モデルのアクセス権を管理する

モデルのアクセス権は、 Model Registryから継承されます。サービング機能を有効または無効にするには、登録済みモデルに対する「管理」権限が必要です。読み取り権限を持つユーザーは、デプロイされた任意のバージョンをスコアリングできます。

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

デプロイされたモデルをスコアリングするには、UI を使用するか、モデル URI に REST API 要求を送信します。

UIによるスコア

これは、モデルをテストする最も簡単で最速の方法です。モデル入力データをJSON形式で挿入し、[ リクエストを送信 ]をクリックできます。モデルが入力例でログ記録されている場合 (上の図を参照)、 Load Example をクリックして入力例を読み込みます。

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

標準の Databricks 認証を使用して、REST API を介してスコアリング要求を送信できます。次の例は、MLflow 1.x でのパーソナルアクセストークンを使用した認証を示しています。

注記

セキュリティのベストプラクティスとして、自動化されたツール、システム、スクリプト、アプリで認証する場合、Databricks では、ワークスペースユーザーではなく、サービスプリンシパルに属する個人用アクセストークンを使用することをお勧めします。サービスプリンシパルのトークンを作成するには、「サービスプリンシパルのトークンの管理」を参照してください。

https://<databricks-instance>/model/iris-classifier/Production/invocations のようなMODEL_VERSION_URI (<databricks-instance> は Databricks インスタンスの名前) と Databricks REST API トークン DATABRICKS_API_TOKENを指定すると、次の例は、提供されるモデルをクエリする方法を示しています。

次の例は、MLflow 1.x で作成されたモデルのスコアリング形式を反映しています。 MLflow 2.0 を使用する場合は、要求ペイロードの形式を更新する必要があります。

Bash
Python
PowerBI

Snippet to query a model accepting dataframe inputs.

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
    }
  ]'

Snippet to query a model accepting tensor inputs. Tensor inputs should be formatted as described in TensorFlow Serving’s API docs.

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
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)

You can score a dataset in Power BI Desktop using the following steps:

Open dataset you want to score.
Go to Transform Data.
Right-click in the left panel and select Create New Query.
Go to View > Advanced Editor.

Replace the query body with the code snippet below, after filling in an appropriate DATABRICKS_API_TOKEN and 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

Name the query with your desired model name.
Open the advanced query editor for your dataset and apply the model function.

モニター提供モデル

サービングページには、サービングクラスターのステータスインジケータと個々のモデルバージョンが表示されます。

サービングクラスターの状態を調べるには、このモデルのすべてのサービングイベントの一覧を表示する [モデルイベント ] タブを使用します。
1 つのモデルバージョンの状態を調べるには、「 モデルバージョン 」タブをクリックし、スクロールして 「ログ 」タブまたは 「バージョンイベント 」タブを表示します。

[サービング] タブ

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

サービングクラスターをカスタマイズするには、[ サービング ] タブの [クラスター設定 ] タブを使用します。

クラスター設定

サービングクラスターのメモリサイズとコア数を変更するには、[ Instance Type ] ドロップダウンメニューを使用して、目的のクラスター設定を選択します。 [保存 ] をクリックすると、既存のクラスターが終了し、指定した設定で新しいクラスターが作成されます。
タグを追加するには、[ タグの追加 ] フィールドに名前と値を入力し、[ 追加 ] をクリックします。
既存のタグを編集または削除するには、「 タグ」(Tags ) テーブルの 「アクション」(Actions ) コラムにあるアイコンの 1 つをクリックします。

既知のエラー

ResolvePackageNotFound: pyspark=3.1.0

このエラーは、モデルが pyspark に依存し、Databricks Runtime 8.x を使用してログに記録されている場合に発生する可能性があります。このエラーが表示された場合は、モデルのログ記録時にpysparkバージョンを明示的に指定します。 conda_env パラメーター。

Unrecognized content type parameters: format

このエラーは、新しい MLflow 2.0 スコアリングプロトコル形式の結果として発生する可能性があります。このエラーが表示される場合は、古いスコアリング要求形式を使用している可能性があります。このエラーを解決するには、次の操作を行います。

スコアリング要求の形式を最新のプロトコルに更新します。

注記

次の例は、MLflow 2.0 で導入されたスコアリング形式を反映しています。 MLflow 1.x を使用する場合は、 log_model() API 呼び出しを変更して、目的の MLflow バージョンの依存関係を extra_pip_requirements パラメーターに含めることができます。これにより、適切なスコアリング形式が使用されることが保証されます。

Python
    mlflow.<flavor>.log_model(..., extra_pip_requirements=["mlflow==1.*"])

Bash
Python
PowerBI

Query a model accepting pandas dataframe inputs.

Bash
curl -X POST -u token:$DATABRICKS_API_TOKEN $MODEL_VERSION_URI \
  -H 'Content-Type: application/json' \
  -d '{
      "dataframe_records": [{"sepal_length (cm)": 5.1, "sepal_width (cm)": 3.5, "petal_length (cm)": 1.4, "petal_width": 0.2},
                            {"sepal_length (cm)": 4.2, "sepal_width (cm)": 5.0, "petal_length (cm)": 0.8, "petal_width": 0.5}]
      }'

Query a model accepting tensor inputs. Tensor inputs should be formatted as described in TensorFlow Serving’s API docs.

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
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_dict = {'dataframe_split': data.to_dict(orient='split')} if isinstance(data, pd.DataFrame) else create_tf_serving_json(data)
  data_json = json.dumps(data_dict)
  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)

You can score a dataset in Power BI Desktop using the following steps:

Open dataset you want to score.
Go to Transform Data.
Right-click in the left panel and select Create New Query.
Go to View > Advanced Editor.

Replace the query body with the code snippet below, after filling in an appropriate DATABRICKS_API_TOKEN and 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

Name the query with your desired model name.
Open the advanced query editor for your dataset and apply the model function.

スコアリング要求で MLflow クライアント ( mlflow.pyfunc.spark_udf()など) を使用している場合は、MLflow クライアントをバージョン 2.0 以降にアップグレードして最新の形式を使用します。 MLflow 2.0 で更新された MLflow モデルスコアリングプロトコルの詳細については、こちらを参照してください。

サーバーが受け入れる入力データ形式Pandas (分割指向形式など) の詳細についてはMLflow の資料を参照してください。

必要条件​

モデルサービング from Model Registry​

モデルサービングの有効化と無効化​

Validate モデルサービング​

モデル バージョン URI​

提供済みバージョンの管理​

モデルのアクセス権を管理する​

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

UIによるスコア​

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

モニター提供モデル​

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

既知のエラー​