【Keras】S3 + Lambda + EC2 で作る画像認識システム

今回は Amazon S3 + AWS Lambda + Amazon EC2 (Keras, Flask) で作る画像認識システムの例について書きます。

AWS構成

今回のAWS構成は以下。画像ファイルの S3 Put イベントをトリガにして Lambda から画像認識 API Server にアクセスし, 推論結果を S3 に保存。

API Gateway (Amazon Cognito) -> Lambda にすることで認証や監視機能を持たせた API サービスにも拡張できる。

画像認識 API Server

Amazon EC2 (Amazon Linux 2, t2.medium) に 画像認識 API Server を立てる。

今回は画像認識に Keras の学習済 VGG16 (ImageNet で学習した16層の CNN) モデル [1] を使うため, Keras 推論環境を構築する。 (Docker コンテナ化する方が望ましい)

$ wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
$ bash Miniconda3-latest-Linux-x86_64.sh
$ export PATH=/home/ec2-user/miniconda3/bin:$PATH
$ conda create -n tensorflow python=3.6 anaconda
$ source activate tensorflow
$ conda install keras

Wab Application Framework は Flask を使う。
HTTP POST で受け取った画像データをそのままモデルに渡し推論結果を返す。 画像は S3 にあり API Server では保存しないため EC2 のディスク容量はほぼ無視できる。

import io
import numpy as np
from keras.applications.vgg16 import VGG16, preprocess_input, decode_predictions
from keras.preprocessing import image
from PIL import Image
import tensorflow as tf
from flask import Flask, jsonify, request
from keras import backend as K

app = Flask(__name__)

def predict(img):
    K.clear_session()
    model = VGG16(weights='imagenet')
    x = image.img_to_array(img)
    x = np.expand_dims(x, axis=0)
    x = preprocess_input(x)

    global graph
    graph = tf.get_default_graph()
    with graph.as_default():
        preds = model.predict(x)

    return preds

@app.route('/v1/vgg16/classify', methods=['POST'])
def classify():
    print(request.headers)
    top_n = request.args.get('n')
    if top_n is None:
        top_n = 3

    bin_data = io.BytesIO(request.data)
    img = Image.open(bin_data)
    img = img.resize((224, 224))

    preds = predict(img)
    decoded_preds = decode_predictions(preds, top=top_n)
    results = [{'id': i[0], 'class': i[1], 'prob': float(i[2])} for i in decoded_preds[0]]
    return jsonify({'model': 'VGG16', 'prediction': results})

def main():
    app.run(debug=False, host='0.0.0.0', port=8080)

if __name__ == '__main__':
    main()

API Server を起動する。

(tensorflow) [ec2-user@ip-172-30-0-175 ~]$ python server.py
Using TensorFlow backend.
 * Serving Flask app "server" (lazy loading)
 * Environment: production
   WARNING: Do not use the development server in a production environment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)

AWS Lambda

S3 に一時的な画像アップロード用バケット, リネームした画像/推論結果を保存するバケットの2つを作成する。

次に Lambda 関数を作成する。関数のロールには S3, CloudWatch Logs のアクセス許可を与える。

S3 のアップロード用バケットにアップロードされた画像ファイルを HTTP POST で API Server に渡し推論結果を受け取り, 画像と共に保存用バケットに保存するコードを書く。

import time
import json
import boto3
import urllib.request
import urllib.parse
import hashlib

def lambda_handler(event, context):
    key =  event['Records'][0]['s3']['object']['key']

    s3 = boto3.resource('s3')
    obj = s3.Bucket('t2sy.upload-data').Object(key)
    response = obj.get()
    binary_data = response['Body'].read()

    host = 'ec2-xx-xxx-xxx-xxx.ap-northeast-1.compute.amazonaws.com'
    port = 8080
    url = 'http://' + host + ':' + str(port) + '/v1/vgg16/classify'
    headers = {'Content-Type': 'image/jpeg'}
    request = urllib.request.Request(url, data=binary_data, method='POST', headers=headers)
    with urllib.request.urlopen(request) as resp:
        resp_body = resp.read().decode("utf-8")
    
    resp_json = json.loads(resp_body)

    msg = (str(time.time())+key).encode("utf-8")
    hash = hashlib.sha256(msg).hexdigest()
    s3c = boto3.client('s3')
    s3c.copy_object(Bucket='t2sy.images',
        Key=hash+'_'+key,
        CopySource={'Bucket': 't2sy.upload-data', 'Key': key})
    s3.Object('t2sy.upload-data', key).delete()

    resp_obj = s3.Object('t2sy.images', hash+'_'+key+'_prediction')
    resp_json['file_name'] = key
    resp_json['hash'] = hash
    resp_obj.put(Body=bytearray(json.dumps(resp_json).encode("utf-8")))
    print(resp_json)

    return {
        'statusCode': 200,
        'body': json.dumps(resp_json)
    }

Lambda関数のテスト機能を使うとデバッグが捗る。今回はテストのイベントテンプレートは Amazon S3 Put を使用した。
続いて, 実際の画像を用いてテストする。試しに猫の画像を S3 アップロード用バケットにアップロードし CloudWatch Logs で結果を確認。

その時の API Server のアクセスログが以下。

...
Accept-Encoding: identity
Content-Length: 125682
Host: ec2-xx-xxx-xxx-xxx.ap-northeast-1.compute.amazonaws.com:8080
User-Agent: Python-urllib/3.6
Content-Type: image/jpeg
Connection: close

13.113.183.196 - - [11/Nov/2018 02:09:10] "POST /v1/vgg16/classify HTTP/1.1" 200 -

保存用バケットに以下のような推論結果が保存される。 tiger_cat と認識された。

{
    "model": "VGG16",
    "prediction": [
        {
            "class": "tiger_cat",
            "id": "n02123159",
            "prob": 0.9348850846290588
        },
        {
            "class": "tabby",
            "id": "n02123045",
            "prob": 0.05961435288190842
        },
        {
            "class": "Egyptian_cat",
            "id": "n02124075",
            "prob": 0.004736203700304031
        }
    ],
    "file_name": "example.jpg",
    "hash": "f5a07bda84bc8499ee12e362df7d54601d5100eb2ad6ad83c1a8cf368cddb7d9"
}

[1] ImageNet: VGGNet, ResNet, Inception, and Xception with Keras
[2] tensor is not an element of this graph. when loading model