NLP模型工程化部署

文章目录

- 一、理论-微服务、测试与GPU
- - 1）微服务架构
  - 2）代码测试
  - 3）GPU使用
- 二、实践-封装微服务，编写测试用例和脚本，并观察GPU
- - 1）微服务封装(RestFul和RPC)
  - 2）测试编写(unit_test\api_test\load_test)
  - 3）GPU使用和监控
- 三、理论-docker、CICD与K8S
- 四、实践-构建镜像与CICD脚本

一、理论-微服务、测试与GPU

学习目标与成果

1）微服务架构

①单服务架构
在这里插入图片描述
②微服务架构

③微服务特点

④主要技术栈

在这里插入图片描述
1）HTTP协议做RESTFul接口

基于flask web框架实现restful接口

2）远程RPC调用

基于grpc

restful还是rpc

2）代码测试

在这里插入图片描述

具体实践举例

代码测试

压力测试

压力测试工具
在这里插入图片描述

3）GPU使用

在这里插入图片描述
可以使用命令，每一秒刷新一次

watch -n 1 nvidia-smi

二、实践-封装微服务，编写测试用例和脚本，并观察GPU

1）微服务封装(RestFul和RPC)

①RestFul接口
（1）配置日志logger
在这里插入图片描述

import logging.handlers

# 获取logger
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

# 生成文件handler，打印到文件
# 按天滚动的log，一天滚动一次，只保留最近7个日志文件（即保留最近7天）
file_handler = logging.handlers.TimedRotatingFileHandler('./logs/root.log', 'D', 1, 7, encoding='utf-8')
file_handler.setLevel(logging.DEBUG)

# 设置formatter
# 打印日志时间、级别、文件名、行号、函数名字、内容
formatter = logging.Formatter(
    '%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s - %(funcName)s() - %(message)s'
)

# 将formatter设置到两个handler
file_handler.setFormatter(formatter)

# 将handler设置到logger
logger.addHandler(file_handler)

（2）编写RESTFUL API
1>编写hello_resource，配置健康检查路由

#hello_resource.py
from flask_restful import Resource


class HelloResource(Resource):
    """
    hello路由
    快速检查服务是否健康
    """

    def get(self):
        return 'hello'

2>编写seg_resource,配置分词路由

import json
import time

from flask import request
from flask_restful import Resource

from online import logger


class SegResource(Resource):
    """
    分词路由
    主要调用segment.seg
    """
    def __init__(self, segment):
        # 使用传过来的segment对象，进行后面的分词
        self.segment = segment

    def post(self):
        data = request.get_json()  # 解析输入json为一个dict

        init_time = time.time()
        result = {
            'status': 'OK',  # 本次请求返回状态
            'msg': ''  # 额外说明
        }

        request_id = data.get('request_id')  # 支持传入request_id，便于线上追踪请求

        try:
            assert data, "请确保输入不为空"

            # 从data取用户输入的各种参数
            content, model, enable_offset, enable_stop_word, use_ner = \
                data['content'], data.get('model'), data.get('enable_offset', False), \
                data.get('enable_stop_word', False), data.get('use_ner', False)
            logger.info('request_id: {}, model: {}, enable_offset: {}, enable_stop_word: {}, use_ner: {}, '
                        'content: {} ...'.format(request_id, model, enable_offset, enable_stop_word, use_ner,
                                                 content[:100]))

            # 调用segment对象的seg方法
            r = self.segment.seg(content, model=model, enable_offset=enable_offset,
                                 enable_stop_word=enable_stop_word, use_ner=use_ner)
            result['result'] = list(r)  # 将分词结果存放在result里面

        except Exception as e:
            # 出现异常，打印异常栈，更改本次请求状态为ERROR
            logger.exception(e)
            result['status'] = 'ERROR'
            result['msg'] = str(e)

        logger.info('request_id: {}, result: {} ..., cost time: {}s'.format(
            request_id, json.dumps(result, ensure_ascii=False)[:200], time.time() - init_time)
        )

        return result

3>编写其余resource
4>编写http server

#server.py
import sys

from flask import Flask
from flask_restful import Api

from online import logger
from online.http.resources.dict_resource import DictResource
from online.http.resources.hello_resource import HelloResource
from online.http.resources.pos_resource import PosResource
from online.http.resources.seg_resource import SegResource
from segment.segment import Segment


def start_server(port=8000):
    # 如果输入第1个参数，将第1个参数解析为端口号
    if len(sys.argv) > 1:
        port = int(sys.argv[1])

    # 实例化flask app
    app = Flask(__name__)
    app.config.update(RESTFUL_JSON=dict(ensure_ascii=False))  # 设置ensure_ascii=False，确保接口返回的中文正常
    api = Api(app)

    # 实例化segment对象，准备传入到各个resource里面
    segment = Segment()
    resource_class_kwargs = {'segment': segment}

    # 为api添加hello路由、seg路由、pos路由、dict路由
    api.add_resource(HelloResource, '/')  # hello路由用于快速检查服务可用性
    api.add_resource(SegResource, '/seg', resource_class_kwargs=resource_class_kwargs)  # seg路由用于分词
    api.add_resource(PosResource, '/pos', resource_class_kwargs=resource_class_kwargs)  # pos路由用于词性标注
    api.add_resource(DictResource, '/dict', resource_class_kwargs=resource_class_kwargs)  # dict路由用于管理词典

    # 启动服务，设置host port
    # host='0.0.0.0'，表示外部机器可以访问，必须设置为0.0.0.0
    # threaded=False，表示我们的主程序是单线程模式，需要一个一个处理请求
    # （我们的word_graph对象不是线程安全的）
    logger.info('server starts port {}'.format(port))
    app.run(debug=False, host='0.0.0.0', port=port, threaded=False)


if __name__ == '__main__':
    start_server()