<file_sep>version: '3.3'

services:
    redis:
        image: redis
        ports:
            - '6379:6379'
    db:
        restart: always
        image: postgres:latest
        container_name: mlflow_db
        expose:
            - ${DB_PORT}
        environment:
            - POSTGRES_DB=${DB_NAME}
            - POSTGRES_USER=${DB_USER}
            - POSTGRES_PASSWORD=${DB_PW}
        volumes:
            - db_datapg:/var/lib/postgresql/data

    mlflow:
        restart: always
        build: ./mlflow
        image: mlflow_server
        container_name: mlflow_server
        ports:
          - 5000:5000
        environment:
            - BACKEND=postgresql://${DB_USER}:${DB_PW}@db:${DB_PORT}/${DB_NAME}
            - ARTIFACTS=/mlruns
        volumes:
            - mlrun_data:/mlruns
        command:
            - sh
            - -c
            - mlflow server
                --port ${MLFLOW_PORT}
                --host 0.0.0.0
                --backend-store-uri $${BACKEND}
                --default-artifact-root $${ARTIFACTS}
        depends_on:
            - db
    celery:
        build: ./celery
        volumes:
            - ./celery:/celery
        depends_on:
            - redis
    app:
        build: ./app
        volumes:
          - ./app:/app
        ports:
          - 9091:8000
        depends_on:
          - celery
        expose:
          - 9091
volumes:
    db_datapg:
    mlrun_data:
<file_sep>from celery import Celery
from celery.result import AsyncResult
from flask import (
    Flask,
    make_response,
    render_template,
    request,
)


celery_app = Celery('tasks', backend='redis://redis', broker='redis://redis')
app = Flask(__name__)


@app.route('/iris', methods=['GET', 'POST'])
def iris_handler():
    if request.method == 'POST':
        sl = request.form['sl']
        sw = request.form['sw']
        pl = request.form['pl']
        pw = request.form['pw']
        sample = [[int(sl), int(sw), int(pl), int(pw)]]

        task = celery_app.send_task('tasks.predict', sample)

        return render_template('check.html', value=task.id)


@app.route('/iris/<task_id>', methods=['GET', 'POST'])
def iris_check_handler(task_id):
    task = AsyncResult(task_id, app=celery_app)
    if task.ready():
        labels = {
            0: 'iris-setosa',
            1: 'iris-versicolor',
            2: 'iris-virginica',
        }
        text = f'DONE, result {labels[int(task.result[0])]}'
    else:
        text = 'IN_PROGRESS'
    response = make_response(text, 200)
    response.mimetype = "text/plain"
    return response


@app.route('/')
def home():
    return render_template('index.html')


if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=8000)
<file_sep>FROM python:3.7
ADD . /celery
WORKDIR /celery
RUN pip install -r requirements.txt
CMD celery -A tasks worker --loglevel=info
<file_sep>import joblib

import mlflow
from sklearn.datasets import load_iris
from sklearn.pipeline import Pipeline
from xgboost import XGBClassifier
from sklearn.base import (
    BaseEstimator,
    TransformerMixin,
)


class ModelTransformer(TransformerMixin, BaseEstimator):
    def __init__(self, mdl):
        self.mdl = mdl

    def fit(self, X, y=None):
        return self.mdl.fit(X, y)

    def transform(self, X):
        return self.mdl.predict(X)


class JSONResponse(TransformerMixin, BaseEstimator):
    def __init__(self, key_name):
        self.key_name = key_name

    def fit(self, X, y=None):
        return X

    def transform(self, X):
        result = {self.key_name: X[0]}
        return result

    def predict(self, X):
        return self.transform(X)


if __name__ == '__main__':
    MLFLOW_SERVER_URL = 'http://127.0.0.1:5000/'
    data = load_iris()
    params = {
        "learning_rate": 0.05,
        "n_estimators": 1000,
        "max_depth": 3,
        "objective": 'multi:softmax',
        "num_class": 3,
    }
    mdl = XGBClassifier(**params)
    mdl.fit(data['data'], data['target'])
    mdl_api = Pipeline([("mdl", ModelTransformer(mdl)), ("pack", JSONResponse('iris'))])
    joblib.dump(mdl, 'iris.mdl')
    with mlflow.start_run(run_name='Experiment 1'):
        mlflow.sklearn.log_model(mdl_api, 'prod_mdl')
<file_sep>FROM python:3.7

ENV MLFLOW_HOME /opt/mlflow
ENV SERVER_PORT 5000
ENV SERVER_HOST 0.0.0.0
ENV FILE_STORE ${MLFLOW_HOME}/fileStore
ENV ARTIFACT_STORE ${MLFLOW_HOME}/artifactStore

RUN pip install mlflow psycopg2 && \
    mkdir -p ${MLFLOW_HOME}/scripts && \
    mkdir -p ${FILE_STORE} && \
    mkdir -p ${ARTIFACT_STORE}

EXPOSE ${SERVER_PORT}/tcp

WORKDIR ${MLFLOW_HOME}
<file_sep># LSML2
## LSML2 final project by <NAME> 08.09.2021

Dataset: https://scikit-learn.org/stable/modules/generated/sklearn.datasets.load_iris.html

Project goals: Create an ML-service to solve any kind of task.

### Architecture:
- Model: trained on iris dataset XGBClassifier model

### Containers:
1) app: Flask web server to serve simple web interface
2) celery: Celery worker to asynchronously serve model
3) redis: For storing celery tasks
4) mlflow: MLFlow for tracking models
5) db: Database server for mlflow

### Installing:
1) Docker https://docs.docker.com/engine/install/
2) Docker compose https://docs.docker.com/compose/install/
4) To train model run: `pip install -r requirements.txt`, then `python train.py`

### Running:
`sudo docker-compose up --build`

##### Web interface:
User Interface - `http://localhost:9091/`
MLFLow web interface - `http://127.0.0.1:5000/#/`
<file_sep>amqp==5.0.6
billiard==3.6.4.0
cached-property==1.5.2
celery==5.1.2
click==7.1.2
click-didyoumean==0.0.3
click-plugins==1.1.1
click-repl==0.2.0
Flask==2.0.1
importlib-metadata==4.8.1
itsdangerous==2.0.1
Jinja2==3.0.1
kombu==5.1.0
MarkupSafe==2.0.1
prompt-toolkit==3.0.20
pytz==2021.1
redis
six==1.16.0
typing-extensions==3.10.0.2
vine==5.0.0
wcwidth==0.2.5
Werkzeug==2.0.1
zipp==3.5.0
scikit-learn~=0.24.2
xgboost~=1.4.2
mlflow<file_sep>import joblib
import numpy as np
from celery import Celery


celery_app = Celery('tasks', backend='redis://redis', broker='redis://redis')
mdl = joblib.load('iris.mdl')


@celery_app.task(name='tasks.predict')
def predict(data):
    result = mdl.predict(np.array([data])).tolist()
    return result
