How to Deploy ML Model to Production

This article talks about 5 easy steps to deploy machine learning and deep learning model into production as a micro-service Flask app. Mostly we will cover the local setup and testing by dockerizing the ML/DL model and then integrating with the AWS cloud framework for data base/ data pipeline integration and bitbucket automated ML deployment setup.

Dockerize the ML/DL model

Dockerizing the ML model is the first step toward deploying your ML model. Following is a clear and concise guide with a sample code template for the setup. This setup will help you test your ML model and get a response in Flask API

1. Setup

a) Create Directory

ML_app/
├── app
│   └── main.py
│   └── app.py
|   └── db.py
|   └── models.py
|   └── requirements.txt
|   └── Dockerfile
└── docker-compose.yml

b) Dockerfile: Describes how to create an image. Create a Docker image of your ML_app. It has no extension like txt etc

Make sure bash as “8200” i.e port number

FROM python:3.9-buster

ENV PYTHONUNBUFFERED 1

COPY . /app #docker files runs here

WORKDIR /app

RUN pip install -r requirements.txt

CMD bash -c "python app.py runserver 0.0.0.0:8200"

c) docker-compose.yml: Describes how the container should run. Tells how to run the image (created by dockerfile) and details like cpu and memory used. Tie the different services like Flask API and database together.

Make sure to change the following in the below:

i) container_name: “my-ml-app-container”

ii) database_name : Same name will be used in db.py file, eg: “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

d) requirements.txt : To install libraries that you’ve used for your ML_app. Example as below:

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: Flask API setup

Make sure port=int(8200) . Here 8200 is same as ports:
— “5100:8200” in 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: Build DB at the backend

This example code to create 3 tables/DB : ml_input, ml_test_input these will be input to model and ml_prediction_response will store prediction values and features

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: Tie DB calls with docker-compose.yml

Make sure following is the same as in docker-compose.yml: database_name: ml_db and database_port: 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: Your ML model wrapped in a class

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

Finally Dockerize

a) In your terminal, cd to the directory “ML_app” created in the setup stage

b) docker-compose up -d — build

c) Identify bug/error using: docker-compose logs -f app

d) To check the DB setup, in your browser visit: http://localhost:8035/

api_tester_file.py: Check results & debug

Before running the api_tester_file, check whether all the DB are created under ml_db, correctly by going to http://localhost:8035/ in your browser.

Make sure to call function from app.py in URL : getMLmodel. ‘http://localhost: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)

Deploying Docker Container to AWS EC2

This ML_app directory can then be committed to github/bitbutcket. On the terminal through the git repo, EC2 can be setup using SSH with ownership access.

Deploying Docker containers with AWS EC2 Instance.

Deploying Docker Container to AWS EC2 into Bitbucket Pipeline

https://gokturksaka.medium.com/deploy-docker-container-to-aws-ec2-with-bitbucket-pipeline-a25a9a1e28a2

Elasticity and Scalability: EC2 allows users to quickly scale up or down their computing capacity based on demand, ensuring that they have enough resources to handle peak traffic and avoiding over-provisioning of resources.