Karbon Platform Services ~NutanixによるマルチクラウドPaaS~でシンプルなAI推論アプリケーションを構築してみる

# はじめに
本記事は[Nutanixアドベントカレンダー2020](https://adventar.org/calendars/4998)の12/2分記事となります。

## 概要
Nutanixは2020年9月、Karbon Platform Services(KPS)を発表しました。これはKubernetesをベースとしたマルチクラウドPaaSを実現するものです。プレスリリースと製品紹介は以下をご覧ください。

[プレスリリース](https://www.nutanix.com/jp/press-releases/2020/nutanix-launches-kubernetes-platform-as-a-service)
[Karbon Platform Services製品紹介ページ](https://www.nutanix.com/jp/products/karbon/platform-services)

# 本記事では
[前編](https://qiita.com/shocnt/items/e3ae4d837ae8957e6682)でサービスドメインと呼ばれるKubernetesクラスタの構築、[後編](https://qiita.com/shocnt/items/634a39f93504180a26c6)でマイクロサービスアプリケーションを構築しました。本記事ではKPSの機能群を駆使し、簡単なAI推論アプリケーションを構築します。[こちら](https://github.com/nutanix/karbon-platform-services/tree/master/projects/ai-inference)のGithubレポジトリのガイドに沿って進めていきます。

# アプリケーションデザイン
構築するアプリケーションについて説明します。ある画像のデータソースをKPSに入力し、KPS内で管理されたTensorflowモデルに掛け、推論結果を出力するというものです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/50e22cf1-c33a-aafe-1c3f-adde609e1316.png)

主なコンポーネントと使用する機能は以下です。
**MQTT Client**: ファッションアイテムの画像データを入力し、推論結果を受け取るMQTTクライアントのアプリケーションです。あなたのローカル環境(著者はMacOS Catalina環境にて実施)上にPythonスクリプトとして構築します。
**MQTT Subscriber/Publisher**: KPSのサービスドメインがビルトインで備えるMQTTの入力/出力のためのコンポーネントです。KPS上で`データソース`として設定します。
**Apparel Classifier**: 入力されたファッションアイテムを識別するためのPython関数です。KPSの`データパイプライン`上に`Function as a Service(FaaS)`機能を用いて実装します。
**ML Model**: 入力されたファッションアイテムを識別するために推論を行うTensorflowベースの機械学習モデルです。KPS上で`AI Inference`機能を用いて実装します。

# 構築手順
## 1. ローカル環境のセットアップ
### 1-1. Githubリポジトリのクローン
KPSのGithubリポジトリをクローンし、プロジェクトのディレクトリ(以下作業ディレクトリと記載します)に移動します。

```sh
git clone https://github.com/nutanix/karbon-platform-services.git
cd karbon-platform-services/projects/ai-inference
```

## 1-2. Pythonモジュールのインストール
本ガイドの実行にはPython3(筆者は3.6.1のpyenv環境にて実施)といくつかのPythonライブラリが必要となります。以下を実行します。

```sh
pip3 install -r requirements.txt
```

## 2. MQTTデータソースの設定
このセクションではKPSおよびローカル環境においてMQTTデータソース、MQTTクライアントの設定を行います。

### 2-1. KPSへのアクセス
[こちら](https://karbon.nutanix.com/)にアクセスし、My Nutanixアカウントでログインします。

### 2-2. カテゴリの作成
Administration - Categoriesに進み、"+Create"をクリックした後、以下を入力して"Update"をクリックします。

**Name**: MQTT
**Values**: ByteArray

![スクリーンショット 2020-12-02 17.59.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/5251447a-e952-1fa3-9aa4-697d9f798bc2.png)

KPSはこのカテゴリを各種リソースに属性情報として付与することでスケーラブルな運用管理を可能としています。

### 2-3. データソースの作成
Infrastructure - Data Sources and IoT Sensorsに進み、"+Add Data Source"をクリックします。以下入力します。

**Type**: Sensor
**Name**: mqtt-camera
**Associated Service Domain**: あなたのサービスドメイン
**Protocol**: MQTT
**Authentication**: Certificates

"+Generate Certificates"をクリックし、"+Download"をクリックして認証情報をダウンロードします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/77f4070e-81e3-2e56-ea9f-00925f25f64c.png)

また、各ファイルを以下のように名前変更し、作業ディレクトリ配下のcertsディレクトリ内に配置します。
xxxxxxxxxxxxx_CACertificate.crt --> `ca.crt`
xxxxxxxxxxxxx_certificate.crt --> `client.crt`
xxxxxxxxxxxxx_privateKey.key --> `client.key`

```sh
mkdir certs/
mv ca.crt client.crt client.key certs/
```

"Next"をクリックし、"Add New Field"をクリックした後、以下入力します。

**Name**: apparel_images
**MQTT Topic**: apparel_images

![スクリーンショット 2020-12-02 18.14.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/472cb01b-7050-7121-4943-e64d5059bfd8.png)

"Next"をクリックし、以下入力し、"Add"をクリックします。

**Attribute**: MQTT ByteArrayをそれぞれドロップダウンから選択

![スクリーンショット 2020-12-02 18.17.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/dac12b0d-824c-747d-1825-40d10b7d12e8.png)

以上によりKPS側でMQTTデータを受け取る準備が出来ました。

### 2-4. MQTTクライアントの作成
本ガイドでは以下PythonコードによるMQTTクライアントを使用します。`あなたのサービスドメインのIPアドレス`部分を環境に合わせてチューニングして下さい。

このクライアントは[Keras Fashion MNIST dataset](https://keras.io/api/datasets/fashion_mnist/)を用いてMQTTデータソースの指定したトピックに画像データを送信します。

```python:mqtt_client.py
import ssl
import time
import random
import msgpack
from tensorflow import keras
import paho.mqtt.client as mqttClient


def on_connect(client, userdata, flags, rc):
	if rc == 0:
		print("Connected to broker")
		global Connected                #Use global variable
		Connected = True                #Signal connection 
	else:
		print("Connection failed")

def on_publish(client, userdata, result):
	print("Published!")

def on_message(client, userdata, message):
    print("New message received!")
    print("Topic:", message.topic)
    print("Message:", message.payload)


def main():
	global Connected
	Connected = False
	# IP address of the edge. Modify this.
	broker_address= "あなたのサービスドメインのIPアドレス"
	port = 1883
	# NOTE: For data pipelines to receive MQTT messages, topic should
	#       be the same as that specified when creating the MQTT datasource.
	input_topic = "apparel_images"
	output_topic = "apparel-predict"
	client = mqttClient.Client()
	# Set callbacks for connection event, publish event and message receive event
	client.on_connect = on_connect
	client.on_publish = on_publish
	client.on_message = on_message
	client.tls_set(ca_certs="certs/ca.crt", certfile="certs/client.crt", keyfile="certs/client.key", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None)
	# Set this to ignore hostname only. TLS is still valid with this setting.
	client.tls_insecure_set(True)
	client.connect(broker_address, port=port)
	client.subscribe(output_topic)
	client.loop_start()

	# Wait for connection
	while Connected != True:
		print("Connecting...")
		time.sleep(1)

	fashion_mnist = keras.datasets.fashion_mnist
	(images, image_labels), (test, labels) = fashion_mnist.load_data()

	# scale the values to 0.0 to 1.0
	images = images / 255.0
	# reshape for feeding into the model
	images = images.reshape(images.shape[0], 28, 28, 1)

	for i in range(10):
		print(f'\nCounter: {i+1}')
		img = random.choice(images)
		print(img.shape)
		payload ={}
		payload['image'] = img.tolist()
		payload['dtype'] = str(images.dtype)
		payload['height'] = 28
		payload['width'] = 28
		try:
			client.publish(input_topic, msgpack.packb(payload))
			time.sleep(5)
		except KeyboardInterrupt:
			client.disconnect()
			client.loop_stop()

if __name__ == "__main__":
	 main()

```

これによりローカル環境からMQTTデータを送信し、推論結果を受け取る準備ができました。

## 3. 機械学習モデル(MLモデル)の構築
このセクションではKPSの`AI Inferencing Service`を用いてMLモデルをKPSにデプロイします。AI Inferencing ServiceではMLモデルのバージョン管理およびAI推論ランタイム(GPU等のハードウェアアクセラレータを抽象化したもの)提供を行います。本ガイドではTensorflowのフレームワークを用いたシンプルなMLモデルを構築し、KPSに登録します。ローカル環境で行いますが、GPUは不要です。

### 3-1. プロジェクトの作成
今回のアプリケーションをデプロイするアプリケーションを作成します。[こちら](https://qiita.com/drafts/634a39f93504180a26c6)の"1. プロジェクトの設定"を参考に作成します。ただし、このプロジェクトで有効化するプラットフォーム機能については
AI Inferencing Serviceはデフォルトで有効化されており、かつTraefik/Istio/Kafka/Prometheusの有効化は不要です。

### 3-2. TensorflowのMLモデルの構築
以下のPythonスクリプトにより機械学習モデルを構築します。このスクリプトでは主に以下2つのタスクを行っています。

- 異なるカテゴリのファッションアイテムの画像を識別するための畳み込みニューラルネットワークを訓練する
- TensorflowのSavedModelフォーマットでモデルを保存する

```python:model_trainer.py
import tensorflow as tf
from tensorflow import keras

# Helper library
import numpy as np

# dataset used to train model
fashion_mnist = keras.datasets.fashion_mnist
(train_images, train_labels), (test_images, test_labels) = fashion_mnist.load_data()

# scale the values to 0.0 to 1.0
train_images = train_images / 255.0
test_images = test_images / 255.0

# reshape for feeding into the model
train_images = train_images.reshape(train_images.shape[0], 28, 28, 1)
test_images = test_images.reshape(test_images.shape[0], 28, 28, 1)

# class labels
class_names = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat',
               'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']

print('\ntrain_images.shape: {}, of {}'.format(train_images.shape, train_images.dtype))
print('test_images.shape: {}, of {}'.format(test_images.shape, test_images.dtype))

# neural network architecture
model = keras.Sequential([
  keras.layers.Conv2D(input_shape=(28,28,1), filters=8, kernel_size=3, 
                      strides=2, activation='relu', name='Conv1'),
  keras.layers.Flatten(),
  keras.layers.Dense(10, activation=tf.nn.softmax, name='Softmax')
])
model.summary()

testing = False
epochs = 5

model.compile(optimizer='adam', 
              loss='sparse_categorical_crossentropy',
              metrics=['accuracy'])
model.fit(train_images, train_labels, epochs=epochs)

test_loss, test_acc = model.evaluate(test_images, test_labels)
print('\nTest accuracy: {}'.format(test_acc))

tf.saved_model.save(model, 'fashion_classifier')
print('\nSaved model:')
```

作業ディレクトリにてこちらのPythonスクリプトを実行します。

```sh
python3 model_trainer.py 
```

作業ディレクトリ配下に`fashion_classifier`ディレクトリが作成されます。この中に構築されたMLモデルが格納されています。これをzipで固めます。Githubレポジトリに既にあるfashion_classifier.zipについては不要のため削除しています。

```sh
rm fashion_classifier.zip
zip -r fashion_classifier.zip fashion_classifier/saved_model.pb fashion_classifier/variables/
```

### 3-3. MLモデルの登録
左上のドロップダウンメニューから3-1で作成したプロジェクトに移動します。左側のナビゲーションペインから**AI Inference**をクリックします。

以下を入力し、"+Add Version"をクリックします。

**Name**: fashion_classifier
**Description**: 任意の情報を入力
**Framework Type**: Tensorflow 2.1.0

![スクリーンショット 2020-12-02 18.52.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/cd90d417-8dac-aabc-a487-579b7817a223.png)

以下入力します。

**ML Model Version**: 1

"ファイルを選択"した上で、3-2で作成した`fashion_classifier.zip`を選択します。
"Upload"をクリックし、モデルのアップロードが終わったら"Done"をクリックします。

![スクリーンショット 2020-12-02 18.56.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/dbdcd16e-7b6b-c257-e99f-0bd5f9d64908.png)

以上で、MLモデルのローカル環境での構築、KPSへの登録が完了しました。

## 4. データパイプラインの作成
最後のステップです。KPSにおいてETL処理を行う`Function as a Service`と`データパイプライン`の作成を行います。

### 4-1. ファンクションの登録
`Function as a Service`機能ではPython, Go, Node．ｊｓのようなデータ処理言語が使用できます。本ガイドでは以下のPythonスクリプトを登録します。このスクリプトは以下のタスクを行っています。

- MQTTデータソースからの入力を取得
- インプット画像に対する推論を行う
- 識別されたクラスと自信度をログする
- MQTTトピックにメッセージをPublishする

```python:apparel_classifier.py
import os
import json
import logging
import requests
import msgpack
import numpy as np

logging.getLogger().setLevel(logging.INFO)
ai_inference_endpoint = os.environ['AI_INFERENCE_ENDPOINT']
logging.info("Apperal Classifier")

# class labels 
class_map = {
        0 : 'T-shirt/top', 
        1 : 'Trouser', 
        2 : 'Pullover', 
        3 : 'Dress', 
        4 : 'Coat',
        5 : 'Sandal', 
        6 : 'Shirt', 
        7 : 'Sneaker', 
        8 : 'Bag', 
        9 : 'Ankle boot'
}

def predict(image):
    data = json.dumps({"signature_name": "serving_default",
                       "instances": [image]})
    headers = {"content-type": "application/json"}
    model_name = "fashion_classifier"
    model_version = 1
    url = "http://%s/v1/models/%s/versions/%d:infer" % (ai_inference_endpoint, model_name, model_version)
    response = requests.post(url, data=data, headers=headers)
    if response.status_code != 200:
        logging.error(response.json())
        return None
    inference_payload = json.loads(response.content)
    pred_array = inference_payload['predictions'][0]
    pred_index = pred_array.index(max(pred_array))
    prediction = class_map[pred_index]
    return prediction, max(pred_array)

def main(ctx, msg):
	payload = msgpack.unpackb(msg)
	img = payload['image']
	prediction, confidence = predict(img)
	output = "Prediction: {}, Confidence: {}".format(prediction, str(confidence))
	logging.info(output)
	ctx.send(msg)
	return
```

Pythonスクリプトの以下部分に注目で、本部分により3で登録した`AI Inferencing`機能のfashion_classifierというMLモデルのバージョン1を参照することになります。

```
ai_inference_endpoint = os.environ['AI_INFERENCE_ENDPOINT']
model_name = "fashion_classifier"
model_version = 1
url = "http://%s/v1/models/%s/versions/%d:infer" % (ai_inference_endpoint, model_name, model_version)
```

左側のナビゲーションペインから**Functions and Data Pipelines**をクリックし、画面上部のタブから**Functions**をクリックします。

"+Create"をクリックし、以下入力し、"Next"をクリックします。

**Name**: apparel classifier
**Language**: python
**Runtime Environment**: Tensorflow Python

"ファイルを選択"し、作業ディレクトリにあるapparel_classifier.pyを選択し、"Create"をクリックします。

![スクリーンショット 2020-12-02 20.28.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/378a5d03-db1a-8fd3-b88e-ba9d2952bc3b.png)

![スクリーンショット 2020-12-02 20.30.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/d31aa89c-0a15-3326-1414-a7ec0c4d10ee.png)

### 4-2. データパイプラインの登録
画面上部のタブから**Data Pipelines**をクリックし、"+Create"をクリックします。

以下入力し、"+Create"をクリックします。

**Data Pipeline Name**: fashion-pipeline
**Input**:
- Data Sourceを選択
- **Select by Categories**: MQTT:ByteArray
**Transformation**: apparel classifier
**Output**: 
- **Destination**: Service Domain
- **Endpoint Type**: MQTT
- **Endpoint Name**: apparel-predict

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134931/f3b6cf4a-f25a-2d14-4d9a-c6f723e99510.png)

これでデータソース、MLモデル、Function、データパイプラインが作成され、すべてがKPSのサービスドメインにコンテナとしてデプロイされています。

# 5. 試してみる
ローカル環境から`mqtt_client.py`を実行し、KPSへのMQTTデータの送信と、推論結果の受信を行います。

```sh
python mqtt_client.py
```

以下のような出力が得られれば期待通りの結果となります。Messageとして推論結果と自信度が得られていることが分かります。

```sh:出力例
Connecting...
Connected to broker

Counter: 1
(28, 28, 1)
Published!
New message received!
Topic: apparel-predict
Message: b'Prediction: Sandal, Confidence: 0.96976918'

Counter: 2
(28, 28, 1)
Published!
New message received!
Topic: apparel-predict
Message: b'Prediction: Shirt, Confidence: 0.539910078'

Counter: 3
(28, 28, 1)
Published!
New message received!
Topic: apparel-predict
Message: b'Prediction: Dress, Confidence: 0.873085737'
```

# まとめ
Karbon Platform Services(KPS)の得意分野であるデータパイプラインやFaaS、AI機能についてご紹介しました。各種ユーティリティを各自で構築することなく、サービスとして利用することでアプリケーション開発における生産性の向上、サービス開発の迅速化を実現します。