次の方法で共有


Machine Learning SDK v2 でコンポーネントを使用して機械学習パイプラインを作成して実行する

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

この記事では、Azure Machine Learning Python SDK v2 を使用して Azure Machine Learning パイプライン を構築し、データの準備、画像分類モデルのトレーニング、モデルのスコア付けという 3 つの手順を含む画像分類タスクを完了する方法について説明します。 Machine Learning パイプラインは、速度、移植性、再利用によってワークフローを最適化するため、インフラストラクチャと自動化の代わりに機械学習に集中できます。

この例のパイプラインでは、小さな Keras 畳み込みニューラルネットワークをトレーニングし、Fashion MNIST データセット内の画像を分類します。 パイプラインは次のようになります。

画像分類の例のパイプライン グラフを示すスクリーンショット。

この記事では、次のタスクを完了します。

  • パイプライン ジョブの入力データを準備します。
  • データの準備、イメージのトレーニング、モデルのスコア付けを行う 3 つのコンポーネントを作成します。
  • コンポーネントからパイプラインを構築します。
  • コンピューティングを持つワークスペースにアクセスできます。
  • パイプライン ジョブを送信します。
  • コンポーネントとトレーニング済みのニューラル ネットワークの出力を確認します。
  • (省略可能)ワークスペース内でさらに再利用および共有するために、コンポーネントを登録します。

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

前提条件

  • Azure Machine Learning ワークスペース。 お持ちでない場合は、「 リソースの作成」チュートリアルを完了してください。

  • Azure Machine Learning Python SDK v2 をインストールした Python 環境。 インストール手順については、「 作業の開始」を参照してください。 この環境は、Azure Machine Learning リソースを定義および制御するための環境であり、実行時にトレーニングに使用される環境とは別です。

  • サンプル リポジトリの複製。

    トレーニング例を実行するには、まずサンプル リポジトリを複製し、 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 Notebook でコード スニペットを対話形式で実行するという前提に基づいて記述されています。

この記事は、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

パイプライン ジョブの入力データを準備する

画像分類パイプラインの入力データを準備する必要があります。

ファッション MNIST は、10 のクラスに分けられたファッション イメージのデータセットです。 各イメージは、28 x 28 のグレースケール イメージです。 トレーニング画像は 60,000 個、テスト画像は 10,000 個あります。 画像分類の問題として、Fashion MNIST は従来の MNIST 手書き数字データベースよりも困難です。 これは、元の手書き数字データベースと同じ圧縮されたバイナリ形式で配布されます。

Input を定義することにより、データソースの場所への参照を作成します。 データは既存の場所に残るので、追加のストレージ コストは発生しません。

パイプラインを構築するためのコンポーネントを作成する

画像分類タスクは、データの準備、モデルのトレーニング、モデルのスコア付けという 3 つの手順に分割できます。

Azure Machine Learning コンポーネントは、機械学習パイプラインの 1 つのステップを完了する自己完結型のコードです。 この記事では、画像分類タスク用の 3 つのコンポーネントを作成します。

  • トレーニング用のデータを準備し、テストします。
  • トレーニング データを使用して、画像分類用のニューラル ネットワークをトレーニングします。
  • テスト データを使用してモデルにスコアを付けます。

コンポーネントごとに、次の手順を実行する必要があります。

  1. 実行ロジックを含む Python スクリプトを準備します。

  2. コンポーネントのインターフェイスを定義します。

  3. ランタイム環境やコンポーネントを実行するコマンドなど、コンポーネントの他のメタデータを追加します。

次のセクションでは、2 つの方法でコンポーネントを作成する方法を示します。 最初の 2 つのコンポーネントでは、Python 関数を使用します。 3 番目のコンポーネントでは、YAML 定義を使用します。

データ準備コンポーネントを作成する

このパイプラインの最初のコンポーネントは、 fashion_ds の圧縮されたデータ ファイルを 2 つの .csv ファイルに変換します。1 つはトレーニング用、もう 1 つはスコアリング用です。 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_datatest_data に対しては 2 つの出力を定義します。 input_data は入力データ パスです。 training_datatest_data はトレーニング データとテスト データの出力データ パスです。

  • コンポーネントは、データをトレーニングするための input_data からデータを training_data .csvに変換し、データをテストするための test_data .csvに変換します。

これは、Studio UI でのコンポーネントの外観です。

  • コンポーネントはパイプライン グラフのブロックです。
  • input_datatraining_data、および test_data は、データ ストリーミングのために他のコンポーネントに接続するコンポーネントのポートです。

UI およびコード内の Prep Data コンポーネントのスクリーンショット。

これで、 Prep Data コンポーネントのすべてのソース ファイルを準備できました。

モデル トレーニング コンポーネントを作成する

このセクションでは、 Prep Data コンポーネントと同様に、Python 関数で画像分類モデルをトレーニングするためのコンポーネントを作成します。

トレーニング ロジックはより複雑であるため、トレーニング コードを別の Python ファイルに配置します。

このコンポーネントのソース ファイルは、Azure Machine Learning サンプル リポジトリtrain フォルダーにあります。 このフォルダーには、コンポーネントを構築するファイルが 3 つあります。

  • train.py には、モデルをトレーニングするためのロジックが含まれています。
  • train_component.py はコンポーネントのインターフェイスを定義し、 train.py内の関数をインポートします。
  • conda.yaml は、コンポーネントのランタイム環境を定義します。

ロジックを含むスクリプトを取得する

train.py ファイルには、画像分類のために Keras ニューラル ネットワークをトレーニングするためのロジックを実行する通常の Python 関数が含まれています。 コードを確認するには、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_datainput_modeloutput_resultの 3 つのコマンドライン引数を受け取ります。 プログラムは、入力データを使用して入力モデルにスコアを付け、結果を出力します。

YAML を使用してコンポーネントを定義する

このセクションでは、有効な YAML コンポーネント仕様形式でコンポーネント仕様を作成する方法について説明します。 このファイルによって、次の情報が指定されます。

  • メタデータ。 名前、表示名、バージョン、種類など。
  • インターフェイス。 入力と出力。
  • コマンド、コード、環境。 コンポーネントの実行に使用されるコマンド、コード、および環境。
$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 関数で定義されているデータ準備コンポーネントとモデル トレーニング コンポーネントをインポートできます。

次のコードでは、prepare_data_component()関数とkeras_train_component()関数をそれぞれ、prep フォルダー内のprep_component.py ファイルと、train フォルダー内の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)

load_component()関数を使用して、YAML によって定義されるスコア コンポーネントを読み込むことができます。

# load component function from yaml
keras_score_component = load_component(source="./score/score.yaml")

パイプラインを構築する

パイプラインを構築するために、すべてのコンポーネントと入力データを作成して読み込んだ。 これで、それらをパイプラインに作成できるようになりました。

サーバーレス コンピューティングを使用するには、ファイルの先頭にfrom azure.ai.ml.entities import ResourceConfigurationを追加します。 そして、次のように置き換えます。

  • default_compute=cpu_compute_targetdefault_compute="serverless"
  • train_node.compute = gpu_compute_targettrain_node.resources = "ResourceConfiguration(instance_type="Standard_NC6s_v3",instance_count=2)
# 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_nodetrain_nodescore_nodeの 3 つのノードが含まれています。

  • input_dataprepare_data_node には pipeline_input_data の値が使われます。

  • train_nodeinput_data は、prepare_data_nodetraining_data 出力です。

  • score_nodeinput_dataprepare_data_nodetest_data出力であり、input_modeltrain_nodeoutput_modelです。

  • train_node CNN モデルをトレーニングするため、そのコンピューティングをgpu_compute_targetとして指定できます。 そうすることで、トレーニングのパフォーマンスを向上させることができます。

パイプライン ジョブを送信する

パイプラインを作成したら、ワークスペースにジョブを送信できます。 ジョブを送信するには、まずワークスペースに接続する必要があります。

ワークスペースにアクセスする

資格情報を構成する

DefaultAzureCredentialを使用してワークスペースにアクセスします。 DefaultAzureCredential を使うと、ほとんどの Azure SDK 認証シナリオに対応できます。

DefaultAzureCredentialが機能しない場合は、この資格情報の構成の例ID パッケージを参照してください。

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 サービスを管理する 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_datafashion_ds を使用します。

実験を送信するための呼び出しはすぐに完了します。 次のような出力が生成されます。

実験 名前 タイプ ステータス 詳細ページ
pipeline_samples sharp_pipe_4gvqx6h1fb パイプライン 準備中 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 でパイプラインをデバッグする方法の詳細については、「 Azure Machine Learning Studio を使用してパイプラインエラーをデバッグする」を参照してください。

(省略可能)ワークスペースにコンポーネントを登録する

前のセクションでは、画像分類タスクを完了するために 3 つのコンポーネントを使用してパイプラインを構築しました。 ワークスペース内でコンポーネントを共有および再利用できるように、コンポーネントをワークスペースに登録することもできます。 次の例は、データ準備コンポーネントを登録する方法を示しています。

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 から以前に読み込まれたコンポーネントを登録できます。

次の手順