在生產環境中使用 Keras、Redis、Flask 和 Apache 進行深度學習
今天我們演示如何在生產環境中使用Keras、Redis、Flask 和 Apache 進行深度學習
項目結構
keras-complete-rest-api ├── helpers.py ├── jemma.png ├── keras_rest_api_app.wsgi ├── run_model_server.py ├── run_web_server.py ├── settings.py ├── simple_request.py └── stress_test.py
1
2
3
4
5
6
7
8
9
文件解釋：
run_web_server.py 包含我們所有的 Flask Web 服務器代碼——Apache 將在啟動我們的深度學習 Web 應用程序時加載它。
run_model_server.py 將：
從磁盤加載我們的 Keras 模型
不斷輪詢Redis尋找新圖像進行分類
對圖像進行分類（批量處理以提高效率）
將推理結果寫回 Redis，以便它們可以通過 Flask 返回給客戶端。
settings.py 包含我們深度學習生產服務的所有基于 Python 的設置，例如 Redis 主機/端口信息、圖像分類設置、圖像隊列名稱等。
helpers.py 包含 run_web_server.py 和 run_model_server.py 都將使用的實用函數（即 base64 編碼）。
keras_rest_api_app.wsgi 包含我們的 WSGI 設置，因此我們可以從我們的 Apache 服務器為 Flask 應用程序提供服務。
simple_request.py 可用于以編程方式使用我們的深度學習 API 服務的結果。
jemma.png 是我家小獵犬的照片。在調用 REST API 以驗證它確實有效時，我們將使用她作為示例圖像。
最后，我們將使用 stress_test.py 來給我們的服務器施加壓力并在整個過程中測量圖像分類。
我們在 Flask 服務器上有一個端點 /predict 。此方法位于 run_web_server.py 中，將根據需要計算輸入圖像的分類。圖像預處理也在 run_web_server.py 中處理。
為了使我們的服務器做好生產準備，我從上周的單個腳本中取出了分類過程函數并將其放置在 run_model_server.py 中。這個腳本非常重要，因為它將加載我們的 Keras 模型并從 Redis 中的圖像隊列中抓取圖像進行分類。結果被寫回 Redis（/predict 端點和 run_web_server.py 中的相應函數監視 Redis 以將結果發送回客戶端）。
但是除非我們知道深度學習 REST API 服務器的功能和局限性，否則它有什么好處呢？
在 stress_test.py 中，我們測試我們的服務器。我們將通過啟動 500 個并發線程來實現這一點，這些線程將我們的圖像發送到服務器進行并行分類。我建議在服務器 localhost 上運行它以啟動，然后從異地客戶端運行它。
構建我們的深度學習網絡應用
圖 1：使用 Python、Keras、Redis 和 Flask 構建的深度學習 REST API 服務器的數據流圖。
這個項目中使用的幾乎每一行代碼都來自我們之前關于構建可擴展深度學習 REST API 的文章——唯一的變化是我們將一些代碼移動到單獨的文件中，以促進生產環境中的可擴展性。
設置和配置
# initialize Redis connection settings REDIS_HOST = "localhost" REDIS_PORT = 6379 REDIS_DB = 0 # initialize constants used to control image spatial dimensions and # data type IMAGE_WIDTH = 224 IMAGE_HEIGHT = 224 IMAGE_CHANS = 3 IMAGE_DTYPE = "float32" # initialize constants used for server queuing IMAGE_QUEUE = "image_queue" BATCH_SIZE = 32 SERVER_SLEEP = 0.25 CLIENT_SLEEP = 0.25
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
在 settings.py 中，您將能夠更改服務器連接、圖像尺寸 + 數據類型和服務器隊列的參數。
# import the necessary packages import numpy as np import base64 import sys def base64_encode_image(a): # base64 encode the input NumPy array return base64.b64encode(a).decode("utf-8") def base64_decode_image(a, dtype, shape): # if this is Python 3, we need the extra step of encoding the # serialized NumPy string as a byte object if sys.version_info.major == 3: a = bytes(a, encoding="utf-8") # convert the string to a NumPy array using the supplied data # type and target shape a = np.frombuffer(base64.decodestring(a), dtype=dtype) a = a.reshape(shape) # return the decoded image return a
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
helpers.py 文件包含兩個函數——一個用于 base64 編碼，另一個用于解碼。
編碼是必要的，以便我們可以在 Redis 中序列化 + 存儲我們的圖像。 同樣，解碼是必要的，以便我們可以在預處理之前將圖像反序列化為 NumPy 數組格式。
深度學習網絡服務器
# import the necessary packages from tensorflow.keras.preprocessing.image import img_to_array from tensorflow.keras.applications.resnet50 import preprocess_input from PIL import Image import numpy as np import settings import helpers import flask import redis import uuid import time import json import io # initialize our Flask application and Redis server app = flask.Flask(__name__) db = redis.StrictRedis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=settings.REDIS_DB) def prepare_image(image, target): # if the image mode is not RGB, convert it if image.mode != "RGB": image = image.convert("RGB") # resize the input image and preprocess it image = image.resize(target) image = img_to_array(image) image = np.expand_dims(image, axis=0) image = preprocess_input(image) # return the processed image return image @app.route("/") def homepage(): return "Welcome to the PyImageSearch Keras REST API!" @app.route("/predict", methods=["POST"]) def predict(): # initialize the data dictionary that will be returned from the # view data = {"success": False} # ensure an image was properly uploaded to our endpoint if flask.request.method == "POST": if flask.request.files.get("image"): # read the image in PIL format and prepare it for # classification image = flask.request.files["image"].read() image = Image.open(io.BytesIO(image)) image = prepare_image(image, (settings.IMAGE_WIDTH, settings.IMAGE_HEIGHT)) # ensure our NumPy array is C-contiguous as well, # otherwise we won't be able to serialize it image = image.copy(order="C") # generate an ID for the classification then add the # classification ID + image to the queue k = str(uuid.uuid4()) image = helpers.base64_encode_image(image) d = {"id": k, "image": image} db.rpush(settings.IMAGE_QUEUE, json.dumps(d)) # keep looping until our model server returns the output # predictions while True: # attempt to grab the output predictions output = db.get(k) # check to see if our model has classified the input # image if output is not None: # add the output predictions to our data # dictionary so we can return it to the client output = output.decode("utf-8") data["predictions"] = json.loads(output) # delete the result from the database and break # from the polling loop db.delete(k) break # sleep for a small amount to give the model a chance # to classify the input image time.sleep(settings.CLIENT_SLEEP) # indicate that the request was a success data["success"] = True # return the data dictionary as a JSON response return flask.jsonify(data) # for debugging purposes, it's helpful to start the Flask testing # server (don't use this for production if __name__ == "__main__": print("* Starting web service...") app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
在 run_web_server.py 中，您將看到 predict ，該函數與我們的 REST API /predict 端點相關聯。
predict 函數將編碼的圖像推送到 Redis 隊列中，然后不斷循環/輪詢，直到它從模型服務器獲取預測數據。 然后我們對數據進行 JSON 編碼并指示 Flask 將數據發送回客戶端。
深度學習模型服務器
# import the necessary packages from tensorflow.keras.applications import ResNet50 from tensorflow.keras.applications.resnet50 import decode_predictions import numpy as np import settings import helpers import redis import time import json # connect to Redis server db = redis.StrictRedis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=settings.REDIS_DB) def classify_process(): # load the pre-trained Keras model (here we are using a model # pre-trained on ImageNet and provided by Keras, but you can # substitute in your own networks just as easily) print("* Loading model...") model = ResNet50(weights="imagenet") print("* Model loaded") # continually pool for new images to classify while True: # attempt to grab a batch of images from the database, then # initialize the image IDs and batch of images themselves queue = db.lrange(settings.IMAGE_QUEUE, 0, settings.BATCH_SIZE - 1) imageIDs = [] batch = None # loop over the queue for q in queue: # deserialize the object and obtain the input image q = json.loads(q.decode("utf-8")) image = helpers.base64_decode_image(q["image"], settings.IMAGE_DTYPE, (1, settings.IMAGE_HEIGHT, settings.IMAGE_WIDTH, settings.IMAGE_CHANS)) # check to see if the batch list is None if batch is None: batch = image # otherwise, stack the data else: batch = np.vstack([batch, image]) # update the list of image IDs imageIDs.append(q["id"]) # check to see if we need to process the batch if len(imageIDs) > 0: # classify the batch print("* Batch size: {}".format(batch.shape)) preds = model.predict(batch) results = decode_predictions(preds) # loop over the image IDs and their corresponding set of # results from our model for (imageID, resultSet) in zip(imageIDs, results): # initialize the list of output predictions output = [] # loop over the results and add them to the list of # output predictions for (imagenetID, label, prob) in resultSet: r = {"label": label, "probability": float(prob)} output.append(r) # store the output predictions in the database, using # the image ID as the key so we can fetch the results db.set(imageID, json.dumps(output)) # remove the set of images from our queue db.ltrim(settings.IMAGE_QUEUE, len(imageIDs), -1) # sleep for a small amount time.sleep(settings.SERVER_SLEEP) # if this is the main thread of execution start the model server # process if __name__ == "__main__": classify_process()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
run_model_server.py 文件包含我們的classify_process 函數。 這個函數加載我們的模型，然后對一批圖像運行預測。 這個過程最好在 GPU 上執行，但也可以使用 CPU。
在這個例子中，為了簡單起見，我們將使用在 ImageNet 數據集上預訓練的 ResNet50。 您可以修改classify_process 以利用您自己的深度學習模型。
WSGI 配置
# add our app to the system path import sys sys.path.insert(0, "/var/www/html/keras-complete-rest-api") # import the application and away we go... from run_web_server import app as application
1
2
3
4
5
文件 keras_rest_api_app.wsgi 是我們深度學習 REST API 的一個新組件。 這個 WSGI 配置文件將我們的服務器目錄添加到系統路徑并導入 Web 應用程序以啟動所有操作。 我們在 Apache 服務器設置文件 /etc/apache2/sites-available/000-default.conf 中指向此文件，稍后將在本博文中介紹。
壓力測試
# import the necessary packages from threading import Thread import requests import time # initialize the Keras REST API endpoint URL along with the input # image path KERAS_REST_API_URL = "http://localhost/predict" IMAGE_PATH = "jemma.png" # initialize the number of requests for the stress test along with # the sleep amount between requests NUM_REQUESTS = 500 SLEEP_COUNT = 0.05 def call_predict_endpoint(n): # load the input image and construct the payload for the request image = open(IMAGE_PATH, "rb").read() payload = {"image": image} # submit the request r = requests.post(KERAS_REST_API_URL, files=payload).json() # ensure the request was sucessful if r["success"]: print("[INFO] thread {} OK".format(n)) # otherwise, the request failed else: print("[INFO] thread {} FAILED".format(n)) # loop over the number of threads for i in range(0, NUM_REQUESTS): # start a new thread to call the API t = Thread(target=call_predict_endpoint, args=(i,)) t.daemon = True t.start() time.sleep(SLEEP_COUNT) # insert a long sleep so we can wait until the server is finished # processing the images time.sleep(300)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
我們的 stress_test.py 腳本將幫助我們測試服務器并確定其限制。 我總是建議對您的深度學習 REST API 服務器進行壓力測試，以便您知道是否（更重要的是，何時）需要添加額外的 GPU、CPU 或 RAM。 此腳本啟動 NUM_REQUESTS 線程和 POST 到 /predict 端點。 這取決于我們的 Flask 網絡應用程序。
編譯安裝Redis
Redis 是一種高效的內存數據庫，它將充當我們的隊列/消息代理。 獲取和安裝Redis非常簡單：
$ wget http://download.redis.io/redis-stable.tar.gz $ tar xvzf redis-stable.tar.gz $ cd redis-stable $ make $ sudo make install
1
2
3
4
5
創建您的深度學習 Python 虛擬環境
安裝附加包：
$ workon dl4cv $ pip install flask $ pip install gevent $ pip install requests $ pip install redis
1
2
3
4
5
安裝 Apache Web 服務器
可以使用其他 Web 服務器，例如 nginx，但由于我對 Apache 有更多的經驗（因此通常更熟悉 Apache），因此我將在此示例中使用 Apache。 Apache 可以通過以下方式安裝：
$ sudo apt-get install apache2
1
如果您使用 Python 3 創建了一個虛擬環境，您將需要安裝 Python 3 WSGI + Apache 模塊：
$ sudo apt-get install libapache2-mod-wsgi-py3 $ sudo a2enmod wsgi
1
2
要驗證是否安裝了 Apache，請打開瀏覽器并輸入 Web 服務器的 IP 地址。 如果您看不到服務器啟動畫面，請確保打開端口 80 和端口 5000。 就我而言，我服務器的 IP 地址是 54.187.46.215（你的會有所不同）。 在瀏覽器中輸入這個，我看到：
Sym-link鏈接您的 Flask + 深度學習應用程序
默認情況下，Apache 提供來自 /var/www/html 的內容。 我建議創建一個從 /var/www/html 到 Flask Web 應用程序的符號鏈接。 我已將我的深度學習 + Flask 應用程序上傳到名為 keras-complete-rest-api 的目錄中的主目錄：
$ ls ~ keras-complete-rest-api
1
2
我可以通過以下方式將其符號鏈接到 /var/www/html：
$ cd /var/www/html/ $ sudo ln -s ~/keras-complete-rest-api keras-complete-rest-api
1
2
更新您的 Apache 配置以指向 Flask 應用程序
為了將 Apache 配置為指向我們的 Flask 應用程序，我們需要編輯 /etc/apache2/sites-available/000-default.conf 。 在你最喜歡的文本編輯器中打開（這里我將使用 vi ）：
$ sudo vi /etc/apache2/sites-available/000-default.conf
1
在文件的頂部提供您的 WSGIPythonHome（Python bin 目錄的路徑）和 WSGIPythonPath（Python 站點包目錄的路徑）配置：
WSGIPythonHome /home/ubuntu/.virtualenvs/keras_flask/bin WSGIPythonPath /home/ubuntu/.virtualenvs/keras_flask/lib/python3.5/site-packages
1
2
3
4
5
在 Ubuntu 18.04 上，您可能需要將第一行更改為：
WSGIPythonHome /home/ubuntu/.virtualenvs/keras_flask
由于我們在本示例中使用 Python 虛擬環境（我將我的虛擬環境命名為 keras_flask ），因此我們為 Python 虛擬環境提供 bin 和 site-packages 目錄的路徑。 然后在 的正文中，在 ServerAdmin 和 DocumentRoot 之后，添加：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
符號鏈接 CUDA 庫（可選，僅限 GPU）
如果您將 GPU 用于深度學習并希望利用 CUDA（您為什么不這樣做），不幸的是，Apache 不了解 /usr/local/cuda/lib64 中的 CUDA 的 *.so 庫。
我不確定什么是“最正確”的方式向 Apache 指示這些 CUDA 庫所在的位置，但“完全破解”解決方案是將所有文件從 /usr/local/cuda/lib64 符號鏈接到 /usr/lib ：
$ cd /usr/lib $ sudo ln -s /usr/local/cuda/lib64/* ./
1
2
重新啟動 Apache Web 服務器
編輯完 Apache 配置文件并可選擇符號鏈接 CUDA 深度學習庫后，請務必通過以下方式重新啟動 Apache 服務器：
$ sudo service apache2 restart
1
測試您的 Apache Web 服務器 + 深度學習端點
要測試 Apache 是否已正確配置以提供 Flask + 深度學習應用程序，請刷新您的 Web 瀏覽器：
您現在應該看到文本“歡迎使用 PyImageSearch Keras REST API！” 在您的瀏覽器中。 一旦你達到這個階段，你的 Flask 深度學習應用程序就應該準備好了。 綜上所述，如果您遇到任何問題，請確保參考下一節……
提示：如果遇到問題，請監控 Apache 錯誤日志
多年來，我一直在使用 Python + Web 框架，例如 Flask 和 Django，但在正確配置環境時仍然會出錯。 雖然我希望有一種防彈的方法來確保一切順利，但事實是，在此過程中可能會出現一些問題。 好消息是 WSGI 將 Python 事件（包括失敗）記錄到服務器日志中。 在 Ubuntu 上，Apache 服務器日志位于 /var/log/apache2/ ：
$ ls /var/log/apache2 access.log error.log other_vhosts_access.log
1
2
調試時，我經常打開一個運行的終端：
$ tail -f /var/log/apache2/error.log
1
…所以我可以看到第二個錯誤滾滾而來。 使用錯誤日志幫助您在服務器上啟動和運行 Flask。
啟動您的深度學習模型服務器
您的 Apache 服務器應該已經在運行。 如果沒有，您可以通過以下方式啟動它：
$ sudo service apache2 start
1
然后，您將要啟動 Redis 存儲：
$ redis-server
1
并在單獨的終端中啟動 Keras 模型服務器：
$ python run_model_server.py * Loading model... ... * Model loaded
1
2
3
4
從那里嘗試向您的深度學習 API 服務提交示例圖像：
$ curl -X POST -F image=@jemma.png 'http://localhost/predict' { "predictions": [ { "label": "beagle", "probability": 0.9461532831192017 }, { "label": "bluetick", "probability": 0.031958963721990585 }, { "label": "redbone", "probability": 0.0066171870566904545 }, { "label": "Walker_hound", "probability": 0.003387963864952326 }, { "label": "Greater_Swiss_Mountain_dog", "probability": 0.0025766845792531967 } ], "success": true }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
如果一切正常，您應該會收到來自深度學習 API 模型服務器的格式化 JSON 輸出，其中包含類別預測 + 概率。
對您的深度學習 REST API 進行壓力測試
當然，這只是一個例子。 讓我們對深度學習 REST API 進行壓力測試。 打開另一個終端并執行以下命令：
$ python stress_test.py [INFO] thread 3 OK [INFO] thread 0 OK [INFO] thread 1 OK ... [INFO] thread 497 OK [INFO] thread 499 OK [INFO] thread 498 OK
1
2
3
4
5
6
7
8
在 run_model_server.py 輸出中，您將開始看到記錄到終端的以下行：
* Batch size: (4, 224, 224, 3) * Batch size: (9, 224, 224, 3) * Batch size: (9, 224, 224, 3) * Batch size: (8, 224, 224, 3) ... * Batch size: (2, 224, 224, 3) * Batch size: (10, 224, 224, 3) * Batch size: (7, 224, 224, 3)
1
2
3
4
5
6
7
8
即使每 0.05 秒有一個新請求，我們的批次大小也不會超過每批次約 10-12 張圖像。 我們的模型服務器可以輕松處理負載而不會出汗，并且可以輕松擴展到此之外。 如果您確實使服務器超載（可能是您的批大小太大并且 GPU 內存不足并顯示錯誤消息），您應該停止服務器，并使用 Redis CLI 清除隊列：
$ redis-cli > FLUSHALL
1
2
從那里您可以調整 settings.py 和 /etc/apache2/sites-available/000-default.conf 中的設置。 然后您可以重新啟動服務器。
將您自己的深度學習模型部署到生產環境的建議
我能給出的最好建議之一是將您的數據，尤其是 Redis 服務器，靠近 GPU。 您可能想啟動一個具有數百 GB RAM 的巨型 Redis 服務器來處理多個圖像隊列并為多個 GPU 機器提供服務。 這里的問題將是 I/O 延遲和網絡開銷。
假設 224 x 224 x 3 圖像表示為 float32 數組，32 張圖像的批量大小將是 ~19MB 的數據。 這意味著對于來自模型服務器的每個批處理請求，Redis 將需要提取 19MB 的數據并將其發送到服務器。 在快速切換上，這沒什么大不了的，但您應該考慮在同一臺服務器上同時運行模型服務器和 Redis，以使數據靠近 GPU。
總結
在今天的博文中，我們學習了如何使用 Keras、Redis、Flask 和 Apache 將深度學習模型部署到生產環境中。 我們在這里使用的大多數工具都是可以互換的。您可以將 TensorFlow 或 PyTorch 換成 Keras。可以使用 Django 代替 Flask。 Nginx 可以換成 Apache。
我不建議換掉的唯一工具是 Redis。 Redis 可以說是內存數據存儲的最佳解決方案。除非您有不使用 Redis 的特定原因，否則我建議您使用 Redis 進行排隊操作。 最后，我們對深度學習 REST API 進行了壓力測試。
我們向我們的服務器提交了總共 500 個圖像分類請求，每個請求之間有 0.05 秒的延遲——我們的服務器沒有分階段（CNN 的批量大小從未超過約 37%）。
Apache Flask Python Redis 深度學習
版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。
相關文章
