Azure Databricks を使用して機械学習モデルを管理する

Azure Databricks を使用した機械学習モデルのトレーニングには、データ処理、モデルトレーニング、デプロイのためのコラボレーション環境を提供する統合された分析プラットフォームの活用が含まれます。 Azure Databricks は MLflow と統合して、実験の追跡やモデルの提供など、機械学習のライフサイクルを管理します。

この演習の所要時間は約 20 分です。

開始する前に

管理レベルのアクセス権を持つ Azure サブスクリプションが必要です。

Azure Databricks ワークスペースをプロビジョニングする

ヒント: 既に Azure Databricks ワークスペースがある場合は、この手順をスキップして、既存のワークスペースを使用できます。

この演習には、新しい Azure Databricks ワークスペースをプロビジョニングするスクリプトが含まれています。このスクリプトは、この演習で必要なコンピューティングコアに対する十分なクォータが Azure サブスクリプションにあるリージョンに、Premium レベルの Azure Databricks ワークスペースリソースを作成しようとします。また、使用するユーザーアカウントのサブスクリプションに、Azure Databricks ワークスペースリソースを作成するための十分なアクセス許可があることを前提としています。十分なクォータやアクセス許可がないためにスクリプトが失敗した場合は、Azure portal で、Azure Databricks ワークスペースを対話形式で作成してみてください。

Web ブラウザーで、https://portal.azure.com の Azure portal にサインインします。
ページ上部の検索バーの右側にある [>_] ボタンを使用して、Azure portal に新しい Cloud Shell を作成します。メッセージが表示されたら、PowerShell 環境を選んで、ストレージを作成します。次に示すように、Azure portal の下部にあるペインに、Cloud Shell のコマンドラインインターフェイスが表示されます。

注: 前に Bash 環境を使ってクラウドシェルを作成している場合は、そのクラウドシェルペインの左上にあるドロップダウンメニューを使って、PowerShell に変更します。
ペインの上部にある区分線をドラッグして Cloud Shell のサイズを変更したり、ペインの右上にある — 、 ◻ 、X アイコンを使用して、ペインを最小化または最大化したり、閉じたりすることができます。 Azure Cloud Shell の使い方について詳しくは、Azure Cloud Shell のドキュメントをご覧ください。
PowerShell のペインで、次のコマンドを入力して、リポジトリを複製します。
```
 rm -r mslearn-databricks -f
 git clone https://github.com/MicrosoftLearning/mslearn-databricks
```
リポジトリをクローンした後、次のコマンドを入力して setup.ps1 スクリプトを実行します。これにより、使用可能なリージョンに Azure Databricks ワークスペースがプロビジョニングされます。
```
 ./mslearn-databricks/setup.ps1
```
メッセージが表示された場合は、使用するサブスクリプションを選択します (これは、複数の Azure サブスクリプションへのアクセス権を持っている場合にのみ行います)。
スクリプトの完了まで待ちます。通常、約 5 分かかりますが、さらに時間がかかる場合もあります。待っている間、Azure Databricks ドキュメントの記事「Databricks Machine Learning とは」を確認してください。

クラスターの作成

Azure Databricks は、Apache Spark “クラスター” を使用して複数のノードでデータを並列に処理する分散処理プラットフォームです。** 各クラスターは、作業を調整するドライバーノードと、処理タスクを実行するワーカーノードで構成されています。この演習では、ラボ環境で使用されるコンピューティングリソース (リソースが制約される場合がある) を最小限に抑えるために、単一ノード クラスターを作成します。運用環境では、通常、複数のワーカーノードを含むクラスターを作成します。

ヒント: Azure Databricks ワークスペースに 13.3 LTS ML 以降のランタイムバージョンを備えたクラスターが既にある場合は、この手順をスキップし、そのクラスターを使用してこの演習を完了できます。

Azure portal で、スクリプトによって作成された msl-xxxxxxx リソースグループ (または既存の Azure Databricks ワークスペースを含むリソースグループ) に移動します
Azure Databricks Service リソース (セットアップスクリプトを使って作成した場合は、databricks-xxxxxxx という名前) を選択します。
Azure Databricks ワークスペースの [概要] ページで、[ワークスペースの起動] ボタンを使用して、新しいブラウザータブで Azure Databricks ワークスペースを開きます。サインインを求められた場合はサインインします。

ヒント: Databricks ワークスペースポータルを使用すると、さまざまなヒントと通知が表示される場合があります。これらは無視し、指示に従ってこの演習のタスクを完了してください。
左側のサイドバーで、[(+) 新規] タスクを選択し、[クラスター] を選択します。
[新しいクラスター] ページで、次の設定を使用して新しいクラスターを作成します。
- クラスター名: “ユーザー名の” クラスター (既定のクラスター名)**
- ポリシー:Unrestricted
- クラスターモード: 単一ノード
- アクセスモード: 単一ユーザー (自分のユーザーアカウントを選択)
- Databricks Runtime のバージョン: “以下に該当する最新の非ベータ版ランタイム (標準ランタイムバージョンではない*) の ML エディションを選択します。”
  - “GPU を使用しない”
  - Scala > 2.11 を含める
  - “3.4 以上の Spark を含む”**
- Photon Acceleration を使用する: オフにする
- ノードタイプ: Standard_D4ds_v5
- 非アクティブ状態が ** 20 ** 分間続いた後終了する
クラスターが作成されるまで待ちます。これには 1、2 分かかることがあります。

注: クラスターの起動に失敗した場合、Azure Databricks ワークスペースがプロビジョニングされているリージョンでサブスクリプションのクォータが不足していることがあります。詳細については、「CPU コアの制限によってクラスターを作成できない」を参照してください。その場合は、ワークスペースを削除し、別のリージョンに新しいワークスペースを作成してみてください。次のように、セットアップスクリプトのパラメーターとしてリージョンを指定できます: ./mslearn-databricks/setup.ps1 eastus

ノートブックを作成する

Spark MLLib ライブラリを使って機械学習モデルをトレーニングするコードを実行するので、最初の手順ではワークスペースに新しいノートブックを作成します。

サイドバーで [(+) 新規] タスクを使用して、Notebook を作成します。
既定のノートブック名 (無題のノートブック [日付]) を “機械学習” に変更し、[接続] ドロップダウンリストでクラスターを選択します (まだ選択されていない場合)。クラスターが実行されていない場合は、起動に 1 分ほどかかる場合があります。

データの取り込みと準備

この演習のシナリオは、南極でのペンギンの観察に基づいており、機械学習モデルをトレーニングして、観察されたペンギンの位置と体の測定値に基づいて種類を予測することを目標としています。

[引用]: この演習で使用するペンギンのデータセットは、Dr. Kristen Gorman と、Long Term Ecological Research Network のメンバーである Palmer Station, Antarctica LTER によって収集されて使用できるようにされているデータのサブセットです。

ノートブックの最初のセル内に次のコードを入力します。これは “シェル” コマンドを使用して、GitHub のペンギンデータをクラスターで使用されるファイルシステムの中にダウンロードします。**
```
 %sh
 rm -r /dbfs/ml_lab
 mkdir /dbfs/ml_lab
 wget -O /dbfs/ml_lab/penguins.csv https://raw.githubusercontent.com/MicrosoftLearning/mslearn-databricks/main/data/penguins.csv
```
そのセルの左側にある [▸ セルの実行] メニューオプションを使用して実行します。その後、コードによって実行される Spark ジョブが完了するまで待ちます。

次に、機械学習用のデータを準備します。既存のコードセルの下で、 [+] アイコンを使用して新しいコードセルを追加します。次に、新しいセルに次のコードを入力して実行します。

不完全な行を削除します
適切なデータ型を適用します
データのランダムなサンプルを表示します
データを 2 つのデータセットに分割します。1 つはトレーニング用、もう 1 つはテスト用です。

from pyspark.sql.types import *
from pyspark.sql.functions import *
   
data = spark.read.format("csv").option("header", "true").load("/hyperopt_lab/penguins.csv")
data = data.dropna().select(col("Island").astype("string"),
                          col("CulmenLength").astype("float"),
                          col("CulmenDepth").astype("float"),
                          col("FlipperLength").astype("float"),
                          col("BodyMass").astype("float"),
                          col("Species").astype("int")
                          )
display(data.sample(0.2))
   
splits = data.randomSplit([0.7, 0.3])
train = splits[0]
test = splits[1]
print ("Training Rows:", train.count(), " Testing Rows:", test.count())

パイプラインを実行してデータを前処理し、ML モデルをトレーニングする

モデルをトレーニングする前に、特徴エンジニアリングの手順を実行して、アルゴリズムをデータに合わせる必要があります。いくつかのテストデータでモデルを使用して予測を生成するには、同じ特徴エンジニアリングの手順をテストデータに適用する必要があります。モデルをより効率的に構築し使用するには、データ準備に使用されるトランスフォーマーと、そのトレーニングに使用されるモデルを “パイプライン” にカプセル化します。

次のコードを使用して、データ準備とモデルのトレーニング手順をカプセル化するパイプラインを作成します。

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler, MinMaxScaler
from pyspark.ml.classification import LogisticRegression
   
catFeature = "Island"
numFeatures = ["CulmenLength", "CulmenDepth", "FlipperLength", "BodyMass"]
   
# Define the feature engineering and model training algorithm steps
catIndexer = StringIndexer(inputCol=catFeature, outputCol=catFeature + "Idx")
numVector = VectorAssembler(inputCols=numFeatures, outputCol="numericFeatures")
numScaler = MinMaxScaler(inputCol = numVector.getOutputCol(), outputCol="normalizedFeatures")
featureVector = VectorAssembler(inputCols=["IslandIdx", "normalizedFeatures"], outputCol="Features")
algo = LogisticRegression(labelCol="Species", featuresCol="Features", maxIter=10, regParam=0.3)
   
# Chain the steps as stages in a pipeline
pipeline = Pipeline(stages=[catIndexer, numVector, numScaler, featureVector, algo])
   
# Use the pipeline to prepare data and fit the model algorithm
model = pipeline.fit(train)
print ("Model trained!")

特徴エンジニアリングの手順は、パイプラインによってトレーニングされたモデルにカプセル化されているため、自分で各変換を適用することなく、テストデータでモデルを使用することができます (変換はモデルによって自動的に適用されます)。

次のコードを使用して、パイプラインをテストデータに適用して、モデルを評価します。

prediction = model.transform(test)
predicted = prediction.select("Features", "probability", col("prediction").astype("Int"), col("Species").alias("trueLabel"))
display(predicted)

# Generate evaluation metrics
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
   
evaluator = MulticlassClassificationEvaluator(labelCol="Species", predictionCol="prediction")
   
# Simple accuracy
accuracy = evaluator.evaluate(prediction, {evaluator.metricName:"accuracy"})
print("Accuracy:", accuracy)
   
# Class metrics
labels = [0,1,2]
print("\nIndividual class metrics:")
for label in sorted(labels):
    print ("Class %s" % (label))
   
    # Precision
    precision = evaluator.evaluate(prediction, {evaluator.metricLabel:label,
                                                    evaluator.metricName:"precisionByLabel"})
    print("\tPrecision:", precision)
   
    # Recall
    recall = evaluator.evaluate(prediction, {evaluator.metricLabel:label,
                                             evaluator.metricName:"recallByLabel"})
    print("\tRecall:", recall)
   
    # F1 score
    f1 = evaluator.evaluate(prediction, {evaluator.metricLabel:label,
                                         evaluator.metricName:"fMeasureByLabel"})
    print("\tF1 Score:", f1)
   
# Weighed (overall) metrics
overallPrecision = evaluator.evaluate(prediction, {evaluator.metricName:"weightedPrecision"})
print("Overall Precision:", overallPrecision)
overallRecall = evaluator.evaluate(prediction, {evaluator.metricName:"weightedRecall"})
print("Overall Recall:", overallRecall)
overallF1 = evaluator.evaluate(prediction, {evaluator.metricName:"weightedFMeasure"})
print("Overall F1 Score:", overallF1)

モデルを登録してデプロイする

パイプラインの実行時に、各実験の実行でトレーニングされたモデルを既にログしています。モデルを “登録” してデプロイし、クライアントアプリケーションに提供することもできます。**

注:モデル提供は Azure Databricks Premium ワークスペースでのみサポートされ、特定のリージョンに限定されています。

左側のペインで [実験] を選択します。
ノートブックの名前で生成された実験を選択し、最新の実験の実行の詳細ページを表示します。
[モデルの登録] ボタンを使用して、その実験で記録されたモデルを登録し、プロンプトが表示されたら、Penguin Predictor という名前の新しいモデルを作成します。
モデルが登録されたら、[モデル] ページ (左側のナビゲーションバー) を表示し、Penguin Predictor モデルを選択します。
Penguin Predictor モデルのページで、[推論にモデルを使用する] ボタンを使用して、次の設定で新しいリアルタイムエンドポイントを作成します。
- モデル:Penguin Predictor
- モデルのバージョン: 1
- エンドポイント: predict-penguin
- コンピューティングサイズ: Small
サービスエンドポイントは、新しいクラスター内でホストされます。エンドポイントが作成されるまで数分かかる場合があります。
エンドポイントが作成されたら、右上にある [エンドポイントのクエリ] ボタンを使用して、エンドポイントをテストできるインターフェイスを開きます。次に、テストインターフェイスの [ブラウザー] タブで、次の JSON 要求を入力し、[要求の送信] ボタンを使用して、エンドポイントを呼び出し、予測を生成します。
```
 {
   "dataframe_records": [
   {
      "Island": "Biscoe",
      "CulmenLength": 48.7,
      "CulmenDepth": 14.1,
      "FlipperLength": 210,
      "BodyMass": 4450
   }
   ]
 }
```
ペンギンの特徴にいくつかの異なる値を試してみて、返される結果を観察します。その後、テストインターフェイスを閉じます。

エンティティを削除する

エンドポイントが不要になったら、余計なコストが生じないように削除する必要があります。

[predict-penguin] エンドポイントページの [⁝] メニューで、[削除] を選択します。

クリーンアップ

Azure Databricks ポータルの [コンピューティング] ページでクラスターを選択し、[■ 終了] を選択してクラスターをシャットダウンします。

Azure Databricks を調べ終わったら、作成したリソースを削除できます。これにより、不要な Azure コストが生じないようになり、サブスクリプションの容量も解放されます。

その他の情報:詳細については、Spark MLLib のドキュメントを参照してください。