使用 Flask API 将机器学习模型部署到生产环境中

Posted 2022-06-15 AI科技大本营

作者 | 云朵君

来源 | 数据STUDIO

在刚开始学习机器学习时，付出了很多努力来建立一个非常好的模型，而在现实工作中，如何部署及上线一个完成的机器学习模型，是成功的最后一步。为了避免尴尬，本文中，将和大家一起学习，如何在 Python 中使用 Flask 框架部署机器学习模型。本文属于入门级别，大佬可提出指导意见。

本文的基本结构：

机器学习模型实现方式

在现实工作中，需要将机器学习模型应用于某个产品功能中，如自动邮件系统或聊天机器人的一个小组件。大多数情况下，研发机器学习模型的算法工程师（或数据挖掘工程师等），与消费这些ML模型的或是完全不同技术栈的软件工程师，这可能会导致不少问题，而现在有下面两种方法可以解决这样的问题：

• 选项 1：用软件工程人员工作的语言重写整个代码。 这似乎是个好主意，但是复制这些复杂模型所需的时间和精力将完全浪费。再者像大多数软件工程人员所使用的 javascript 语言都没有很好的库来执行机器学习。这种方法通常是需要避免的。

• 选项 2：API 优先方法。 Web API 提供了良好的衔接，使跨语言应用程序很容易运行。如果前端开发人员需要使用算法工程师 ML 模型来创建支持 ML 的 Web 应用程序，他们只需要从提供 API 的位置获取 URL 端点。

什么是API

简而言之，API 是两个软件之间的（假设的）合同，如果用户软件以预定义的格式提供输入，则后者扩展其功能并将结果提供给用户软件。可以这样想，图形用户界面 (GUI) 或命令行界面 (CLI) 允许人类与代码交互，而应用程序可编程接口 (API) 允许一段代码与其他代码交互。

API 最常见的用例之一是在 Web 上。可以这么说，既然你能看到这篇文章，就说明你肯定使用过 API。在社交媒体上分享内容、通过网络支付、通过社交句柄显示推文列表——所有这些服务都在后面使用 API。

开发人员广泛使用 API 在其软件中实现各种功能。他们只需在软件中使用简单的 API 调用来实现复杂的功能，而不必自己编写代码。

API的基本元素

API 具有三个主要元素：

• Access：是否允许用户或谁请求数据或服务

• Request：是实际请求的数据或服务。请求有两个主要部分：

  • Methods：即你可以提出的问题，假设你可以访问（它还定义了可用的响应类型）。

  • Parameters：你可以在问题或响应中包含的其他详细信息。

• Response：根据你的请求提供的数据或服务。

API 类别

基于网络的系统

Web API 是 Web 服务器或 Web 浏览器的接口。这些 API 广泛用于 Web 应用程序的开发。这些 API 在服务器端或客户端工作。Baidu、Google 等公司都提供基于 Web 的 API。

操作系统

有多个基于操作系统的 API 提供了各种操作系统特性的功能，可以在创建 windows 或 mac 应用程序时合并。

数据库系统

与大多数数据库的交互是使用对数据库的 API 调用完成的。这些 API 的定义方式是以请求客户端可以理解的预定义格式传递请求的数据。

这使得与数据库的交互过程通用化，从而增强了应用程序与各种数据库的兼容性。它们非常健壮，并为数据库提供结构化接口。

硬件系统

这些 API 允许访问系统的各种硬件组件。它们对于建立与硬件的通信至关重要。因此，它可以实现从传感器数据收集到甚至显示在屏幕上的一系列功能。

大多数大型云提供商和专注于机器学习的小型公司都提供即用型 API。它们满足了不具备 ML 专业知识的开发人员/企业的需求，他们希望在其流程或产品套件中实施 ML。

在本文中，我们将了解如何使用 Python 中的 Web 框架 Flask 创建我们自己的机器学习 API。

注意： Flask 不是唯一可用的网络框架。有 Django、Falcon、Hug 等等

Python 环境设置

使用 Anaconda 创建虚拟环境。如果你需要在 Python 中创建工作流并保持依赖项分离或共享环境设置，Anaconda 发行版是一个不错的选择。

• Python 的 miniconda 安装指南

• wget https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh

• bash Miniconda3-latest-Linux-x86_64.sh

• source .bashrc

• 创建新环境：conda create --name <environment-name> python=3.6

• 完成后运行：source activate <environment-name>

• 安装你需要的python包，两个重要的是：flask & gunicorn。

一些术语

Route

该route是python flask中的装饰器。它基本上告诉app要run哪个函数或应该在哪个 URL 上呈现用户。在路由函数中，转义序列描述了 URL。定义路由后的函数被创建，你可以像普通的python函数一样传递参数。

Flask 也支持动态路由。可以修改 URL，或者在渲染时可以将各种条件与自定义数据一起发送。

HTTP 方法

HTTP 方法是worldwide web上各方之间的核心通信块。它有助于从不同的网站或文件获取、发送、缓存数据。我们探索一下 Flask 支持的不同 HTTP 方法，以及哪种方法用于什么目的。

1) GET

它是通过将内容连接到 URL 来向网站发送数据的最基本形式。GET 方法最常用于从文件中获取数据或加载新的 html 页面。它可用于你发送非机密数据的地方，如果披露这些数据不会成为问题。

2) POST

POST 方法是 GET 请求之后最常用的方法。它用于使用加密将数据发送到服务器。数据不附加到 URL，它使用 jinja python 模板发送并显示在 HTML 正文中。POST 方法主要用于当我们使用表单发送接收用户数据以及处理发送输出返回以在 HTML 正文中显示之后。

POST 方法是最受信任的方法，用于发送登录凭据等机密数据。

3) HEAD

Head 方法类似于 Get 方法，但它可以缓存在系统上。传递的数据是未加密的，它必须有响应。假设如果某些 URL 请求下载大文件，现在通过使用 HEAD 方法 URL 可以请求获取文件大小。

4) PUT

PUT 方法类似于 POST 方法。唯一的区别在于，当我们多次调用 POST 请求时，会发出多次请求，而在 PUT 方法中，它反对多个请求并用旧响应替换新请求。

5) DELETE

delete 是一种简单的 HTTP 方法，用于将某些特定资源删除到服务器。

Flask 基础

我们将尝试一个简单的 Flask Hello-World app应用程序，并使用 gunicorn 提供服务：

• 打开文本编辑器并在文件夹中创建hello-world.py文件

• 编写以下代码：

# Filename: hello-world.py
from flask import Flask
app = Flask(__name__)
@app.route('/users/<string:username>')
def hello_world(username=None):
   return("Hello !".format(username))
if __name__ == '__main__':
   app.run()

在应用程序运行app run函数中可以定义一些参数。run 函数基本上是在本地开发服务器上运行应用程序app。

app.run(host, port, debug, options)

• host -- 主机，它定义了要接受的主机名，在本地主机上运行（默认为 127.0.0.1）

• port -- 端口调用应用程序的端口，默认端口为 5000。

• debug -- 调试默认为 false。如果它设置为 true，那么它会在命令提示符中提供调试信息，当修改应用程序并保存它时，它会自动重新加载服务器，因此在实现时最好将调试保持为 True。

• options -- 选项被转发到 werkzeug 服务器。

gunicorn

Gunicorn是一个unix上被广泛使用的高性能的Python WSGI UNIX HTTP Server。和大多数的web框架兼容，并具有实现简单，轻量级，高性能等特点。

pip install gunicorn

• 保存文件并返回终端。

• 为了提供API(开始运行它)，在终端上执行:gunicorn——bind 0.0.0.0:8000 hello-world:app。

• 如果你得到了以下答案，那么就对了：

$ gunicorn --bind 0.0.0.0:5000 hello-world:app
[2021-05-01 10:36:09 +0000] [10673] [INFO] Starting gunicorn 19.7.1
[2021-05-01 10:36:09 +0000] [10673] [INFO] Listening at: http://0.0.0.0:5000 (10673)
[2021-05-01 10:36:09 +0000] [10673] [INFO] Using worker: sync
[2021-05-01 10:36:09 +0000] [10677] [INFO] Booting worker with pid: 10677

• 在你浏览器上输入https://localhost:8000/users/any-name

至此我们就完成了第一个 Flask 应用程序。正如现在通过几个简单的步骤所体验的那样，我们能够创建可以在本地访问的 Web 端点。

我们可以使用 Flask 包装我们的机器学习模型并将它们轻松地用作 Web API。此外，如果我们想创建更复杂的 Web 应用程序（包括 JavaScript *gasps*），我们只需要进行一些修改。

保存机器学习模型：序列化和反序列化

在保存模型之前，有一个关键步骤，即模型创建，但这并不是本节的重点，有大量的往期往期文章供大家参考。

在计算机科学中，在数据存储的上下文中，序列化是将数据结构或对象状态转换为可以存储（例如，在文件或内存缓冲区中，或通过网络连接链路传输）和重构的格式的过程稍后在相同或另一个计算机环境中。

在 Python 中，pickle是一种标准方法，该方法将存储对象存储并将其作为原始状态检索。举一个简单的例子：

list_to_pickle = [1, 'here', 123, 'walker']
import pickle
list_pickle = pickle.dumps(list_to_pickle)
list_pickle

b'\\x80\\x03]q\\x00(K\\x01X\\x04\\x00\\x00\\x00hereq
  \\x01KX\\x06\\x00\\x00\\x00walkerq\\x02e.'

当重新加载pickle时:

loaded_pickle = pickle.loads(list_pickle)
loaded_pickle

[1, 'here', 123, 'walker']

当然也可以将pickle的对象保存到文件中并使用它。如果不使用pickle 进行序列化，h5py也是另一种可替代方案。

我们有一个自定义类，我们需要在运行训练时导入它，因此使用 dill 模块将估计器类与网格搜索对象打包在一起。

建议创建一个单独的training.py文件，其中包含用于训练模型的所有代码：

上下滑动查看更多源码

import os 
import json
import numpy as np
import pandas as pd
import dill as pickle
from sklearn.externals import joblib
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.ensemble import RandomForestClassifier

from sklearn.pipeline import make_pipeline

import warnings
warnings.filterwarnings("ignore")

def build_and_train():

 data = pd.read_csv('../data/training.csv')
 data = data.dropna(subset=['Gender', 'Married', 'Credit_History', 'LoanAmount'])

 pred_var = ['Gender','Married','Dependents','Education',
              'Self_Employed','ApplicantIncome','CoapplicantIncome',
              'LoanAmount','Loan_Amount_Term','Credit_History','Property_Area']

 X_train, X_test, y_train, y_test = train_test_split(data[pred_var],
                                                      data['Loan_Status'],
                                                      test_size=0.25, 
                                                      random_state=42)
 y_train = y_train.replace('Y':1, 'N':0).as_matrix()
 y_test = y_test.replace('Y':1, 'N':0).as_matrix()

 pipe = make_pipeline(PreProcessing(),
      RandomForestClassifier())

 param_grid = 
       "randomforestclassifier__n_estimators" : [10, 20, 30],
     "randomforestclassifier__max_depth" : [None, 6, 8, 10],
     "randomforestclassifier__max_leaf_nodes": [None, 5, 10, 20], 
     "randomforestclassifier__min_impurity_split": [0.1, 0.2, 0.3]

 grid = GridSearchCV(pipe, param_grid=param_grid, cv=3)

 grid.fit(X_train, y_train)

 return(grid)


class PreProcessing(BaseEstimator, TransformerMixin):
    """
    为我们的用例定制预处理评估器
    """

    def __init__(self):
        pass

    def transform(self, df):
        """
        常规transform()可以帮助训练、验证和测试数据集
        """
        pred_var = ['Gender','Married','Dependents','Education',
                    'Self_Employed','ApplicantIncome', 'CoapplicantIncome',
                    'LoanAmount','Loan_Amount_Term','Credit_History','Property_Area']
        
        df = df[pred_var]
        
        df['Dependents'] = df['Dependents'].fillna(0)
        df['Self_Employed'] = df['Self_Employed'].fillna('No')
        df['Loan_Amount_Term'] = df['Loan_Amount_Term'].fillna(self.term_mean_)
        df['Credit_History'] = df['Credit_History'].fillna(1)
        df['Married'] = df['Married'].fillna('No')
        df['Gender'] = df['Gender'].fillna('Male')
        df['LoanAmount'] = df['LoanAmount'].fillna(self.amt_mean_)
        
        gender_values = 'Female' : 0, 'Male' : 1 
        married_values = 'No' : 0, 'Yes' : 1
        education_values = 'Graduate' : 0, 'Not Graduate' : 1
        employed_values = 'No' : 0, 'Yes' : 1
        property_values = 'Rural' : 0, 'Urban' : 1, 'Semiurban' : 2
        dependent_values = '3+': 3, '0': 0, '2': 2, '1': 1
        df.replace('Gender': gender_values, 
                    'Married': married_values, 
                    'Education': education_values, 
                    'Self_Employed': employed_values, 
                    'Property_Area': property_values, 
                    'Dependents': dependent_values, inplace=True)
        
        return df.as_matrix()

    def fit(self, df, y=None, **fit_params):
        """
        拟合Training数据集并从train计算所需的值，
        例如:我们需要X_train['Loan_Amount_Term']的均值，
        它将用于X_test的转换
        """
        
        self.term_mean_ = df['Loan_Amount_Term'].mean()
        self.amt_mean_ = df['LoanAmount'].mean()
        return self

if __name__ == '__main__':
 model = build_and_train()

 filename = 'model_v1.pk'
 with open('../flask_api/models/'+filename, 'wb') as file:
  pickle.dump(model, file)

安装和使用dill模块。

# !pip install dill
import dill as pickle
filename = 'model_v1.pk'
with open('../flask_api/models/' + filename, 'wb') as file:
  pickle.dump(grid, file)

所以我们的模型将保存在上面的位置。现在模型已经pickle好了，下一步就是围绕它创建一个 Flask 包装器。

在此之前，为了确保我们的pickle文件工作正常——将其加载回来并进行预测：

with open('../flask_api/models/'+filename ,'rb') as f:
    loaded_model = pickle.load(f)
loaded_model.predict(test_df)

array([1, 1, 1, 1, 1])

我们已经有了一个可以满足新传入数据所需的预处理步骤管道，所以我们只需要运行 predict() 函数即可。在scikit-learn 中使用管道相对容易。

即使在最初的实现看起来比较麻烦，但成体系的评估器和管道都可以节省相当多的时间。

使用 Flask 创建 API

构造我们的包装函数有三个重要部分apicall()：

• request获取数据（要进行预测）

• 加载pickle后的评估器

• jsonify序列化我预测结果并将响应发送回，并设置status code: 200

HTTP 消息由请求头header和正文body组成。发送的大部分正文内容设置为标准的json格式的。以批处理的方式发送(POST url-endpoint/)传入的数据并获得预测结果。

注意：可以直接发送纯文本、XML、csv或图像，但为了格式的可转换性，建议使用json)

上下滑动查看更多源码

# Filename: server.py
import os
import pandas as pd
from sklearn.externals import joblib
from flask import Flask, jsonify, request

app = Flask(__name__)
@app.route('/predict', methods=['POST'])
def apicall():
    """
    API Call
    Pandas dataframe (sent as a payload) from API Call
    """
    try:
        test_json = request.get_json()
        test = pd.read_json(test_json, orient='records')
        #来解决如下错误问题TypeError: Cannot compare types 'ndarray(dtype=int64)' and 'str'
        test['Dependents'] = [str(x) for x in list(test['Dependents'])]

        #拆分出 Loan_IDs
        loan_ids = test['Loan_ID']

    except Exception as e:
        raise e

    clf = 'model_v1.pk'

    if test.empty:
        return(bad_request())
    else:
        #导入存储的模型
        print("Loading the model...")
        loaded_model = None
        with open('./models/'+clf,'rb') as f:
            loaded_model = pickle.load(f)

        print("The model has been loaded...doing predictions now...")
        predictions = loaded_model.predict(test)

        """
        将预测作为Series添加到新的pandas数据框架中
        或根据用例，整个测试数据都附加了新文件
        """
        prediction_series = list(pd.Series(predictions))
        final_predictions = pd.DataFrame(list(zip(loan_ids, prediction_series)))

        """
        我们在发送回复时需要发送响应码。
        """
        responses = jsonify(predictions=final_predictions.to_json(orient="records"))
        responses.status_code = 200

        return (responses)

完成后，我们运行：“

gunicorn --bind 0.0.0.0:8000 server:app

我们生成一些预测数据并查询在本地运行的 APIhttps:0.0.0.0:8000/predict

import json
import requests

# 设置发送和接受json响应的头

header = 'Content-Type': 'application/json',
           'Accept': 'application/json'
# 读取测试数据
df = pd.read_csv('../data/test.csv', encoding="utf-8-sig")
df = df.head()
# 将Pandas Dataframe 转为 json
data = df.to_json(orient='records')
data

上下滑动查看更多
'["Loan_ID":"LP001015","Gender":"Male","Married":"Yes",
 "Dependents":"0","Education":"Graduate","Self_Employed":"No",
 "ApplicantIncome":5720,"CoapplicantIncome":0,"LoanAmount":110.0,
 "Loan_Amount_Term":360.0,"Credit_History":1.0,"Property_Area":"Urban",
 "Loan_ID":"LP001022","Gender":"Male","Married":"Yes","Dependents":"1",
   "Education":"Graduate","Self_Employed":"No","ApplicantIncome":3076,
   "CoapplicantIncome":1500,"LoanAmount":126.0,"Loan_Amount_Term":360.0,
   "Credit_History":1.0,"Property_Area":"Urban",
 "Loan_ID":"LP001031","Gender":"Male","Married":"Yes","Dependents":"2",
   "Education":"Graduate","Self_Employed":"No","ApplicantIncome":5000,
   "CoapplicantIncome":1800,"LoanAmount":208.0,"Loan_Amount_Term":360.0,
   "Credit_History":1.0,"Property_Area":"Urban",
 "Loan_ID":"LP001035","Gender":"Male","Married":"Yes","Dependents":"2",
   "Education":"Graduate","Self_Employed":"No","ApplicantIncome":2340,
   "CoapplicantIncome":2546,"LoanAmount":100.0,"Loan_Amount_Term":360.0,
   "Credit_History":null,"Property_Area":"Urban",
 "Loan_ID":"LP001051","Gender":"Male","Married":"No","Dependents":"0",
   "Education":"NotGraduate","Self_Employed":"No","ApplicantIncome":3276,
   "CoapplicantIncome":0,"LoanAmount":78.0,"Loan_Amount_Term":360.0,
   "Credit_History":1.0,"Property_Area":"Urban"]'

# POST <url>/predict
resp = requests.post("http://0.0.0.0:8000/predict", 
                      data = json.dumps(data),
                      headers= header)
resp.status_code

200

得到的最终响应如下:

resp.json()

'predictions': '["0":"LP001015","1":1,...

写在最后

到目前为止，我们仅仅完成了对使用 Flask 创建 API 的一小步尝试，将我们的 ML 解决方案直接集成到我们的产品中。这是一个非常基本的 API，将有助于对数据产品进行原型设计，使其成为功能齐全的生产所需要用的 API，还需要一些不属于机器学习范围的附加内容。在模型部署方面还有很长的路要走。

在采用 API 时，有几点需要注意：

• 为了节约更多的精力，在构建机器学习工作流程时需要额外注意，就好像你需要创建一个干净、可用的 API 作为可交付成果一样。

• 尝试对模型和 API 代码使用版本控制，Flask 对版本控制没有很好的支持。保存和跟踪 ML 模型相对困难，请找出适合你的最简单的方法，推荐使用git。

• 如果使用自定义估计器进行预处理或任何其他相关任务，需要确保将估计器和训练代码放在一起，以便pickle的模型将估计器类标记在一起。

CSDN音视频技术开发者在线调研正式上线！

现邀开发者们扫码在线调研

往期回顾

Gain 算法实现缺失值预测

数据筛选query函数详细介绍

M2芯片重磅问世，性能提升18%！

AI考生挑战高考作文，平均1秒1篇

分享
点收藏
点点赞
点在看