基于Keras和Gunicorn+Flask部署深度学习模型

本文主要记录在进行Flask部署过程当中所使用的流程，遇到的问题以及相应的解决方案。

一、项目简介

该部分简要介绍一下前一段时间所作的工做：

• 基于深度学习实现一个简单的图像分类问题
• 借助flask框架将其部署到web应用中
• 并发要求较高

这是第一次进行深度学习模型的web应用部署，在整个过程当中，进一步折射出之前知识面之窄，在不断的入坑、解坑中实现一版。

二、项目流程

这部分从项目实施的流程入手，记录所作的工做及用到的工具。

2.1 图像分类模型

1. 模型的选择

须要进行图像分类，第一反应是利用较为成熟与经典的分类网络结构，如VGG系列（VGG16, VGG19），ResNet系列（如ResNet50），InceptionV3等。

考虑到是对未知类型的图像进行分类，且没有直接可用的训练数据，所以使用在Imagenet上训练好的预训练模型，基本知足要求。

若是对性能（耗时）要求较为严格，则建议使用深度较浅的网络结构，如VGG16, MobileNet等。

其中，MobileNet网络是为移动端和嵌入式端深度学习应用设计的网络，使得在cpu上也能达到理想的速度要求。是一种轻量级的深度网络结构。

MobileNet 由 Google 团队提出，发表于 CVPR-2017，论文标题： 《MobileNets: Efficient Convolutional Neural Networks for Mobile Vision Applications》

2. 框架选择

• 平时使用Keras框架比较多，Keras底层库使用Theano或Tensorflow，也称为Keras的后端。Keras是在Tensorflow基础上构建的高层API，比Tensorflow更容易上手。

• 上述提到的分类网络，在Keras中基本已经实现，Keras中已经实现的网络结构以下所示：

  

• 使用方便，直接导入便可，以下：

  

所以，选择Keras做为深度学习框架。

3. 代码示例

以Keras框架，VGG16网络为例，进行图像分类。

from keras.models import Model
from keras.applications.vgg16 import VGG16, preprocess_input
import keras.backend.tensorflow_backend as KTF
import tensorflow as tf
os.environ["CUDA_VISIBLE_DEVICES"] = "0,1" #使用GPU
# 按需占用GPU显存
gpu_options = tf.GPUOptions(allow_growth=True)
sess = tf.Session(config=tf.ConfigProto(gpu_options=gpu_options))
KTF.set_session(sess)

# 构建model
base_model = VGG16(weights=‘imagenet’, include_top=True)
model = Model(inputs=base_model.input,
outputs=base_model.get_layer(layer).output) # 获取指定层的输出值，layer为层名

# 进行预测
img = load_image(img_name, target_size=(224, 224))  # 加载图片并resize成224x224

# 图像预处理
x = image.img_to_array(img)
x = np.expand_dims(x, axis=0)
x = preprocess_input(x) 

feature = model.predict(x) # 提取特征
复制代码

2.2 模型性能测试

将分类模型跑通后，咱们须要测试他们的性能，如耗时、CPU占用率、内存占用以及GPU显存占用率等。

1. 耗时

耗时是为了测试图像进行分类特征提取时所用的时间，包括图像预处理时间和模型预测时间的总和。

# 使用python中的time模块
import time
t0 = time.time()
....
图像处理和特征提取
....

print(time.time()-t0) #耗时，以秒为单位
复制代码

2. GPU显存占用

使用英伟达命令行nvidia-smi能够查看显存占用。

3. CPU, MEM占用

使用top命令或htop命令查看CPU占用率以及内存占用率。

内存占用还可使用free命令来查看：

• free -h : 加上-h选项，输出结果较为友好，会给出合适单位

• 须要持续观察内存情况时，可使用-s选项指定间隔的秒数： free -h -s 3（每隔3秒更新一次，中止更新时按下Ctrl+c）

Ubuntu 16.04版本中默认的free版本有bug，使用-s选项时会报错。

根据以上三个测试结果适时调整所采用的网络结构及显存占用选项。

命令具体含义可参考博文：

Linux查看CPU和内存使用状况

2.3 Redis使用

Redis=Remote DIctionary Server，是一个由Salvatore Sanfilippo写的高性能的key-value存储系统。Redis是一个开源的使用ANSI C语言编写、遵照BSD协议、支持网络、可基于内存亦可持久化的日执行、key-value数据库，并提供多种语言的API。

Redis支持存储的类型有string, list, set, zset和hash，在处理大规模数据读写的场景下运用比较多。

1. 基本使用

安装redis

pip install redis

# 测试
import redis
复制代码

基本介绍

redis.py提供了两个类：Redis，StrictRedis用于实现Redis的命令 StrictRedis用于实现大部分官方命令，并使用官方的语法和命令 Redis是StrictRedis的子类，用于向前兼容redis.py 通常状况下咱们就是用StrictRedis。

使用示例

# 1. 导入redis
from redis import StrictRedis

# 2. 链接数据库，指定host,端口号，数据库
r = StrictRedis(host=‘localhost’, port=6379, db=2)

# 3. 存储到redis中
r.set('test1', 'value1')  # 单个数据存储
r.set('test2', 'value2')

# 4. 从redis中获取值
r.get('test1')

# 5. 批量操做
r.mset(k1='v1', k2='v2')
r.mset({'k1':'v1', 'k2':'v2'})
r.mget('k1', 'k2')
r.mget(['k1', 'k2'])
复制代码

2. Redis存储数组

Redis是不能够直接存储数组的，若是直接存储数组类型的数值，则获取后的数值类型发生变化，以下，存入numpy数组类型，获取后的类型是bytes类型。

import numpy as np
from redis import StrictRedis

r = StrictRedis(host=‘localhost’, port=6379, db=2)
x1 = np.array(([0.2,0.1,0.6],[10.2,4.2,0.9]))
r.set('test1', x1)
>>> True
r.get('test1')
>>> b'[[ 0.2  0.1  0.6]\n [10.2  4.2  0.9]]'
type(r.get('test1')) #获取后的数据类型
>>> <class 'bytes'>
复制代码

为了保持数据存储先后类型一致，在存储数组以前将其序列化，获取数组的时候将其反序列化便可。

借助于python的pickle模块进行序列化操做。

import pickle
r.set('test2', pickle.dumps(x1))
>>> True
pickle.loads(r.get('test2'))
>>> array([[ 0.2,  0.1,  0.6],
         [10.2,  4.2,  0.9]])
复制代码

这样，就能够保持数据存入前和取出后的类型一致。

2.4 web开发框架——Flask

以前学习python语言，历来没有关注过Web开发这一章节，由于工做内容并无涉及这一部分。现在须要从新看一下。

早期软件主要运行在桌面上，数据库这样的软件运行在服务器端，这种Client/Server模式简称CS架构。随着互联网的兴起，CS架构不适合Web，最大缘由是Web应用程序的修改和升级很是频繁，CS架构须要每一个客户端逐个升级桌面App，所以，Browser/Server模式开始流行，简称BS架构。

在BS架构下，客户端只须要浏览器，应用程序的逻辑和数据存储在服务器端，浏览器只须要请求服务器，获取Web页面，并把Web页面展现给用户便可。当前，Web页面也具备极强的交互性。

Python的诞生历史比Web还要早，因为Python是一种解释型的脚本语言，开发效率高，因此很是适合用来作Web开发。

Python有上百个开源的Web框架，比较熟知的有Flask, Django。接下来以Flask为例，介绍如何利用Flask进行web部署。

关于web开发框架的介绍，能够参考下面这篇博文： 三个目前最火的Python Web开发框架，你值得拥有！

有关Flask的具体用法可参考其余博文，这方面的资料比较全。下面主要以具体使用示例来讲明：

1. 安装使用

1. 安装Flask

   pip install flask
   
   import flask # 导入
   flask.__version__ # 版本
   >>> '1.1.1' #当前版本
   复制代码

2. 一个简单的Flask示例

   Flask使用Python的装饰器在内部自动的把URL和函数给关联起来。

   # hello.py
   from flask import Flask, request
   
   app = Flask(__name__) #建立Flask类的实例，第一个参数是模块或者包的名称
   app.config['JSON_AS_ASCII']=False # 支持中文显示
   
   @app.route('/', methods=['GET', 'POST']) # 使用methods参数处理不一样HTTP方法
   def home():
       return 'Hello, Flask'
   
   if __name__ == '__main__':
       app.run()
   复制代码

   • 使用 route() 装饰器来告诉 Flask 触发函数的 URL;
   • 函数名称被用于生成相关联的 URL。函数最后返回须要在用户浏览器中显示的信息。

   运行该文件，会提示* Running on http://127.0.0.1:5000/，在浏览器中打开此网址，会自动调用home函数，返回Hello, Flask，则在浏览器页面上就会看到Hello, Flask字样。

   app.run的参数

   app.run(host="0.0.0.0", port="5000", debug=True, processes=2, threaded=False)
   复制代码

   • host设定为0.0.0.0,则可让服务器被公开访问
   • port：指定端口号，默认为5000
   • debug：是否开启debug模型，若是你打开 调试模式，那么服务器会在修改应用代码以后自动重启，而且当应用出错时还会提供一个 有用的调试器。
   • processes：线程数量，默认是1
   • threaded：bool类型，是否开启多线程。注：当开启多个进程时，不支持同时开启多线程。

   注意：绝对不能在生产环境中使用调试器

2. Flask响应

视图函数的返回值会自动转换为一个响应对象。若是返回值是一个字符串，那么会被 转换为一个包含做为响应体的字符串、一个 200 OK 出错代码 和一个 text/html 类型的响应对象。若是返回值是一个字典，那么会调用 jsonify() 来产生一个响应。如下是转换的规则：

• 若是视图返回的是一个响应对象，那么就直接返回它。
• 若是返回的是一个字符串，那么根据这个字符串和缺省参数生成一个用于返回的 响应对象。
• 若是返回的是一个字典，那么调用 jsonify 建立一个响应对象。
• 若是返回的是一个元组，那么元组中的项目能够提供额外的信息。元组中必须至少 包含一个项目，且项目应当由 (response, status) 、 (response, headers) 或者 (response, status, headers) 组成。 status 的值会重载状态代码， headers 是一个由额外头部值组成的列表 或字典。
• 若是以上都不是，那么 Flask 会假定返回值是一个有效的 WSGI 应用并把它转换为一个响应对象。

JSON格式的API

JSON格式的响应是常见的，用Flask写这样的 API 是很容易上手的。若是从视图 返回一个 dict ，那么它会被转换为一个 JSON 响应。

@app.route("/me")
def me_api():
    user = get_current_user()
    return {
        "username": user.username,
        "theme": user.theme,
        "image": url_for("user_image", filename=user.image),
    }
复制代码

若是 dict 还不能知足需求，还须要建立其余类型的 JSON 格式响应，可使用 jsonify() 函数。该函数会序列化任何支持的 JSON 数据类型。

@app.route("/users")
def users_api():
    users = get_all_users()
    return jsonify([user.to_json() for user in users])
复制代码

3. 运行开发服务器

1. 经过命令行使用开发服务器

   强烈推荐开发时使用 flask 命令行脚本（ 命令行接口 ），由于有强大的重载功能，提供了超好的重载体验。基本用法以下:

   $ export FLASK_APP=my_application
   $ export FLASK_ENV=development
   $ flask run
   复制代码

   这样作开始了开发环境（包括交互调试器和重载器），并在 http://localhost:5000/提供服务。

   经过使用不一样 run 参数能够控制服务器的单独功能。例如禁用重载器:

   $ flask run --no-reload

2. 经过代码使用开发服务器

   另外一种方法是经过 Flask.run() 方法启动应用，这样当即运行一个本地服务器，与使用 flask 脚本效果相同。

   示例:

   if __name__ == '__main__':
       app.run()
   复制代码

   一般状况下这样作不错，可是对于开发就不行了。

   

2.5 使用Gunicorn

当咱们执行上面的app.py时，使用的flask自带的服务器，完成了web服务的启动。在生产环境中，flask自带的服务器，没法知足性能要求，咱们这里采用Gunicorn作wsgi容器，来部署flask程序。

Gunicorn（绿色独角兽）是一个Python WSGI UNIX HTTP服务器。从Ruby的独角兽（Unicorn ）项目移植。该Gunicorn服务器做为wsgi app的容器，可以与各类Web框架兼容，实现很是简单，轻量级的资源消耗。Gunicorn直接用命令启动，不须要编写配置文件，相对uWSGI要容易不少。

web开发中，部署方式大体相似。

1. 安装及使用

pip install gunicorn
复制代码

若是想让Gunicorn支持异步workers的话须要安装如下三个包：

pip install gevent
pip install eventlet
pip install greenlet
复制代码

指定进程和端口号，启动服务器：

gunicorn -w 4 -b 127.0.0.1:5001 运行文件名称:Flask程序实例名

以上述hello.py文件为例：

gunicorn -w 4 -b 127.0.0.1:5001 hello:app

参数： -w: 表示进程（worker）。 -b：表示绑定ip地址和端口号（bind）

查看gunicorn的具体参数，可执行gunicorn -h 一般将配置参数写入到配置文件中，如gunicorn_conf.py

重要参数：

• bind: 监听地址和端口
• workers: worker进程的数量。建议值：2~4 x (NUM_CORES)，缺省值是1.
• worker_class：worker进程的工做方式。有：sync (缺省值),eventlet, gevent, gthread, tornado
• threads：工做进程中线程的数量。建议值：2~4 x (SUM_CORES)，缺省值是1.
• reload: 当代码有修改时，自动重启workers。适用于开发环境，默认为False
• daemon：应用是否以daemon方式运行，是否以守护进程启动，默认False
• accesslog：访问日志文件路径
• errorlog：错误日志路径
• loglevel： 日志级别。debug， info， warning, error, critical.

一个参数配置示例：

# gunicorn_conf.py
bind: '0.0.0.0:5000' # 监听地址和端口号
workers = 2 # 进程数
worker_class = 'sync' #工做模式，可选sync, gevent, eventlet, gthread, tornado等
threads = 1 # 指定每一个进程的线程数，默认为1
worker_connections = 2000 # 最大客户并发量
timeout = 30 # 超时时间，默认30s
reload = True # 开发模式，代码更新时自动重启
daemon = False # 守护Gunicorn进程，默认False

accesslog = './logs/access.log' # 访问日志文件
errorlog = './logs/error.log'
loglevel = 'debug' # 日志输出等级，debug, info, warning, error, critical
复制代码

调用命令：

gunicorn -c gunicorn_conf.py hello:app

参数配置文件示例可见： gunicorn/example_config.py at master · benoitc/gunicorn

三、代码示例

#flask_feature.app
import numpy as np
from flask import Flask, jsonify
from keras.models import Model
from keras.applications.vgg16 import VGG16
from keras.backend.tensorflow_backend import set_session

app = Flask(__name__)
app.config['JSON_AS_ASCII']=False

@app.route("/", methods=["GET", "POST"])
def feature():
    img_feature = extract()
    return jsonify({'result':'true', 'msg':'成功'})

def extract(img_name):
    # 图像预处理
    img = load_image(img_name, target_size=(feature_params["size"], feature_params["size"])) 

    x = image.img_to_array(img)
    x = np.expand_dims(x, axis=0)
    x = preprocess_input(x)
    
    with graph.as_default():
        set_session(sess)
        res = model.predict(x)
    
    return res
    
    
if __name__ == '__main__':
    tf_config = some_custom_config
    sess = tf.Session(config=tf_config)
    set_session(sess)
    base_model = VGG16(weights=model_weights, include_top=True)
    model = Model(inputs=base_model.input,
                    outputs=base_model.get_layer(layer).output)
    graph = tf.get_default_graph()
    
    app.run()
复制代码

使用gunicorn启动服务命令：

gunicorn -c gunicorn_conf.py flask_feature:app

四、遇到的问题

在此记录整个部署工做中遇到的问题及对应解决方法。

4.1 Flask多线程与多进程问题

因为对算法的时间性能要求较高，所以尝试使用Flask自带的多线程与多进程选项测试效果。 在Flask的app.run()函数中，上面有介绍到processes参数，用于指定开启的多进程数量，threaded参数用于指定是否开启多线程。

flask开启debug模式，启动服务时，dubug模式会开启一个tensorflow的线程，致使调用tensorflow的时候，graph产生了错位。

4.1 Flask与Keras问题

使用Flask启动服务的时候，将遇到的问题及参考的资料记录在此。

Q1：Tensor is not an element of this graph

错误信息：

"Tensor Tensor(\"pooling/Mean:0\", shape=(?, 1280), dtype=float32) is not an element of this graph.", 
复制代码

描述：使用Keras中预训练模型进行图像分类特征提取的代码能够正常跑通，当经过Flask来启动服务，访问预测函数时，出现上述错误。

缘由：使用了动态图，即在作预测的时候，加载的graph并非第一次初始化模型时候的Graph，全部里面并无模型里的参数和节点等信息。

有人给出以下解决方案：

import tensorflow as tf
global graph, model
graph = tf.get_default_graph()

#当须要进行预测的时候
with graph.as_default():
    y = model.predict(x)
复制代码

Q2：使用Flask启动服务，加载两次模型，占用两份显存

出现该问题的缘由是使用Flask启动服务的时候，开启了debug模式，即debug=True。dubug模式会开启一个tensorflow的线程，此时查看GPU显存占用状况，会发现有两个进程都占用相同份的显存。

关闭debug模型（debug=False）便可。

参考资料：

[1]:Keras + Flask 提供接口服务的坑~~~

4.2 gunicorn启动服务相关问题

当使用gunicorn启动服务的时候，遇到如下问题：

Q1: Failed precondition

具体问题：

2 root error(s) found.\n 
(0) Failed precondition: Error while reading resource variable block5_conv2/kernel from Container: localhost. This could mean that the variable was uninitialized. Not found: Container localhost does not exist. (Could not find resource: localhost/block5_conv2/kernel)\n\t [[{{node block5_conv2/convolution/ReadVariableOp}}]]\n\t [[fc2/Relu/_7]]\n 
(1) Failed precondition: Error while reading resource variable block5_conv2/kernel from Container: localhost. This could mean that the variable was uninitialized. Not found: Container localhost does not exist. (Could not find resource: localhost/block5_conv2/kernel)\n\t [[{{node block5_conv2/convolution/ReadVariableOp}}]]\n0 successful operations.\n0 derived errors ignored." 复制代码

解决方法：

经过建立用于加载模型的会话的引用，而后在每一个须要使用的请求中使用keras设置session。具体以下：

from tensorflow.python.keras.backend import set_session
from tensorflow.python.keras.models import load_model

tf_config = some_custom_config
sess = tf.Session(config=tf_config)
graph = tf.get_default_graph()

# IMPORTANT: models have to be loaded AFTER SETTING THE SESSION for keras! 
# Otherwise, their weights will be unavailable in the threads after the session there has been set
set_session(sess)
model = load_model(...)

# 在每个request中：
global sess
global graph
with graph.as_default():
    set_session(sess)
    model.predict(...)
复制代码

有网友分析缘由： tensorflow的graph和session不是线程安全的，默认每一个线程建立一个新的session（不包含以前已经加载的weights, models 等）。所以，经过保存包含全部模型的全局会话并将其设置为在每一个线程中由keras使用，能够解决问题。

有网友提取一种改进方式：

# on thread 1
session = tf.Session(graph=tf.Graph())
with session.graph.as_default():
    k.backend.set_session(session)
    model = k.models.load_model(filepath)

# on thread 2
with session.graph.as_default():
    k.backend.set_session(session)
    model.predict(x, **kwargs)
复制代码

这里的新颖性容许（一次）加载多个模型并在多个线程中使用。默认状况下，加载模型时使用“默认”Session和“默认”graph。可是在这里是建立新的。还要注意，Graph存储在Session对象中，这样更加方便。

测试了一下好像不行

Q2：没法启动服务，CRITICAL WORKER TIMEOUT

当使用gunicorn启动flask服务时，查看服务器状态和日志文件发现一直在尝试启动，可是一直没有成功。

CRITICAL WORKER TIMEOUT

这是gunicorn配置参数timeout致使的。默认值为30s，即超过30s，就会kill掉进程，而后从新启动restart。

当启动服务进行初始化的时间超过timeout值时，就会一直启动，kill, restart。

可根据具体状况，适当增长该值。

参考资料：

tensorflow - GCP ML-engine FailedPreconditionError (code: 2) - Stack Overflow

五、参考资料

欢迎来到 Flask 的世界 — Flask 中文文档（ 1.1.1 ）

Gunicorn-配置详解

At Runtime : "Error while reading resource variable softmax/kernel from Container: localhost" · Issue #28287 · tensorflow/tensorflow

【已解决】线上环境经过gunicorn去运行Flask出错：CRITICAL WORKER TIMEOUT