Dockerize модель ML/DL
Докеризация модели машинного обучения — это первый шаг к развертыванию вашей модели машинного обучения. Ниже приведено четкое и краткое руководство с примером шаблона кода для установки. Эта настройка поможет вам протестировать вашу модель машинного обучения и получить ответ в Flask API.
1. Настройка
а) Создать каталог
ML_app/ ├── app │ └── main.py │ └── app.py | └── db.py | └── models.py | └── requirements.txt | └── Dockerfile └── docker-compose.yml
б) Dockerfile: создайте Docker-образ вашего ML_app.
Убедитесь, что bash указан как «8200», то есть номер порта.
FROM python:3.9-buster ENV PYTHONUNBUFFERED 1 COPY . /app WORKDIR /app RUN pip install -r requirements.txt CMD bash -c "python app.py runserver 0.0.0.0:8200"
в) docker-compose.yml: свяжите вместе различные сервисы, такие как API Flask и базу данных.
Обязательно измените следующее:
i) container_name: «мой-мл-приложение-контейнер»
ii) имя_базы_данных: то же имя будет использоваться в файле db.py, например: «ml_db»
version: "3"
services:
app:
build:
context: "./app"
dockerfile: "Dockerfile"
args:
AWS_ACCOUNT_ID: 0
BASE_REPO_NAME: python-base
container_name: "my-ml-app-container"
restart: always
depends_on:
- db
environment:
database_username : 'root'
database_password : 'root'
database_ip : 'db'
database_name : 'ml_db'
database_port : '3306'
ENV : 'dev'
ports:
- "5100:8200"
volumes:
- ./app:/app
phpmyadmin:
image: phpmyadmin
restart: always
ports:
- 8035:80
environment:
- PMA_HOST=db
db:
platform: linux/x86_64
image: mysql:5.7
ports:
- "33000:3306"
environment:
MYSQL_ROOT_PASSWORD: root
г) requirements.txt:дляустановкибиблиотек, которые вы использовали для своего приложения ML_app. Пример, как показано ниже:
scikit-learn==0.23.2 pandas==1.2.5 numpy==1.19.2 Flask==2.0.1 sqlalchemy==1.4.20 sqlalchemy-utils mysql-connector==2.2.9 requests
2. app.py: настройка API Flask
Убедитесь, что port=int(8200) . Здесь 8200 совпадает с портами:
— «5100:8200» в docker-compose.yml
from flask import Flask,request,jsonify
import db
import sqlalchemy
from sqlalchemy.orm import scoped_session
import pandas as pd
import _thread
# import datetime
import datetime as dt
import json
import os
import traceback
import io
from abc import ABCMeta, abstractmethod
import models
from ml_model_main import MLmodel
app = Flask(__name__)
database_connection,session_local = db.get_connection()
app.session = scoped_session(session_local)
#app.session = scoped_session(session_local,scopefunc = _app_ctx_stack.__ident_func__)
models.Base.metadata.create_all(bind = app.session.connection().engine)
@app.route("/", methods=['GET'])
def home():
return "ML Version 0 API"
@app.route("/getMLmodel", methods=['GET', 'POST'])
def getMLmodel():
ml_input_dict = {}
ml_input_dict['product_id'] = request.args.get("product_id")
ml_input_dict['date_time'] = request.args.get("date_time")
ml_input_dict['inputA'] = request.args.get("inputA")
ml_input_dict['inputB'] = request.args.get("inputB")
ml_input_dict['test_product_id'] = request.args.get("test_product_id")
ml_input_dict['test_date_time'] = request.args.get("date_time")
ml_input_dict['test_inputA'] = request.args.get("inputA")
ml_input_dict['test_inputB'] = request.args.get("inputB")
parameter_df = pd.DataFrame(ml_input_dict, index=[0])
ml_object = MLmodel(app.session)
ml_object.push_data(parameter_df, 'ml_input')
pred_dict = ml_object.ml_caller(ml_input_dict)
return_dict = {
'product_id':ml_input_dict['product_id'],\
'date_time':pred_dict['date_time'],\
'prediction':pred_dict['prediction'],\
'status_code':'SUCCESS'
}
return return_dict
if __name__ == "__main__":
app.run(host='0.0.0.0', port=int(8200), debug=True)
3. models.py: сборка БД на бэкенде
Этот пример кода для создания 3 таблиц/БД: ml_input, ml_test_input они будут вводиться в модель, а ml_prediction_response будет хранить значения прогноза и функции
from sqlalchemy import Column,Integer,String,Time,Text,Float,ForeignKey, UniqueConstraint, Boolean, ForeignKeyConstraint
from sqlalchemy.types import DateTime
from sqlalchemy.ext.declarative import declarative_base
import datetime
from sqlalchemy.sql import func
Base = declarative_base()
class ml_input(Base):
__tablename__ = 'ml_input'
product_id = Column(Integer, primary_key=True, nullable=False)
date_time = Column(Float, nullable=False)
inputA = Column(Float, nullable=False)
inputB = Column(String(45), nullable=False)
updated_at = Column(DateTime,server_default=func.now())
def __init__(self, product_id,date_time,inputA,inputB, updated_at=func.now()):
self.product_id = product_id
self.date_time = date_time
self.inputA = inputA
self.inputB = inputB
self.updated_at = updated_at
class ml_test_input(Base):
__tablename__ = 'ml_test_input'
test_product_id = Column(Integer, primary_key=True, nullable=False)
test_date_time = Column(Float, nullable=False)
test_inputA = Column(Float, nullable=False)
test_inputB = Column(String(45), nullable=False)
updated_at = Column(DateTime,server_default=func.now())
def __init__(self, test_product_id,test_date_time,test_inputA,test_inputB, updated_at=func.now()):
self.test_product_id = test_product_id
self.test_date_time = date_time
self.test_inputA = inputA
self.test_inputB = inputB
self.updated_at = updated_at
class ml_prediction_response(Base):
__tablename__ = 'ml_prediction_response'
product_id = Column(String(100), nullable=False, primary_key=True)
ml_prediction = Column(Float, nullable=False)
date_time = Column(Float, nullable=False)
feature_1 = Column(Float, nullable=False)
feature_2 = Column(String(20), nullable=False)
updated_at = Column(DateTime,server_default=func.now())
def __init__(self, product_id, ml_prediction, date_time, feature_1, feature_2,\
updated_at=func.now()):
self.product_id = product_id
self.ml_prediction = ml_prediction
self.date_time = date_time
self.feature_1 = feature_1
self.feature_2 = feature_2
self.updated_at = updated_at
4. db.py: связывание вызовов БД с docker-compose.yml
Убедитесь, что следующее совпадает с docker-compose.yml:имя_базы_данных: ml_db и порт_базы_данных: 33000.
import sqlalchemy
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert
import os
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import database_exists, create_database
_connection = None
def get_connection(_app = "Flask"):
_connection = None
if _app == "Flask" :
database_username = os.environ['database_username']
database_password = os.environ['database_password']
database_ip = os.environ['database_ip']
database_name = os.environ['database_name']
database_port = os.environ['database_port']
if _app != "Flask" :
database_username = "root"
database_password = "root"
database_ip = "localhost"
database_name = "ml_db"
database_port = "33000"
db_url = 'mysql+mysqlconnector://{0}:{1}@{2}:{3}/{4}?auth_plugin=mysql_native_password'.format(database_username, database_password,database_ip,database_port,database_name)
if not database_exists(db_url ):
create_database(db_url )
if not _connection:
_connection = sqlalchemy.create_engine(db_url , pool_recycle=3600, pool_size=5, max_overflow=10)
local_session = sessionmaker(autocommit= True,autoflush=False,bind=_connection )
return _connection,local_session
5. main.py: ваша модель машинного обучения, обернутая в класс
import pandas as pd
pd.options.mode.chained_assignment = None
import numpy as np
import math
from sklearn.ensemble import RandomForestClassifier
from helper_function import *
import boto3
from sqlalchemy import create_engine
import sqlalchemy
class MLmodel:
def __init__(self,session):
self.session = session
#Can read data directly from AWS datalake S3 ( raw csv file) with boto
self.s3 = client('s3', aws_access_key_id=self.ACCESS_KEY , aws_secret_access_key=self.SECRET_KEY)
csv_obj_1 = self.s3.get_object(Bucket=self.bucket_name, Key= 'file_name.csv')
self.df_aws = pd.read_csv(StringIO(csv_obj_1['Body'].read().decode('utf-8')))
# can read data directlt from AWS data wharehouse Redshift (structiured DB table) with Sqlalchemy
redshift_engine = create_engine('postgresql://admin:....')
self.df_redshift = pd.read_sql('SELECT * FROM structure_table_name;', redshift_engine)
def push_data(self, df, table_name):
df.to_sql(con=self.session.connection(), name=table_name, if_exists='append', index=False, chunksize=1000)
self.session.commit()
def get_column_names_db(self,table_name):
results = self.session.execute("SHOW columns FROM {};".format(table_name))
results = list(results)
column_names = []
for result in results:
column_names.append(result[0])
return column_names
def ml_prediction(self, lrs_input_dict):
# get raw training input from DB on API call
dat = self.session.execute("select * from ml_input where product_id='{}' ".format(ml_input_dict['product_id']))
self.train = pd.DataFrame(dat, columns = self.get_column_names_db("ml_input")).sort_values(by='date_time')
self.train['date_time'] = pd.to_datetime(self.train['date_time']).dt.tz_localize(None)
self.train = self.train.reset_index()
self.train.sort_values(by=['date_time'],inplace=True
# get test input from DB on API call
test_dat = self.session.execute("select * from ml_test_input where product_id='{}' ".format(ml_input_dict['test_product_id']))
self.test = pd.DataFrame(test_dat, columns = self.get_column_names_db("ml_test_input")).sort_values(by='date_time')
self.test['date_time'] = pd.to_datetime(self.test['date_time']).dt.tz_localize(None)
########### list of features
x_train_full= self.train[['product_id','date_time','inputA']]
x_test= self.test[['test_product_id','date_time','inputA']]
y_train_full = self.train['inputB']
ml_model = RandomForestClassifier(max_depth=2, random_state=0)
ml_model.fit(x_train_full,y_train_full)
y_test = ml_model.predict(x_test)
result=y_test
# Pushing the ML prediction to the DB
prediction_dict = {
'product_id':lrs_input_dict['test_product_id']
'date_time':lrs_input_dict['test_date_time'],
'predcition':result
}
prediction_df = pd.DataFrame(prediction_dict, index=[0])
try:self.push_data(prediction_df, 'ml_prediction_response')
except:print('Unable to push data to the ml_prediction_response DB')
return prediction_dict
def ml_caller(self, lrs_input_dict):
prediction_dict = self.ml_prediction(self.df, ml_input_dict)
return prediction_dict
Наконец-то Dockerize
а) В вашем терминале перейдите в каталог «ML_app», созданный на этапе установки
б) docker-compose up -d — собрать
c) Определите ошибку/ошибку, используя: docker-compose logs -f app
г) Чтобы проверить настройку БД, перейдите в браузере: http://localhost:8035/
api_tester_file.py: проверка результатов и отладка
Обязательно вызовите функцию из app.py в URL: getMLmodel. «http://локальный: 5100/getMLmodel»
import pandas as pd
import requests
simulation_data = pd.read_csv(r'/Users/shivikakbisen/Desktop/test_data.csv')
headers = {
'Content-type': 'application/json'
}
output_df = pd.DataFrame()
for i in range(len(simulation_data)):
# Converting from datetime to epoch time
epoch_time = (pd.to_datetime(simulation_data['device_time'][i]) - pd.to_datetime('1970-01-01 00:00:00')).total_seconds()
ml_input_dict = {
'product_id': simulation_data['shipment_id'][i],
'date_time': epoch_time,
'inputA': simulation_data['inputA'][i],
'inputB': simulation_data['inputB'][i],
'test_product_id': simulation_data['test_product_id'][i],
'test_date_time': simulation_data['test_date_time'][i],
'inputA': simulation_data['inputA'][i],
'inputB': simulation_data['inputB'][i],
}
response = requests.get('http://localhost:5100/getMLmodel',headers= headers, params=lrs_input_dict)
response.json()
print('API response:',response.content ,'\n')
output_df = output_df.append(response.json(), ignore_index=True)
output_df.to_csv('/Users/shivikakbisen/Desktop/test_data_output.csv',index= None)
Развертывание контейнера Docker на AWS EC2
Затем этот каталог ML_app можно зафиксировать в github/bitbutcket. На терминале через репозиторий git EC2 можно настроить с помощью SSH с доступом к собственности.
Развертывание контейнера Docker на AWS EC2 в конвейере Bitbucket
Эластичность и масштабируемость: EC2 позволяет пользователям быстро увеличивать или уменьшать свои вычислительные мощности в зависимости от спроса, гарантируя, что у них достаточно ресурсов для обработки пикового трафика и избегая чрезмерного выделения ресурсов.
