前処理を使用してバッチ スコアリングを実行するパイプラインをデプロイする方法

適用対象:Azure CLI ml extension v2 (現行)Python SDK azure-ai-ml v2 (現行)

この記事では、バッチ エンドポイントの下に推論 (またはスコアリング) パイプラインをデプロイする方法について説明します。 パイプラインは、登録されたモデルに対してスコアリングを実行すると同時に、モデルがトレーニングされたときの前処理コンポーネントを再利用します。 同じ前処理コンポーネントを再利用すると、スコアリング中に同じ前処理が確実に適用されます。

次のことを学習します。

• ワークスペースから既存のコンポーネントを再利用するパイプラインを作成する
• パイプラインをエンドポイントにデプロイする
• パイプラインによって生成された予測を使用する

この例の概要

この例では、推論にモデルを使用する前に、前処理コードと前処理中に学習したパラメーターを再利用する方法を示します。 前処理コードと学習したパラメーターを再利用することで、トレーニング中に入力データに適用されたのと同じ変換 (正規化や機能エンコードなど) も推論中に適用されるようにすることができます。 推論に使用されるモデルは、UCI 心臓病データ セットの表形式データに対して予測を実行します。

パイプラインを視覚化すると次のようになります。

この記事の例は、azureml-examples リポジトリに含まれているコード サンプルを基にしています。 YAML などのファイルをコピーして貼り付けることなくコマンドをローカルで実行するには、最初にリポジトリを複製してから、ディレクトリをそのフォルダーに変更します。

Azure CLI

git clone https://github.com/Azure/azureml-examples --depth 1
cd azureml-examples/cli

Python

git clone https://github.com/Azure/azureml-examples --depth 1
cd azureml-examples/sdk/python

この例のファイルは、次の場所にあります。

cd endpoints/batch/deploy-pipelines/batch-scoring-with-preprocessing

Jupyter ノートブックで作業を進める

この例の Python SDK バージョンに沿って作業を進めることができます。そのためには、複製されたリポジトリで sdk-deploy-and-test.ipynb ノートブックを開きます。

前提条件

この記事の手順に従う前に、次の前提条件が満たされていることをご確認ください。

• Azure サブスクリプション。 Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。 無料版または有料版の Azure Machine Learning をお試しください。

• Azure Machine Learning ワークスペース。 準備できていない場合は、Microsoft Azure Machine Learning ワークスペースの管理に関する記事の手順を使用して作成します。

• ワークスペースに次のアクセス許可があることを確認します。

  • バッチ エンドポイントとバッチ デプロイを作成または管理する: 所有者または共同作成者のロール、あるいは Microsoft.MachineLearningServices/workspaces/batchEndpoints/* を許可するカスタム ロールを使用します。

  • ワークスペース リソース グループに ARM デプロイを作成する: 所有者または共同作成者のロール、あるいはワークスペースがデプロイされているリソース グループで Microsoft.Resources/deployments/write を許可するカスタム ロールを使用します。

• Azure Machine Learning を使用するには、次のソフトウェアをインストールする必要があります。

  Azure CLI

  Azure CLI と mlAzure Machine Learning 用の 拡張機能。

  az extension add -n ml
  

  注意

  Batch エンドポイントのパイプライン コンポーネント デプロイは、Azure CLI 用 ml 拡張機能のバージョン 2.7 で導入されました。 az extension update --name ml を使用して、最新バージョンを取得します。

  Python

  Azure Machine Learning SDK for Python。

  pip install azure-ai-ml
  

  注意

  クラス ModelBatchDeployment と PipelineComponentBatchDeployment は、SDK のバージョン 1.7.0 で導入されました。 pip install -U azure-ai-ml を使用して、最新バージョンを取得します。

ワークスペースに接続する

ワークスペースは、Azure Machine Learning の最上位のリソースで、Azure Machine Learning を使用するときに作成するすべての成果物を操作するための一元的な場所を提供します。 このセクションでは、デプロイ タスクを実行するワークスペースに接続します。

Azure CLI

次のコードで、サブスクリプション ID、ワークスペース、場所、リソース グループの値を渡します。

az account set --subscription <subscription>
az configure --defaults workspace=<workspace> group=<resource-group> location=<location>

Python

1. 必要なライブラリをインポートします。

   from azure.ai.ml import MLClient, Input, load_component
   from azure.ai.ml.entities import BatchEndpoint, ModelBatchDeployment, ModelBatchDeploymentSettings, PipelineComponentBatchDeployment, Model, AmlCompute, Data, BatchRetrySettings, CodeConfiguration, Environment, Data
   from azure.ai.ml.constants import AssetTypes, BatchDeploymentOutputAction
   from azure.ai.ml.dsl import pipeline
   from azure.identity import DefaultAzureCredential
   

2. ワークスペースの詳細を構成し、ワークスペースへのハンドルを取得します。

   次のコードで、サブスクリプション ID、ワークスペース、リソース グループの値を渡します。

   subscription_id = "<subscription>"
   resource_group = "<resource-group>"
   workspace = "<workspace>"
   
   ml_client = MLClient(DefaultAzureCredential(), subscription_id, resource_group, workspace)
   

推論パイプラインを作成する

このセクションでは、推論パイプラインに必要なすべてのアセットを作成します。 まず、パイプラインのコンポーネントに必要なライブラリを含む環境を作成します。 次に、バッチ デプロイを実行するコンピューティング クラスターを作成します。 その後、推論パイプラインを構築するために必要なコンポーネント、モデル、変換を登録します。 最後に、パイプラインをビルドしてテストします。

環境の作成

この例のコンポーネントでは、XGBoost および scikit-learn ライブラリを含む環境を使用します。 environment/conda.yml ファイルには、環境の構成が含まれます。

environment/conda.yml

channels:
- conda-forge
dependencies:
- python=3.8.5
- pip
- pip:
  - mlflow
  - azureml-mlflow
  - datasets
  - jobtools
  - cloudpickle==1.6.0
  - dask==2023.2.0
  - scikit-learn==1.1.2
  - xgboost==1.3.3
name: mlflow-env

次のように環境を作成します。

1. 環境を定義します。

   Azure CLI

   environment/xgboost-sklearn-py38.yml

   $schema: https://azuremlschemas.azureedge.net/latest/environment.schema.json
   name: xgboost-sklearn-py38
   image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest
   conda_file: conda.yml
   description: An environment for models built with XGBoost and Scikit-learn.
   

   Python

   environment = Environment(
       name="xgboost-sklearn-py38",
       description="An environment for models built with XGBoost and Scikit-learn.",
       image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest",
       conda_file="environment/conda.yml",
   )

2. 環境を作成します。

   Azure CLI

   az ml environment create -f environment/xgboost-sklearn-py38.yml
   

   Python

   try:
       ml_client.environments.create_or_update(environment)
   except ResourceExistsError:
       pass

コンピューティング クラスターを作成する

バッチ エンドポイントとバッチ デプロイは、コンピューティング クラスター上で実行されます。 これらは、ワークスペースに既に存在する任意の Azure Machine Learning コンピューティング クラスター上で実行できます。 したがって、複数のバッチ デプロイが同じコンピューティング インフラストラクチャを共有できます。 この例では、batch-cluster という名前の Azure Machine Learning コンピューティング クラスター上で作業します。 ワークスペースにコンピューティングが存在することを確認し、存在しない場合は作成します。

Azure CLI

az ml compute create -n batch-cluster --type amlcompute --min-instances 0 --max-instances 5

Python

compute_name = "batch-cluster"
if not any(filter(lambda m: m.name == compute_name, ml_client.compute.list())):
    compute_cluster = AmlCompute(
        name=compute_name,
        description="Batch endpoints compute cluster",
        min_instances=0,
        max_instances=5,
    )
    ml_client.begin_create_or_update(compute_cluster).result()

コンポーネントとモデルを登録する

推論パイプラインを構築するために必要なコンポーネント、モデル、変換を登録します。 トレーニング ルーチン用のこうしたアセットの一部は再利用できます。

ヒント

このチュートリアルでは、以前のトレーニング パイプラインからモデルと前処理コンポーネントを再利用します。 バッチ エンドポイントを使用してトレーニング パイプラインをデプロイする方法の例に従って、それらが作成された方法を確認できます。

1. 予測に使用するモデルを登録します。

   Azure CLI

   az ml model create --name heart-classifier --type mlflow_model --path model
   

   Python

   model_name = "heart-classifier"
   model_local_path = "model"
   
   model = ml_client.models.create_or_update(
       Model(name=model_name, path=model_local_path, type=AssetTypes.MLFLOW_MODEL)
   )

2. 登録されたモデルは、入力データに対して直接トレーニングされていませんでした。 代わりに、入力データは、準備コンポーネントを使用してトレーニング前に前処理 (または変換) されていました。 このコンポーネントも登録する必要があります。 準備コンポーネントを登録します。

   Azure CLI

   az ml component create -f components/prepare/prepare.yml
   

   Python

   prepare_data = load_component(source="components/prepare/prepare.yml")
   
   ml_client.components.create_or_update(prepare_data)

   ヒント

   準備コンポーネントを登録したら、ワークスペースから参照できるようになります。 たとえば、azureml:uci_heart_prepare@latest では準備コンポーネントの最後のバージョンを取得します。

3. 準備コンポーネントのデータ変換の一環として、入力データは、予測器を中心に配置して、その値を [-1, 1] の範囲に制限するように正規化されました。 変換パラメーターは scikit-learn 変換でキャプチャされました。これは、後で新しいデータを取得したときに適用するために登録することもできます。 次のように変換を登録します。

   Azure CLI

   az ml model create --name heart-classifier-transforms --type custom_model --path transformations
   

   Python

   transformation_name = "heart-classifier-transforms"
   transformation_local_path = "transformations"
   
   transformations = ml_client.models.create_or_update(
       Model(
           name=transformation_name,
           path=transformation_local_path,
           type=AssetTypes.CUSTOM_MODEL,
       )
   )

4. 特定のモデルの予測を計算する score という名前の別のコンポーネントを使用して、登録済みモデルの推論を実行します。 コンポーネントをその定義から直接参照します。

   ヒント

   ベスト プラクティスは、コンポーネントを登録し、それをパイプラインから参照することです。 ただし、この例では、トレーニング パイプラインから再利用されるコンポーネントと新しいコンポーネントが分かるように、定義からコンポーネントを直接参照します。

パイプラインを構築する

次に、すべての要素をバインドします。 デプロイする推論パイプラインには、次の 2 つのコンポーネント (ステップ) があります。

• preprocess_job: このステップでは、入力データを読み取り、準備されたデータと適用された変換を返します。 このステップは、次の 2 つの入力を受け取ります。
  • data: スコアリングする入力データを含むフォルダー
  • transformations: (オプション) 適用される変換へのパス (使用可能な場合)。 指定すると、変換はパスに示されているモデルから読み取られます。 ただし、パスが指定されない場合、変換は入力データから学習されます。 ただし、推論の場合、トレーニング中に学習したのと同じパラメーター値を使用する必要があるため、入力データから変換パラメーター (この例では正規化係数) を学習することはできません。 この入力はオプションであるため、トレーニングとスコアリング中に preprocess_job コンポーネントを使用できます。
• score_job: このステップでは、入力モデルを使用して、変換されたデータに対して推論を実行します。 コンポーネントが MLflow モデルを使用して推論を実行していることに注意してください。 最後に、スコアは読み取られたのと同じ形式で書き戻されます。

Azure CLI

パイプライン構成は、pipeline.yml ファイルで定義されます。

pipeline.yml

$schema: https://azuremlschemas.azureedge.net/latest/pipelineComponent.schema.json
type: pipeline

name: batch_scoring_uci_heart
display_name: Batch Scoring for UCI heart
description: This pipeline demonstrates how to make batch inference using a model from the Heart Disease Data Set problem, where pre and post processing is required as steps. The pre and post processing steps can be components reusable from the training pipeline.

inputs:
  input_data:
    type: uri_folder
  score_mode:
    type: string
    default: append

outputs: 
  scores:
    type: uri_folder
    mode: upload

jobs:
  preprocess_job:
    type: command
    component: azureml:uci_heart_prepare@latest
    inputs:
      data: ${{parent.inputs.input_data}}
      transformations: 
        path: azureml:heart-classifier-transforms@latest
        type: custom_model
    outputs:
      prepared_data:
  
  score_job:
    type: command
    component: components/score/score.yml
    inputs:
      data: ${{parent.jobs.preprocess_job.outputs.prepared_data}}
      model:
        path: azureml:heart-classifier@latest
        type: mlflow_model
      score_mode: ${{parent.inputs.score_mode}}
    outputs:
      scores: 
        mode: upload
        path: ${{parent.outputs.scores}}

Python

prepare_data = ml_client.components.get("uci_heart_prepare", label="latest")
score_data = load_component(source="components/score/score.yml")

関数でパイプラインをビルドしてみましょう。

@pipeline()
def uci_heart_classifier_scorer(
    input_data: Input(type=AssetTypes.URI_FOLDER), score_mode: str
):
    """This pipeline demonstrates how to make batch inference using a model from the Heart Disease Data Set problem, where pre and post processing is required as steps. The pre and post processing steps can be components reusable from the training pipeline."""
    prepared_data = prepare_data(
        data=input_data,
        transformations=Input(type=AssetTypes.CUSTOM_MODEL, path=transformations.id),
    )
    scored_data = score_data(
        data=prepared_data.outputs.prepared_data,
        model=Input(type=AssetTypes.MLFLOW_MODEL, path=model.id),
        score_mode=score_mode,
    )

    return {"scores": scored_data.outputs.scores}

パイプラインを視覚化すると次のようになります。

パイプラインをテストする

いくつかのサンプル データを使用してパイプラインをテストします。 これを行うために、パイプラインと、以前に作成した batch-cluster コンピューティング クラスターを使用してジョブを作成します。

Azure CLI

次の pipeline-job.yml ファイルには、パイプライン ジョブの構成が含まれます。

pipeline-job.yml

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

display_name: uci-classifier-score-job
description: |-
  This pipeline demonstrate how to make batch inference using a model from the Heart \
  Disease Data Set problem, where pre and post processing is required as steps. The \
  pre and post processing steps can be components reused from the training pipeline.

compute: batch-cluster
component: pipeline.yml
inputs:
  input_data:
    type: uri_folder
  score_mode: append
outputs: 
  scores:
    mode: upload

Python

pipeline_job = uci_heart_classifier_scorer(
    input_data=Input(type="uri_folder", path="data/unlabeled/"), score_mode="append"
)

次に、テストを実行するためにいくつかの実行設定を構成します。

pipeline_job.settings.default_datastore = "workspaceblobstore"
pipeline_job.settings.default_compute = "batch-cluster"

テスト ジョブを作成します。

Azure CLI

az ml job create -f pipeline-job.yml --set inputs.input_data.path=data/unlabeled

Python

pipeline_job_run = ml_client.jobs.create_or_update(
    pipeline_job, experiment_name="uci-heart-score-pipeline"
)
pipeline_job_run

バッチ エンドポイントを作成する

1. エンドポイントの名前を指定します。 バッチ エンドポイントの名前は、呼び出し URI の構成に使用されるため、各リージョンで一意である必要があります。 一意性を確保するために、次のコードで指定する名前に末尾文字を追加します。

   Azure CLI

   ENDPOINT_NAME="uci-classifier-score"
   

   Python

   endpoint_name = "uci-classifier-score"

2. エンドポイントを構成します。

   Azure CLI

   endpoint.yml ファイルには、エンドポイントの構成が含まれます。

   endpoint.yml

   $schema: https://azuremlschemas.azureedge.net/latest/batchEndpoint.schema.json
   name: uci-classifier-score
   description: Batch scoring endpoint of the Heart Disease Data Set prediction task.
   auth_mode: aad_token
   

   Python

   endpoint = BatchEndpoint(
       name=endpoint_name,
       description="Batch scoring endpoint of the Heart Disease Data Set prediction task",
   )

3. エンドポイントを作成します。

   Azure CLI

   az ml batch-endpoint create --name $ENDPOINT_NAME -f endpoint.yml
   

   Python

   ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

4. エンドポイント URI にクエリを実行します。

   Azure CLI

   az ml batch-endpoint show --name $ENDPOINT_NAME
   

   Python

   endpoint = ml_client.batch_endpoints.get(name=endpoint_name)
   print(endpoint)

パイプライン コンポーネントをデプロイする

パイプライン コンポーネントをデプロイするには、バッチ デプロイを作成する必要があります。 デプロイは、実際の作業を行うアセットをホスティングするために必要なリソースのセットです。

1. 展開を構成する

   Azure CLI

   deployment.yml ファイルには、デプロイの構成が含まれます。 追加のプロパティについては、完全なバッチ エンドポイント YAML スキーマを確認してください。

   deployment.yml

   $schema: https://azuremlschemas.azureedge.net/latest/pipelineComponentBatchDeployment.schema.json
   name: uci-classifier-prepros-xgb
   endpoint_name: uci-classifier-batch
   type: pipeline
   component: pipeline.yml
   settings:
       continue_on_step_failure: false
       default_compute: batch-cluster
   

   Python

   パイプラインは関数で定義されています。 それをコンポーネントに変換するには、それから component プロパティを使用します。 パイプライン コンポーネントは再利用可能なコンピューティング グラフであり、バッチ デプロイに含めたり、より複雑なパイプラインを構成するために使用したりできます。

   pipeline_component = ml_client.components.create_or_update(
       uci_heart_classifier_scorer().component
   )

   これで、デプロイを定義できるようになりました。

   deployment = PipelineComponentBatchDeployment(
       name="uci-classifier-prepros-xgb",
       description="A sample deployment with pre and post processing done before and after inference.",
       endpoint_name=endpoint.name,
       component=pipeline_component,
       settings={"continue_on_step_failure": False, "default_compute": compute_name},
   )

2. 配置を作成する

   Azure CLI

   次のコードを実行して、バッチ エンドポイントの下にバッチ デプロイを作成し、それを既定のデプロイとして設定します。

   az ml batch-deployment create --endpoint $ENDPOINT_NAME -f deployment.yml --set-default
   

   ヒント

   この新しいデプロイが既定になったことを示すために --set-default フラグが使用されていることに注目してください。

   Python

   このコマンドは、デプロイの作成を開始し、デプロイの作成が続行されている間に確認応答を返します。

   ml_client.batch_deployments.begin_create_or_update(deployment).result()

   作成したら、この新しいデプロイを既定として構成します。

   endpoint = ml_client.batch_endpoints.get(endpoint_name)
   endpoint.defaults.deployment_name = deployment.name
   ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

3. デプロイを使用する準備が整いました。

展開をテスト

デプロイが作成されると、ジョブを受け取る準備が整います。 次の手順に従ってテストします。

1. デプロイでは、1 つのデータ入力と 1 つのリテラル入力を指定する必要があります。

   Azure CLI

   inputs.yml ファイルには、入力データ アセットの定義が含まれます。

   inputs.yml

   inputs:
     input_data:
       type: uri_folder
       path: data/unlabeled
     score_mode:
       type: string
       default: append
   outputs:
     scores:
       type: uri_folder
       mode: upload
   

   Python

   入力データ アセットの定義:

   input_data = Input(type="uri_folder", path="data/unlabeled/")
   score_mode = Input(type="string", default="append")

   ヒント

   入力を指定する方法の詳細については、「バッチ エンドポイントのジョブと入力データを作成する」を参照してください。

2. 次のように、既定のデプロイを呼び出すことができます。

   Azure CLI

   JOB_NAME=$(az ml batch-endpoint invoke -n $ENDPOINT_NAME --f inputs.yml --query name -o tsv)
   

   Python

   ヒント

   エンドポイントを呼び出すときの inputs と input の違いは何ですか?

   一般的には invoke メソッドで辞書 inputs = {} を使用すると、"モデル デプロイ" または "パイプライン デプロイ" を含むバッチ エンドポイントに、必要な入力を任意の数だけ与えることができます。

   "モデル デプロイ" では常に 1 つのデータ入力しかとらないため、モデル デプロイの場合、input を使用して、デプロイの入力データの場所を簡単に指定することができます。

   job = ml_client.batch_endpoints.invoke(
       endpoint_name=endpoint.name,
       inputs={"input_data": input_data, "score_mode": score_mode},
   )

3. 次を使用して、ログの表示およびストリーミングの進行状況を監視できます。

   Azure CLI

   az ml job stream -n $JOB_NAME
   

   Python

   ml_client.jobs.get(job.name)

   ジョブが完了するまで待つには、次のコードを実行します。

   ml_client.jobs.stream(name=job.name)

ジョブ出力にアクセスする

ジョブが完了すると、その出力にアクセスできます。 このジョブには、scores という名前の出力が 1 つだけ含まれています。

Azure CLI

az ml job download を使用して、関連する結果をダウンロードできます。

az ml job download --name $JOB_NAME --output-name scores

Python

結果をダウンロードします。

ml_client.jobs.download(name=job.name, download_path=".", output_name="scores")

スコアリングされたデータを読み取ります。

import pandas as pd
import glob

output_files = glob.glob("named-outputs/scores/*.csv")
score = pd.concat((pd.read_csv(f) for f in output_files))
score

出力は次のようになります。

age	sex	...	thal	prediction
0.9338	1	...	2	0
1.3782	1	...	3	1
1.3782	1	...	4	0
-1.954	1	...	3	0

出力には、予測と、前処理された score コンポーネントに指定されたデータが含まれます。 たとえば、列 age は正規化されており、列 thal には元のエンコード値が含まれています。 実際には、予測のみを出力してから、元の値と連結したいと考える可能性が高いです。 この作業は読者に任せられます。

リソースをクリーンアップする

完了したら、関連付けられているリソースをワークスペースから削除します。

Azure CLI

次のコードを実行して、バッチ エンドポイントとその基になっているデプロイを削除します。 --yes は、削除を確認するために使用します。

az ml batch-endpoint delete -n $ENDPOINT_NAME --yes

Python

エンドポイントを削除します:

ml_client.batch_endpoints.begin_delete(endpoint_name)

(オプション) 後のデプロイでコンピューティング クラスターを再利用する予定がない場合は、コンピューティングを削除します。

Azure CLI

az ml compute delete -n batch-cluster

Python

ml_client.compute.begin_delete(name="batch-cluster")

次のステップ

• パイプライン ジョブからバッチ エンドポイントを作成する
• バッチ エンドポイント ジョブからデータにアクセスする
• バッチ エンドポイントのトラブルシューティング