Azure Machine Learning SDK v2 でコンポーネントを使用して機械学習パイプラインを作成して実行する
適用対象: Python SDK azure-ai-ml v2 (現行)
この記事では、Python SDK v2 を使って Azure Machine Learning パイプラインを構築し、データの準備、画像分類モデルのトレーニング、モデルのスコア付けという 3 つのステップを含む画像分類タスクを完了する方法について説明します。 機械学習パイプラインは、速度、移植性、再利用によってワークフローを最適化します。これにより、ユーザーはインフラストラクチャや自動化にではなく、機械学習に専念することができます。
この例では、Fashion MNIST データセットの画像を分類するために、小規模な Keras の畳み込みニューラル ネットワークをトレーニングします。 パイプラインは次のようになります。
この記事では、次のタスクを完了します。
- パイプライン ジョブの入力データを準備する
- データの準備、トレーニング、スコア付けを行う 3 つのコンポーネントを作成する
- コンポーネントからパイプラインを構成する
- コンピューティングを使ってワークスペースにアクセスする
- パイプライン ジョブを送信する
- コンポーネントとトレーニングされたニューラル ネットワークの出力を確認する
- (省略可能) コンポーネントを登録し、ワークスペース内で再利用および共有する
Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。 無料版または有料版の Azure Machine Learning を今すぐお試しください。
前提条件
Azure Machine Learning ワークスペース - このワークスペースがまだない場合は、「リソースの作成チュートリアル」を完了してください。
Azure Machine Learning Python SDK v2 をインストールした Python 環境 (インストール手順)。概要のセクションを参照してください。 この環境は、Azure Machine Learning リソースを定義および制御するためのものであり、トレーニングのためにランタイムに使用される環境とは分離されています。
examples リポジトリを複製する
トレーニング例を実行するには、最初に examples リポジトリを複製し、
sdk
ディレクトリに変更します。git clone --depth 1 https://github.com/Azure/azureml-examples cd azureml-examples/sdk
対話型の Python セッションを開始する
この記事では、Azure Machine Learning 用の Python SDK を使用して、Azure Machine Learning パイプラインを作成および制御します。 この記事では、Python REPL 環境または Jupyter ノートブックでコード スニペットを対話的に実行することを前提としています。
この記事は、Azure Machine Learning の例リポジトリの sdk/python/jobs/pipelines/2e_image_classification_keras_minist_convnet
ディレクトリにある image_classification_keras_minist_convnet.ipynb ノートブックに基づいています。
必要なライブラリをインポートする
この記事に必要な Azure Machine Learning の必須ライブラリをすべてインポートします。
# import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azure.ai.ml import MLClient
from azure.ai.ml.dsl import pipeline
from azure.ai.ml import load_component
パイプライン ジョブの入力データを準備する
この画像分類パイプラインの入力データを準備する必要があります。
Fashion-MNIST は、10 のクラスに分割されたファッションの画像のデータセットです。 各画像は 28x28 のグレースケール画像であり、6 万のトレーニングと 1 万のテストの画像があります。 Fashion-MNIST には従来の MNIST の手書き数字データベースよりも画像分類が難しいという問題があります。 これは、元の手書き数字データベースと同じ圧縮されたバイナリ形式で配布されます。
必要な Azure Machine Learning の必須ライブラリをすべてインポートします。
Input
を定義することにより、データソースの場所への参照を作成します。 データは既存の場所に残るので、追加のストレージ コストは発生しません。
パイプラインを構築するためのコンポーネントを作成する
画像分類タスクは、データの準備、モデルのトレーニング、モデルのスコア付けという 3 つのステップに分割できます。
Azure Machine Learning コンポーネントは、機械学習パイプラインで 1 つのステップを実行する自己完結型のコードです。 この記事では、画像分類タスクのために 3 つのコンポーネントを作成します。
- トレーニングとテストのためにデータを準備する
- トレーニング データを使って画像分類のためにニューラル ネットワークをトレーニングする
- テスト データを使ってモデルにスコアを付ける
各コンポーネントについて、以下を準備する必要があります。
実行ロジックを含む Python スクリプトを準備する
コンポーネントのインターフェイスを定義する
実行時環境、コンポーネントを実行するコマンドなど、コンポーネントのその他のメタデータを追加します。
次のセクションでは、コンポーネントを 2 種類の方法で作成します。最初の 2 つのコンポーネントには Python 関数を使用し、3 つ目のコンポーネントには YAML 定義を使用します。
データ準備コンポーネントを作成する
このパイプラインの最初のコンポーネントを使って、fashion_ds
の圧縮されたデータ ファイルを、トレーニング用とスコアリング用の 2 つの csv ファイルに変換します。 このコンポーネントの定義には Python 関数を使います。
Azure Machine Learning 例リポジトリ内の例に従っている場合は、ソース ファイルは既に prep/
フォルダー内にあります。 このフォルダー内には、コンポーネントを構築するファイルが 2 つあります。コンポーネントを定義する prep_component.py
と、コンポーネントの実行時環境を定義する conda.yaml
です。
Python 関数を使ってコンポーネントを定義する
command_component()
関数をデコレーターとして使うことで、コンポーネントのインターフェイス、メタデータ、実行するコードを Python 関数から簡単に定義できます。 修飾された各 Python 関数は、パイプライン サービスで処理できる 1 つの静的仕様 (YAML) に変換されます。
# Converts MNIST-formatted files at the passed-in input path to training data output path and test data output path
import os
from pathlib import Path
from mldesigner import command_component, Input, Output
@command_component(
name="prep_data",
version="1",
display_name="Prep Data",
description="Convert data to CSV file, and split to training and test data",
environment=dict(
conda_file=Path(__file__).parent / "conda.yaml",
image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
),
)
def prepare_data_component(
input_data: Input(type="uri_folder"),
training_data: Output(type="uri_folder"),
test_data: Output(type="uri_folder"),
):
convert(
os.path.join(input_data, "train-images-idx3-ubyte"),
os.path.join(input_data, "train-labels-idx1-ubyte"),
os.path.join(training_data, "mnist_train.csv"),
60000,
)
convert(
os.path.join(input_data, "t10k-images-idx3-ubyte"),
os.path.join(input_data, "t10k-labels-idx1-ubyte"),
os.path.join(test_data, "mnist_test.csv"),
10000,
)
def convert(imgf, labelf, outf, n):
f = open(imgf, "rb")
l = open(labelf, "rb")
o = open(outf, "w")
f.read(16)
l.read(8)
images = []
for i in range(n):
image = [ord(l.read(1))]
for j in range(28 * 28):
image.append(ord(f.read(1)))
images.append(image)
for image in images:
o.write(",".join(str(pix) for pix in image) + "\n")
f.close()
o.close()
l.close()
上記のコードでは、@command_component
デコレーターを使って、表示名 Prep Data
のコンポーネントを定義しています。
name
は、コンポーネントの一意の識別子です。version
はコンポーネントの現在のバージョンです。 1 つのコンポーネントに複数のバージョンを設定することができます。display_name
は、UI 内のコンポーネントを示すわかりやすい表示名であり、一意ではありません。通常、
description
を使って、このコンポーネントで完了できるタスクを記述します。environment
を使って、このコンポーネントの実行時環境を指定します。 このコンポーネントの環境では、Docker イメージを指定し、conda.yaml
ファイルを参照します。conda.yaml
ファイルには、次のようにコンポーネントに使われるすべてのパッケージが含まれています。name: imagekeras_prep_conda_env channels: - defaults dependencies: - python=3.7.11 - pip=20.0 - pip: - mldesigner==0.1.0b4
prepare_data_component
関数を使って、input_data
に対して 1 つの入力を、training_data
とtest_data
に対しては 2 つの出力を定義します。input_data
は入力データのパスです。training_data
とtest_data
はトレーニング データとテスト データの出力データ パスです。このコンポーネントを使って、
input_data
のデータをトレーニング データ csv に変換してtraining_data
にして、テスト データ csv をtest_data
に変換します。
スタジオ UI のコンポーネントは次のような外観です。
- コンポーネントはパイプライン グラフのブロックです。
input_data
、training_data
、test_data
は、他のコンポーネントに接続してデータのストリーミングを行うコンポーネントのポートです。
これで Prep Data
コンポーネントのソース ファイルをすべて準備できました。
モデルのトレーニング コンポーネントを作成する
このセクションでは、Prep Data
コンポーネントと同様に、画像分類モデルをトレーニングするコンポーネントを Python 関数で作成します。
違いは、トレーニングのロジックがより複雑になるため、元のトレーニング コードを別の Python ファイルに配置できる点です。
このコンポーネントのソース ファイルは、Azure Machine Learning 例リポジトリの train/
フォルダーの下にあります。 このフォルダーには、コンポーネントを構築するファイルが 3 つあります。
train.py
: モデルをトレーニングする実際のロジックが含まれています。train_component.py
: コンポーネントのインターフェイスを定義し、train.py
の関数をインポートします。conda.yaml
: コンポーネントの実行時環境を定義します。
実行ロジックを含むスクリプトを取得する
train.py
ファイルには通常の Python 関数が含まれており、これを使うと、画像分類のために Keras ニューラル ネットワークをトレーニングするトレーニング モデル ロジックを実行できます。 コードを確認するには、GitHub 上の train.py ファイルを参照してください。
Python 関数を使ってコンポーネントを定義する
トレーニング関数を適切に定義したら、Azure Machine Learning SDK v2 で @command_component
を使って関数をコンポーネントとしてラップし、Azure Machine Learning パイプラインで使用できます。
import os
from pathlib import Path
from mldesigner import command_component, Input, Output
@command_component(
name="train_image_classification_keras",
version="1",
display_name="Train Image Classification Keras",
description="train image classification with keras",
environment=dict(
conda_file=Path(__file__).parent / "conda.yaml",
image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
),
)
def keras_train_component(
input_data: Input(type="uri_folder"),
output_model: Output(type="uri_folder"),
epochs=10,
):
# avoid dependency issue, execution logic is in train() func in train.py file
from train import train
train(input_data, output_model, epochs)
上記のコードでは、@command_component
を使って、表示名 Train Image Classification Keras
のコンポーネントを定義しています。
keras_train_component
関数を使って、トレーニング データの取得元となる 1 つの入力input_data
、トレーニング時にエポックを指定する 1 つの入力epochs
、モデル ファイルを出力する 1 つの出力output_model
を定義します。epochs
の既定値は 10 です。 このコンポーネントの実行ロジックは、上記のtrain.py
のtrain()
関数によるものです。
モデルのトレーニング コンポーネントは、データの準備コンポーネントよりも構成が少し複雑です。 conda.yaml
は次のような内容です。
name: imagekeras_train_conda_env
channels:
- defaults
dependencies:
- python=3.8
- pip=20.2
- pip:
- mldesigner==0.1.0b12
- azureml-mlflow==1.50.0
- tensorflow==2.7.0
- numpy==1.21.4
- scikit-learn==1.0.1
- pandas==1.3.4
- matplotlib==3.2.2
- protobuf==3.20.0
これで Train Image Classification Keras
コンポーネントのソース ファイルをすべて準備できました。
モデルのスコア付けコンポーネントを作成する
このセクションでは、これまでのコンポーネントとは別に、Yaml の仕様とスクリプトを使って、トレーニング済みモデルにスコアを付けるコンポーネントを作成します。
Azure Machine Learning 例リポジトリ内の例に従っている場合は、ソース ファイルは既に score/
フォルダー内にあります。 このフォルダーには、コンポーネントを構築するファイルが 3 つあります。
score.py
: コンポーネントのソース コードを格納しますscore.yaml
: コンポーネントのインターフェイスとその他の詳細を定義します。conda.yaml
: コンポーネントの実行時環境を定義します。
実行ロジックを含むスクリプトを取得する
score.py
ファイルには、トレーニング モデル ロジックを実行する通常の Python 関数が含まれています。
from tensorflow import keras
from keras.models import Sequential
from keras.layers import Dense, Dropout, Flatten
from keras.layers import Conv2D, MaxPooling2D
from tensorflow.keras.layers import BatchNormalization
from tensorflow.keras.utils import to_categorical
from keras.callbacks import Callback
from keras.models import load_model
import argparse
from pathlib import Path
import numpy as np
import pandas as pd
import os
import matplotlib.pyplot as plt
import mlflow
def get_file(f):
f = Path(f)
if f.is_file():
return f
else:
files = list(f.iterdir())
if len(files) == 1:
return files[0]
else:
raise Exception("********This path contains more than one file*******")
def parse_args():
# setup argparse
parser = argparse.ArgumentParser()
# add arguments
parser.add_argument(
"--input_data", type=str, help="path containing data for scoring"
)
parser.add_argument(
"--input_model", type=str, default="./", help="input path for model"
)
parser.add_argument(
"--output_result", type=str, default="./", help="output path for model"
)
# parse args
args = parser.parse_args()
# return args
return args
def score(input_data, input_model, output_result):
test_file = get_file(input_data)
data_test = pd.read_csv(test_file, header=None)
img_rows, img_cols = 28, 28
input_shape = (img_rows, img_cols, 1)
# Read test data
X_test = np.array(data_test.iloc[:, 1:])
y_test = to_categorical(np.array(data_test.iloc[:, 0]))
X_test = (
X_test.reshape(X_test.shape[0], img_rows, img_cols, 1).astype("float32") / 255
)
# Load model
files = [f for f in os.listdir(input_model) if f.endswith(".h5")]
model = load_model(input_model + "/" + files[0])
# Log metrics of the model
eval = model.evaluate(X_test, y_test, verbose=0)
mlflow.log_metric("Final test loss", eval[0])
print("Test loss:", eval[0])
mlflow.log_metric("Final test accuracy", eval[1])
print("Test accuracy:", eval[1])
# Score model using test data
y_predict = model.predict(X_test)
y_result = np.argmax(y_predict, axis=1)
# Output result
np.savetxt(output_result + "/predict_result.csv", y_result, delimiter=",")
def main(args):
score(args.input_data, args.input_model, args.output_result)
# run script
if __name__ == "__main__":
# parse args
args = parse_args()
# call main function
main(args)
score.py のコードは、input_data
、input_model
、output_result
という 3 つのコマンドライン引数を受け取ります。 このプログラムを実行すると、入力データを使って入力モデルにスコアを付け、スコアリング結果を出力することができます。
Yaml を使ってコンポーネントを定義する
このセクションでは、有効な YAML コンポーネント仕様形式でコンポーネント仕様を作成する方法について説明します。 このファイルによって、次の情報が指定されます。
- メタデータ: name、display_name、version、type など。
- インターフェイス: 入力と出力
- コマンド、コード、および環境: コンポーネントの実行に使用されるコマンド、コード、および環境
$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json
type: command
name: score_image_classification_keras
display_name: Score Image Classification Keras
inputs:
input_data:
type: uri_folder
input_model:
type: uri_folder
outputs:
output_result:
type: uri_folder
code: ./
command: python score.py --input_data ${{inputs.input_data}} --input_model ${{inputs.input_model}} --output_result ${{outputs.output_result}}
environment:
conda_file: ./conda.yaml
image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
name
は、コンポーネントの一意の識別子です。 その表示名はScore Image Classification Keras
です。- このコンポーネントは、2 つの入力と 1 つの出力があります。
- このソース コード パスは
code
セクションで定義されており、クラウド内でコンポーネントを実行すると、そのパスのすべてのファイルがこのコンポーネントのスナップショットとしてアップロードされます。 command
セクションには、このコンポーネントの実行時に実行するコマンドを指定します。environment
セクションには、docker イメージと conda yaml ファイルが含まれています。 ソース ファイルはサンプル リポジトリにあります。
これで、モデルのスコア付けコンポーネントのソース ファイルはすべて揃いました。
コンポーネントを読み込んでパイプラインを構築する
Python 関数で定義したデータの準備コンポーネントとモデルのトレーニング コンポーネントの場合は、通常の Python 関数と同様にコンポーネントをインポートできます。
次のコードでは、prep
フォルダーの下にある prep_component.py
ファイルと、train
フォルダーの下にある train_component
ファイルから、それぞれ prepare_data_component()
と keras_train_component()
の関数をインポートしています。
%load_ext autoreload
%autoreload 2
# load component function from component python file
from prep.prep_component import prepare_data_component
from train.train_component import keras_train_component
# print hint of components
help(prepare_data_component)
help(keras_train_component)
yaml で定義されたスコア付けコンポーネントの場合は、load_component()
関数を使って読み込むことができます。
# load component function from yaml
keras_score_component = load_component(source="./score/score.yaml")
パイプラインを構築する
これで、パイプラインの構築に使うすべてのコンポーネントと入力データの作成と読み込みを完了しました。 これらをパイプラインに構成することができます。
Note
サーバーレス コンピューティングを使うには、先頭に from azure.ai.ml.entities import ResourceConfiguration
を追加します。
そして、次のように置き換えます。
default_compute="serverless",
を含むdefault_compute=cpu_compute_target,
train_node.resources = "ResourceConfiguration(instance_type="Standard_NC6s_v3",instance_count=2)
を含むtrain_node.compute = gpu_compute_target
# define a pipeline containing 3 nodes: Prepare data node, train node, and score node
@pipeline(
default_compute=cpu_compute_target,
)
def image_classification_keras_minist_convnet(pipeline_input_data):
"""E2E image classification pipeline with keras using python sdk."""
prepare_data_node = prepare_data_component(input_data=pipeline_input_data)
train_node = keras_train_component(
input_data=prepare_data_node.outputs.training_data
)
train_node.compute = gpu_compute_target
score_node = keras_score_component(
input_data=prepare_data_node.outputs.test_data,
input_model=train_node.outputs.output_model,
)
# create a pipeline
pipeline_job = image_classification_keras_minist_convnet(pipeline_input_data=mnist_ds)
パイプラインには既定のコンピューティング cpu_compute_target
があります。つまり、特定のノードにコンピューティングを指定しない場合、そのノードは既定のコンピューティング上で実行されます。
パイプラインは、パイプライン レベルの入力 pipeline_input_data
があります。 パイプライン ジョブを送信するときに、パイプライン入力に値を割り当てることができます。
パイプラインには、prepare_data_node、train_node、score_node の 3 つのノードが含まれています。
prepare_data_node
のinput_data
にはpipeline_input_data
の値が使われます。train_node
のinput_data
は prepare_data_node のtraining_data
出力のものです。score_node の
input_data
は prepare_data_node のtest_data
出力、input_model
は train_node のoutput_model
出力のものです。train_node
によって CNN モデルがトレーニングされるので、そのコンピューティングを gpu_compute_target として指定できます。こうすることで、トレーニングのパフォーマンスが向上します。
パイプライン ジョブを送信する
パイプラインを構築したら、ワークスペースに送信します。 ジョブを送信するには、まずワークスペースに接続する必要があります。
ワークスペースにアクセスする
資格情報を構成する
DefaultAzureCredential
を使用してワークスペースにアクセスします。 DefaultAzureCredential
を使うと、ほとんどの Azure SDK 認証シナリオに対応できます。
うまくいかない場合は、その他の使用できる資格情報については、資格情報の構成例の記事と azure-identity のリファレンス ドキュメントを参照してください。
try:
credential = DefaultAzureCredential()
# Check if given credential can get token successfully.
credential.get_token("https://management.azure.com/.default")
except Exception as ex:
# Fall back to InteractiveBrowserCredential in case DefaultAzureCredential not work
credential = InteractiveBrowserCredential()
コンピューティングを使ってワークスペースへのハンドルを取得する
Azure Machine Learning Services を管理する MLClient
オブジェクトを作成します。 サーバーレス コンピューティングを使用する場合は、これらのコンピューティングを作成する必要はありません。
# Get a handle to workspace
ml_client = MLClient.from_config(credential=credential)
# Retrieve an already attached Azure Machine Learning Compute.
cpu_compute_target = "cpu-cluster"
print(ml_client.compute.get(cpu_compute_target))
gpu_compute_target = "gpu-cluster"
print(ml_client.compute.get(gpu_compute_target))
重要
このコード スニペットでは、ワークスペースの構成 json ファイルが現在のディレクトリまたはその親に保存されていることを想定しています。 ワークスペースの作成方法について詳しくは、ワークスペース リソースの作成に関するページを参照してください。 構成をファイルに保存する方法について詳しくは、「ワークスペース構成ファイルを作成する」を参照してください。
パイプライン ジョブをワークスペースに送信する
ワークスペースのハンドルを取得したので、パイプライン ジョブを送信できます。
pipeline_job = ml_client.jobs.create_or_update(
pipeline_job, experiment_name="pipeline_samples"
)
pipeline_job
上記のコードでは、画像分類パイプライン ジョブを pipeline_samples
という実験に送信しています。 実験が存在しない場合は、自動的に作成されます。 pipeline_input_data
は fashion_ds
を使います。
pipeline_job
を呼び出すと、次のような出力が生成されます。
Experiment
の submit
の呼び出しが迅速完了し、次のような出力が生成されます。
実験 | 名前 | 種類 | 状態 | 詳細ページ |
---|---|---|---|---|
pipeline_samples | sharp_pipe_4gvqx6h1fb | pipeline | 準備 | Azure Machine Learning スタジオへのリンク。 |
パイプラインの実行を監視するには、リンクを開くか、次を実行して完了するまでブロックします。
# wait until the job completes
ml_client.jobs.stream(pipeline_job.name)
重要
パイプラインの初回実行には約 "15 分" かかります。 依存関係をすべてダウンロードする必要があるほか、Docker イメージの作成と Python 環境のプロビジョニングおよび作成が行われます。 パイプラインを再度実行する際は、それらのリソースが作成されるのではなく再利用されるので、時間が大幅に短縮されます。 ただし、パイプラインの総実行時間は、実際のスクリプトのワークロードと、各パイプライン ステップで実行される処理によって異なります。
出力をチェックアウトし、UI でパイプラインをデバッグする
パイプラインのジョブ詳細ページである Link to Azure Machine Learning studio
を開くことができます。 次のようなパイプラインのグラフが表示されます。
コンポーネントを右クリックして各コンポーネントのログと出力を確認することや、コンポーネントを選んで詳細ウィンドウを開くことができます。 UI でパイプラインをデバッグする方法については、パイプライン エラーをデバッグする方法に関する記事を参照してください。
(省略可能) コンポーネントをワークスペースに登録する
前のセクションでは、3 つのコンポーネントを使ってパイプラインを構築し、E2E で画像分類タスクを完了しました。 ワークスペースにコンポーネントを登録することで、ワークスペース内でコンポーネントを共有したり、再利用したりすることもできます。 たとえば、次のようにデータの準備コンポーネントを登録することができます。
try:
# try get back the component
prep = ml_client.components.get(name="prep_data", version="1")
except:
# if not exists, register component using following code
prep = ml_client.components.create_or_update(prepare_data_component)
# list all components registered in workspace
for c in ml_client.components.list():
print(c)
ml_client.components.get()
を使うと、登録されているコンポーネントを名前とバージョンを指定して取得できます。 ml_client.components.create_or_update()
を使うと、あらかじめ Python 関数または yaml から読み込んだコンポーネントを登録できます。
次の手順
- 機械学習 SDK を使用してパイプラインを作成するその他の例については、例のリポジトリを参照してください。
- スタジオ UI を使ってパイプラインを送信し、デバッグする方法については、UI でコンポーネントを使ってパイプラインを作成する方法に関する記事を参照してください。
- Azure Machine Learning CLI を使ってコンポーネントとパイプラインを作成する方法については、CLI でコンポーネントを使ってパイプラインを作成する方法に関する記事を参照してください。
- バッチ エンドポイントを使って運用環境にパイプラインをデプロイする方法については、バッチ エンドポイントを使ってパイプラインをデプロイする方法に関する記事をご覧ください。