AzureMLで機械学習/推論を実行する（データ登録、コンピューティングクラスタでの学習、AKSへのデプロイ）

#目的
AzureMLでデータを準備する方法と、それをAKSにデプロイ、使用する方法についてまとめました。
基本的には、Microsoftのチュートリアル通りです。

#準備編

##Workspaceへの接続
まずは、AzureMLのWorkspaceに接続します。

import azureml.core
from azureml.core import Workspace

# check core SDK version number
print("Azure ML SDK Version: ", azureml.core.VERSION)

ws = Workspace.from_config()
print(ws.name, ws.location, ws.resource_group, sep='\t')

ちなみにWorkspace接続は、以下のInteractiveLoginAuthenticationからでも可能です。

from azureml.core.authentication import InteractiveLoginAuthentication

interactive_auth = InteractiveLoginAuthentication(tenant_id="テナントID")

ws = Workspace(subscription_id="サブスクリプションID",
               resource_group="リソースグループ名",
               workspace_name="ワークスペース名",
               auth=interactive_auth)

この際、デバイスログイン( https://microsoft.com/devicelogin )によるAzure ADへの認証が要求される場合があります。いつまでたっても接続できないときは、ちゃんと出力ログをチェックしましょう。
(筆者は、最初、気づかずになかなかできないなぁ、とずっと待っていました^^;)
デバイスログインを避けるには、Service principalを使用したRBACによる自動ログインもできます。この話は別途、記述予定です。

#データの準備
次は、学習に使うデータの準備です。
AzureMLでは、学習に使うデータはDatasetsにデータの場所を登録する必要があります。
Datasetsとして、利用可能なデータの場所は、"ローカルデータ"、"Datastoreに格納したデータ"、"任意のウェブ上のデータ"、"オープンデータセット"が利用可能です。
"Datastoreに格納したデータ"以外はあえて説明する必要もないので、説明は割愛します。
ここでは、pythonからローカルに準備したデータをDatastoreへ格納、Datasetsに登録する方法を示します。

from azureml.core import Datastore, Dataset
import os
import glob

# workspaceblobstoreというデータストアを取得
# workspaceblobstoreはAzureMLを作成した際に自動的に作成されているはずです
datastore = Datastore.get(ws, datastore_name='workspaceblobstore')

target_path = 'data/train' # Datastore内でのデータの格納パス
# ローカルのtrainディレクトリ内のpngファイルをDatastoreへアップロード
datastore.upload_files(files=glob.glob(os.path.join('train', '*.png')),
                       target_path=target_path,
                       overwrite=True,
                       show_progress=True)

# データセットをDatastore上のパスから選択して登録
dataset = Dataset.File.from_files(path=[(datastore, target_path)])
print(dataset.to_path())

##学習用のコンピューティングクラスタにアタッチ
コンピューティングクラスタとは、学習の時だけに立ち上げるコンピューティング環境です。
Tesla M60などの超高性能なGPUも利用可能です。稼働時間を学習時だけに限定できるので、コストを抑えることができます。
新たにコンピューティングクラスタを作成する際は、Pythonからもできますが、グラフィカルに比較しながら評価できるAzure Machine Learning StudioのGUIから作成したほうがベターです。

PriorityをLow priorityにすると劇的にコストが下がります。
Standard_NV6 (Nvidia Tesla M60)の場合:
Dedicated 1.58$/hr ⇒ Low priority 0.32$/hr
Low priorityに設定すると他のAzureML使用者が利用開始した場合に学習の実行が待たされる場合もあるようです。
ただ、高価なGPUをお安く利用できるメリットは計り知れないですね...

既存のコンピューティングクラスタへアタッチは以下の通りです。

from azureml.core.compute import AmlCompute
from azureml.core.compute import ComputeTarget

compute_name = os.environ.get("AML_COMPUTE_CLUSTER_NAME", "作成したクラスタ名")

if compute_name in ws.compute_targets:
    compute_target = ws.compute_targets[compute_name]
    if compute_target and type(compute_target) is AmlCompute:
        print("found compute target: " + compute_name)
else:
    # クラスタが作成されていない場合の処理を記載

#学習の実行

##Experimentの設定

学習の前に学習の実行を管理するExperimentを作成または設定する必要があります。
適当なExperiment名を設定しましょう。

from azureml.core import Experiment

exp_name = 'my_exp'
exp = Experiment(workspace=ws, name=exp_name)

##コンピューティングクラスタでの学習
コンピューティングクラスタでの学習は、Anaconda仮想環境がインストールされたDockerイメージを作成して実行しています。
Dockerイメージの作成は、SKLearnTensorFlow Estimatorなどのライブラリによって、自動的に作成されるので、Dockerの知識がなくとも問題ありません。
学習の実行で主にやることは、学習実行スクリプト、必要なPython/condaパッケージ指定となります。
Tensorflow Estimatorを使った場合の例を示します。

from azureml.train.dnn import TensorFlow
from azureml.core import Workspace, Datastore, Dataset

# train.pyへの入力引数定義
script_params = {
    '--data-path': 'data/train',
    '--batch-size': 10,
    '--epoch': 100
}

est = TensorFlow(
    source_directory='scripts',                   # ローカルのscripts以下のスクリプトを使用
    entry_script='train.py',                      # 学習実行のエントリーポイントとなるスクリプト
    script_params=script_params                   # train.pyへの入力引数
    compute_target=compute_target,                # コンピューティングクラスタインスタンス
    conda_packages=['mesa-libgl-cos6-x86_64', 'opencv=3.4.2'], # インストールするcondaパッケージ
    pip_packages=['keras<=2.3.1', 'scikit-learn'] # インストール pipパッケージ
)

# 学習に必要な設定を登録
run = exp.submit(est)
# 学習の実行
run.wait_for_completion(show_output=True)

画像前処理としてOpenCVを使うケースがあるかと思いますが、OpenCVを使用するためには、libGL.so.1が必要となるため、condaパッケージからインストールすることになります。
実行時にDockerイメージの作成をキャッシュなしで毎回行うので、実行開始までしばらく時間を要します。
このイメージの作成時間を短縮するためにキュレーションされた環境も利用できるようです。今後は、こちらが主流になりそうですね。
選別された環境を使用する
Azure Machine Learning のキュレーションされた環境

##学習用スクリプト
学習用スクリプト(train.py)は以下のように書きます。
入力として引数を受け付ける部分と出力を特殊ディレクトリ(outputs)を指定する以外に変わったところはないです。

train.py

import argparse
import numpy as np
import os
import glob
import cv2
import keras
import re
from keras.callbacks import Callback, ModelCheckpoint
from keras.layers import *
from keras.models import Model
from keras import losses
from keras.utils.np_utils import to_categorical
from sklearn.model_selection import train_test_split

# Azure ML上で実行の結果をプロットするのに必要
import matplotlib.pyplot as plt

from azureml.core import Run

# 引数に指定した値を取得
parser = argparse.ArgumentParser()

parser.add_argument('--data-path', type=str, dest='data_path', default='data_path', help='target data path')
parser.add_argument('--batch-size', type=int, dest='batch_size', default=10, help='mini batch size for training')
parser.add_argument('--epoch', type=int, dest='epoch', default=40, help='number of epochs')

# 変数data_path, batch_size, epochに目的の値が格納される
args = parser.parse_args()

data_path = args.data_path
print('training dataset is stored here:', data_path)

# 画像データの読み込み
train_files = glob.glob(os.path.join(data_path, '*.png'))
train_imgs = [cv2.imread(path) for path in train_files]
# 画像ファイル名から正解ラベルを指定 [ラベル]_0.png, [ラベル]_1_.pngなどを想定
targets = [os.path.basename(file).split('_')[0] for file in train_files]
targets = to_categorical(targets, num_classes = len(np.unique(targets))

# 画像のshapeをkeras向けに変換
train_imgs = np.reshape(train_imgs, (train_imgs.shape[0], train_imgs.shape[1], train_imgs.shape[2], 1))


# 学習用データと訓練データの分割
x_train, x_test, y_train, y_test = train_test_split(train_imgs, targets, test_size=0.1)

n_epochs = args.epoch
batch_size = args.batch_size

# TensorFlowのサイト通りのネットワーク(https://www.tensorflow.org/tutorials/images/cnn?hl=ja)
model = models.Sequential()
model.add(layers.Conv2D(32, (3, 3), activation='relu', input_shape=(28, 28, 1)))
model.add(layers.MaxPooling2D((2, 2)))
model.add(layers.Conv2D(64, (3, 3), activation='relu'))
model.add(layers.MaxPooling2D((2, 2)))
model.add(layers.Conv2D(64, (3, 3), activation='relu'))
model.add(layers.Flatten())
model.add(layers.Dense(64, activation='relu'))
model.add(layers.Dense(10, activation='softmax'))
model.summary()
model.compile(optimizer='adam', loss=loss='sparse_categorical_crossentropy')

# 学習の実行
run = Run.get_context()

# outputsディレクトリにモデルが保存されます
# outputsは特殊ディレクトリで、AzureML向けに作成されたのモデル保存用のAzure永続化ボリュームをマウントしています。
# この例ではModelCheckPointを使って、バリデーションロスが更新された場合のモデルを保存しています
history = model.fit(x_train, y_train, 
                    batch_size=batch_size, 
                    epochs=n_epochs, 
                    callbacks=[ModelCheckpoint('outputs/model/weights.{epoch:03d}-{val_loss:.3f}.hdf5', monitor='val_loss', save_best_only=True, period=10)],
                    validation_data=(x_test, y_test))

score = model.evaluate(x_test, y_test, verbose=0)

# log a single value
run.log("Final test loss", score)

plt.figure(figsize=(6, 3))
plt.plot(history.history['val_loss'], 'r--', label='Loss', lw=4, alpha=0.5)
plt.legend(fontsize=12)
plt.grid(True)
# log an image
run.log_image('Accuracy vs Loss', plot=plt)

# serialize NN architecture to JSON
model_json = model.to_json()
# モデルそのものも保存
with open('outputs/model/model.json', 'w') as f:
    f.write(model_json)
# 最終エポック時のモデルも保存
model.save('outputs/model/model.h5')

#モデルの登録
学習が完了した後は、デプロイのためのモデルの登録です。

# 最終の重みを'cnn_model'というモデル名として登録
model = run.register_model(model_name='cnn_model', model_path='outputs/model/
model.h5')

学習が終わったモデルは、以下のコマンドでダウンロードして使用も可能です。

model_path = Model(ws, 'cnn_model').download(target_dir=os.getcwd(), exist_ok=True)

あとは通常通り推論に使用できます。

モデルのデプロイ

AzureMLでサポートされているモデルのデプロイ先は、ACI(Azure Container Instances)かAKS(Azure Kubernetes Service)のどちらかになります。
ACIは、Web APIの呼び出し時だけコンテナを起動して結果を返すサーバーレスサービスです。そのため、コストはかなり抑えられますが、重みのロードに時間がかかる画像処理系のサービスではオーバーヘッドが大きくなるため向いていません。
AKSは常時起動する仮想マシンクラスタです。常時起動のためコストはかかりますが、オーバーヘッドは小さくなります。

今回は画像解析系のサービスを動かすのでAKS一択です。

##デプロイイメージの環境設定

デプロイ先の実行イメージ作成に必要な環境に必要なpip/anacondaのライブラリを定義したyamlファイルを作成します。
CondaDependenciesライブラリを使用して作成します。

from azureml.core.conda_dependencies import CondaDependencies 

myenv = CondaDependencies()
myenv.add_pip_package("azureml-defaults")
myenv.add_pip_package("scikit-learn==0.22.1")
myenv.add_pip_package("tensorflow==2.2.0")
myenv.add_pip_package("joblib")
myenv.add_pip_package("keras")
myenv.add_pip_package("scipy")

myenv.add_conda_package('opencv=3.4.2')
myenv.add_conda_package('mesa-libgl-cos6-x86_64')

with open("myenv.yml","w") as f:
    f.write(myenv.serialize_to_string())

元のdockerイメージの相性の問題で、ライブラリごとにcondaとpipどちらでインストールしたらいいかはトライ＆エラーが必要になるかもしれません。
少なくともOpenCVを含める場合はcondaでインストールしましょう。

出力されたmyenv.ymlは以下のようになります。

myenv.yml

# Conda environment specification. The dependencies defined in this file will
# be automatically provisioned for runs with userManagedDependencies=False.

# Details about the Conda environment file format:
# https://conda.io/docs/user-guide/tasks/manage-environments.html#create-env-file-manually

name: project_environment
dependencies:
  # The python interpreter version.
  # Currently Azure ML only supports 3.5.2 and later.
- python=3.6.2

- pip:
  - tensorflow==2.1.0
  - azureml-defaults
  - joblib
  - keras==2.3.1
- opencv=3.4.2
- mesa-libgl-cos6-x86_64
channels:
- anaconda
- conda-forge

#AKSへの接続
デプロイ先のAKSへ接続します。
この中でKubernetes内でこのサービスが利用するリソース(CPU数、メモリ量)を定義します。
また、HTTPSアクセスにしたい場合は、証明書の設定も行います。
自前で準備するのが大変であれば、Microsoftが証明書を自動的に準備してくれます。
自前の証明書を使用することも可能です。

from azureml.core.webservice import AksWebservice, Webservice
from azureml.core.compute import AksCompute, ComputeTarget
from azureml.core.compute_target import ComputeTargetException

resource_group = 'リソースグループ名'
cluster_name = 'クラスタ名'
name = 'サービス名'

attach_config = AksCompute.attach_configuration(resource_group=resource_group, cluster_name=cluster_name)

# microsoft発行のSSHを有効化する場合
ssl_enable = True

if ssl_enable:
    domain_label = 'azureml-test-deploy'
    provisioning_config = AksCompute.provisioning_configuration()
    provisioning_config.enable_ssl(leaf_domain_label=domain_label)
    attach_config.enable_ssl(leaf_domain_label=domain_label)

try:
    aks_target = ComputeTarget(workspace=ws, name=name)
    # すでにサービス名が存在してる場合(モデルをアップデートする場合)
    print('found exsiting')
except ComputeTargetException:
    # サービス名が存在しない場合は新規に作成して、アタッチ
    print('not found')
    aks_target = ComputeTarget.attach(ws, name, attach_config)

# デプロイの構成を設定
aksconfig = AksWebservice.deploy_configuration(
                cpu_cores=1,                    # 使用するCPUコア数
                memory_gb=1,                    # 最大メモリ量
                tags={"tag1": "任意のタグ値"},   # 任意のタグ指定
                description='デプロイサンプル',  # 任意の説明
# 以下は、自前のTLS証明を使用する場合
#               ssl_enabled=True,
#               ssl_cert_pen_file='cert.pem',
#               ssl_key_pem_file='key.pem',
#               ssl_cname='ドメイン名'
                )

##デプロイ
事前に用意したデプロイイメージの環境設定と、推論時に実行するスクリプトを指定して、モデルのデプロイを行います。

from azureml.core.webservice import AksWebservice, Webservice
from azureml.core.model import Model
from azureml.core.model import InferenceConfig
from azureml.core.environment import Environment

# イメージの環境設定
myenv = Environment.from_conda_specification(name="myenv", file_path="myenv.yml")
# 推論用スクリプト
inference_config = InferenceConfig(entry_script="scripts/score.py", environment=myenv)

# デプロイ
service = Model.deploy(ws, name, [model], inference_config, aksconfig, aks_target, overwrite=True)
service.wait_for_deployment(show_output = True)
print(service.state)
print(service.get_logs())

##推論スクリプト
推論スクリプトにはデプロイ直後に1度だけ実行されるinit()とリクエストごとに実行するrun()を実装します。
init()は1度しか実行されないので、ここでモデルをメモリのロードしておくことで、リクエスト時の処理を高速化できます。
ここでは、rawhttpを使って、生の画像バイナリデータを受け付ける特殊な場合です。

import json
import numpy as np
import os
import keras
import cv2
from azureml.core.model import Model
from azureml.contrib.services.aml_request import AMLRequest, rawhttp
from azureml.contrib.services.aml_response import AMLResponse

def init():
    global model
    # 登録したモデルを指定
    model_path = Model.get_model_path(model_name = 'cnn_model')
    model = keras.models.load_model(model_path)
    
# 画像の生データを受け付ける場合
@rawhttp
def run(request):
    try:
        if request.method == 'GET':
            respBody = str.encode(request.full_path)
            return AMLResponse(respBody, 200)
        elif request.method == 'POST':
            data = request.get_data(False)
            data = np.frombuffer(data, dtype=np.uint8)
            img = cv2.imdecode(data, cv2.IMREAD_UNCHANGED)            
            result = model.predict(img)
            return AMLResponse(result, 200)
        else:
            return AMLResponse('bad request', 500)
        
    except Exception as e:
        result = str(e)
        return {'error': result}
    return 

##デプロイしたモデルを使用する

デプロイに成功するとAzure Machine Learning StudioでEndpointが確認できるようになります。

このEndpointsに記載されているREST endpointとprimary keyもしくはsecondary keyを使ってアクセスします。
Swaggerに関するエラーが出力されていますが、Swaggerを定義していなくとも利用自体は問題ありません。ただ、定義しておくと通常のRESTfulなAPIとして利用が可能となり、PowerBI等からアクセスすることが可能となります。これは別途、記事を書きたいと思います。

以下にpythonでの使用例を示します。
この場合もやはり、生の画像データを送信する特殊なパターンです。
jsonデータをやり取りする場合は、headerのContent-Typeをapplication/jsonにすればよいです。

import requests

key1 = 'primary key'
key2 = 'secondary key'
endpoint = 'https://ドメイン名.japaneast.cloudapp.azure.com:443/api/v1/service/サービス名/score'
headers = {'Content-Type':'application/octet-stream', 'Authorization':('Bearer ' + key1)}

with open(filename, 'rb') as f:
    img = f.read()

response = requests.post(endpoint, headers=headers, data=img)

結構長くなってしまいましたが、これでAzure MLの一通りの操作ができるはずです。
お疲れ様でした。

##参考文献
Azure Machine Learning ワークスペースを作成して管理する
機械学習モデルを Azure にデプロイする
Web サービスとしてデプロイされた Azure Machine Learning モデルを使用する