Microsoft Fabric での PREDICT を使用した機械学習モデルのスコアリング

Microsoft Fabric を使用すると、スケーラブルな PREDICT 関数を使用して機械学習モデルを運用化できます。この関数は、任意のコンピューティングエンジンでのバッチスコアリングをサポートしています。バッチ予測は、Microsoft Fabric ノートブックまたは特定の ML モデルの項目ページから直接生成できます。

この記事では、自分でコードを記述するか、ガイド付き UI エクスペリエンスを使用してバッチスコアリングを処理して、PREDICT を適用する方法について説明します。

前提条件

Microsoft Fabric サブスクリプションを取得します。または、無料の Microsoft Fabric 試用版にサインアップします。
Microsoft Fabric にサインインします。
ホームページの左下にあるエクスペリエンススイッチャーを使用して Fabric に切り替えます。

制限事項

PREDICT 関数は現在、次の ML モデルフレーバーのみをサポートしています。
- CatBoost
- Keras
- LightGBM
- ONNX
- Prophet
- PyTorch
- Sklearn
- Spark
- Statsmodels
- TensorFlow
- XGBoost
PREDICT では、署名が設定された ML モデルを MLflow 形式で保存する必要があります。
PREDICT では、 マルチテンソル入力または出力を持つ ML モデルはサポートされていません。

ノートブックから PREDICT を呼び出す

PREDICT では、Microsoft Fabric レジストリで MLflow パッケージモデルがサポートされています。既にトレーニングしてワークスペースに登録した ML モデルが存在する場合は、手順 2 までスキップできます。そうでない場合は、手順 1 でサンプルのロジスティック回帰モデルのトレーニングを導くサンプルコードが提供されます。このモデルを使用して、プロシージャの最後にバッチ予測を生成します。

ML モデルをトレーニングして、それを MLflow に登録します。次のコードサンプルでは、MLflow API を使用して機械学習実験を作成し、scikit-learn ロジスティック回帰モデルの MLflow 実行を開始します。その後、モデルバージョンが Microsoft Fabric レジストリに格納され、登録されます。モデルのトレーニングと独自の実験の追跡の詳細については、 scikit-learn を使用して ML モデルをトレーニングする方法を参照してください。

import mlflow
import numpy as np 
from sklearn.linear_model import LogisticRegression 
from sklearn.datasets import load_diabetes
from mlflow.models.signature import infer_signature 

mlflow.set_experiment("diabetes-demo")
with mlflow.start_run() as run:
    lr = LogisticRegression()
    data = load_diabetes(as_frame=True)
    lr.fit(data.data, data.target) 
    signature = infer_signature(data.data, data.target) 

    mlflow.sklearn.log_model(
        lr,
        "diabetes-model",
        signature=signature,
        registered_model_name="diabetes-model"
    )

テストデータを Spark DataFrame として読み込みます。 前の手順でトレーニングした ML モデルを使用してバッチ予測を生成するには、Spark DataFrame 形式のテストデータが必要です。次のコードでは、test 変数の値を独自のデータに置き換えます。
```
# You can substitute "test" below with your own data
test = spark.createDataFrame(data.frame.drop(['target'], axis=1))
```
推論のために ML モデルを読み込む MLFlowTransformer オブジェクトを作成します。 バッチ予測を生成する MLFlowTransformer オブジェクトを作成するには、次のアクションを実行します。
- 必要な test DataFrame 列をモデル入力として指定します (この場合は、そのすべて)。
- 新しい出力列の名前 (この場合は predictions) を選択します。
- これらの予測を生成するための正しいモデル名とモデルバージョンを指定します。
独自の ML モデルを使用している場合は、入力列、出力列名、モデル名、およびモデルバージョンの値を置き換えます。
```
from synapse.ml.predict import MLFlowTransformer

# You can substitute values below for your own input columns,
# output column name, model name, and model version
model = MLFlowTransformer(
    inputCols=test.columns,
    outputCol='predictions',
    modelName='diabetes-model',
    modelVersion=1
)
```
PREDICT 関数を使用して予測を生成します。 PREDICT 関数を呼び出すには、Transformer API、Spark SQL API、または PySpark ユーザー定義関数 (UDF) を使用します。以降のセクションでは、PREDICT 関数を呼び出すためのさまざまなメソッドを使用して、前の手順で定義したテストデータと ML モデルを使用してバッチ予測を生成する方法を示します。

Transformer API を使用して PREDICTを実行する

このコードでは、Transformer API を使用して PREDICT 関数を呼び出します。独自の ML モデルを使用している場合は、モデルとテストデータの値を置き換えます。

# You can substitute "model" and "test" below with values  
# for your own model and test data 
model.transform(test).show()

Spark SQL API を使用した PREDICT

このコードでは、Spark SQL API を使用して PREDICT 関数を呼び出します。独自の ML モデルを使用する場合は、 model_name、 model_version、および features の値をモデル名、モデルバージョン、および特徴列に置き換えます。

注記

Spark SQL API を使用して予測を生成する場合でも、手順 3 に示すように、 MLFlowTransformer オブジェクトを作成する必要があります。

from pyspark.ml.feature import SQLTransformer 

# You can substitute "model_name," "model_version," and "features" 
# with values for your own model name, model version, and feature columns
model_name = 'diabetes-model'
model_version = 1
features = test.columns

sqlt = SQLTransformer().setStatement( 
    f"SELECT PREDICT('{model_name}/{model_version}', {','.join(features)}) as predictions FROM __THIS__")

# You can substitute "test" below with your own test data
sqlt.transform(test).show()

ユーザー定義関数を使用した PREDICT

このコードでは、PySpark UDF を使用して PREDICT 関数を呼び出します。独自の ML モデルを使用する場合は、モデルと特徴の値を置き換えます。

from pyspark.sql.functions import col, pandas_udf, udf, lit

# You can substitute "model" and "features" below with your own values
my_udf = model.to_udf()
features = test.columns

test.withColumn("PREDICT", my_udf(*[col(f) for f in features])).show()

ML モデルの項目ページから PREDICT コードを生成する

任意の ML モデルの項目ページから、PREDICT 関数を使用して、特定のモデルバージョンのバッチ予測生成を開始するには、次のいずれかのオプションを選択できます。

コードテンプレートをノートブックにコピーし、パラメーターを自分でカスタマイズします。
ガイド付き UI エクスペリエンスを使用して PREDICT コードを生成します。

ガイド付き UI エクスペリエンスを使用する

ガイド付き UI エクスペリエンスでは、以下の手順が示されます。

スコアリングのソースデータを選択します。
データを ML モデルの入力に正しくマップします。
モデルの出力先を指定します。
PREDICT を使用して予測結果を生成して格納するノートブックを作成します。

ガイド付きエクスペリエンスを使用するには、

特定の ML モデルバージョンの項目ページに移動します。
[適用するバージョン] ドロップダウンから [ウィザードでこのモデルを適用する] を選択します。

[入力テーブルの選択] ステップで [ML モデル予測の適用] ウィンドウが開きます。
現在のワークスペースの Lakehouses から入力テーブルを選択します。
[次へ] を選択して、[入力列のマッピング] ステップに進みます。
ソーステーブルの列名を、ML モデルのシグネチャから取得されたモデルの入力フィールドにマッピングします。モデルのすべての必須フィールドに入力列を指定する必要があります。さらに、ソース列のデータ型は、モデルの予想されるデータ型と一致している必要があります。

ヒント

入力テーブルの列の名前が ML モデル署名に記録されている列名と一致する場合、ウィザードはこのマッピングを事前設定します。
[次へ] を選択して、[出力テーブルの作成] ステップに移動します。
現在のワークスペースの選択したレイクハウス内にある、新しいテーブルの名前を指定します。この出力テーブルには、ML モデルの入力値が格納され、そのテーブルに予測値が追加されます。既定では、出力テーブルは入力テーブルと同じレイクハウスに作成されます。宛先のレイクハウスは変更できます。
[次へ] を選択して、[出力列のマッピング] ステップに進みます。
指定したテキストフィールドを使用して、ML モデルの予測を格納する出力テーブルの列に名前を付けます。
[次へ] を選択して、[ノートブックを設定する] ステップに進みます。
生成された PREDICT コードを実行する新しいノートブックの名前を指定します。ウィザードには、この手順で生成されたコードのプレビューが表示されます。必要に応じて、コードをクリップボードにコピーし、既存のノートブックに貼り付けることができます。
[次へ] を選択して、[確認と完了] ステップに進みます。
概要ページの詳細を確認し、[ノートブックの作成] を選択して、生成されたコードを含む新しいノートブックをワークスペースに追加します。そのノートブックに直接移動されます。ここで、コードを実行して予測を生成し、格納できます。

カスタマイズ可能なコードテンプレートを使用する

バッチ予測を生成するためにコードテンプレートを使用するには:

特定の ML モデルバージョンの項目ページに移動します。
[適用するバージョン] ドロップダウンから [適用するコードのコピー] を選択します。選択すると、カスタマイズ可能なコードテンプレートがコピーされます。

このコードテンプレートをノートブックに貼り付けて、ML モデルでバッチ予測を生成できます。コードテンプレートを正常に実行するには、次の値を手動で置き換えます。

<INPUT_TABLE>: ML モデルへの入力を提供するテーブルのファイルパス。
<INPUT_COLS>: ML モデルにフィードする入力テーブルの列名の配列。
<OUTPUT_COLS>: 予測を格納する出力テーブル内の新しい列の名前。
<MODEL_NAME>: 予測の生成に使用する ML モデルの名前。
<MODEL_VERSION>: 予測の生成に使用する ML モデルのバージョン。
<OUTPUT_TABLE>: 予測を格納するテーブルのファイルパス。

import mlflow 
from synapse.ml.predict import MLFlowTransformer 
 
df = spark.read.format("delta").load( 
    <INPUT_TABLE> # Your input table filepath here
) 
 
model = MLFlowTransformer( 
    inputCols=<INPUT_COLS>, # Your input columns here
    outputCol=<OUTPUT_COLS>, # Your new column name here
    modelName=<MODEL_NAME>, # Your ML model name here
    modelVersion=<MODEL_VERSION> # Your ML model version here
) 
df = model.transform(df) 
 
df.write.format('delta').mode("overwrite").save( 
    <OUTPUT_TABLE> # Your output table filepath here
)

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-03-01