Develop (#64)

* add atmos ETL process
ETL DAG 추가
DB engine관련 오류 발생 수정 중

* Add new train process

MLflow & Prefect & Ray tune을 적용한 새로운 training process 입니다.

* Add save best model logic

실험이 끝난 뒤에 가장 좋은 모델을 찾아 db에 그 정보를 기록합니다.

이후 predict에서는 db조회로 가장 좋은 모델을 찾아 predict를 수행합니다.

* modify atmos ETL pipeline
1. 데이터 요청 후 반환받은 데이터가 없을 시 flow를 종료하도록 설정하였습니다.
2. 데이터 검증 후 조건에 맞지 않는 데이터가 있을 시 저장하지 않고 flow를 종료하도록 하였습니다.
3. cron schedule을 설정하는 코드를 추가하셨습니다.

* Fix save model logic

best모델을 db에  기록할때 artifact_uri 를 저장하는데 artifact_path도 포함하여 저장합니다.

* Add redis

redis를 이용해서 비효율적인 읽어옴을 개선했습니다.

* Add mnist training

mnist classification training process를 추가하였습니다.

* add training model process after data ETL
데이터 수집이 성공적으로 종료되면 수집된 데이터를 포함하여 학습을 진행하고 성능이 좋으면 모델을 교체하는 과정을 추가하였습니다.

* Fix save logic

저장때  run_id를 저장 하는것으로 수정하였습니다.

* Fix save logic

return true false를 위해서 로직 수정중에 있습니다.

* Add knn model train & save

knn 모델을 학습하고 저장합니다.
train_df 는 고정되어 있습니다.
knn 모델을 cnn 모델이 업데이트 된 경우만 학습을하고 저장하게됩니다.
db에 저장하는 것은 가장 최근에 logging된 knn 모델을 저장하게 됩니다.

* add redis caching& modify load model process
1. predict API에 redis를 이용하여 모델을 캐시하는 방법을 추가하였습니다. 모델이 redis에 존재하지 않으면 database에서 받아와서 redis에 저장한 후 일정시간 동안 해당 모델에 대한 예측요청이 없으면 삭제합니다.
2. artifact path 대신 run id를 사용하여 모델을 불러오게 변경하였습니다.

* Modify predict redis
오작동하는부분 수정

* Add redis update time logic
1. redis로 모델을 캐싱하는 알고리즘을 개선하였습니다.
기존방식: 모델을 캐싱한 후 일정시간이 지나면 삭제되게 하였습니다.
개선방식: 캐싱된 모델이 예측요청을 받으면 만료되는 시간이 초기화되게 하였습니다.

* Add mnist prediction route

mnist 모델을 불러와서 예측하는 route를 추가하였습니다.
임시적으로 가지고 있던 파일에서 predict를 진행하지만 추후 input값을 받아서 예측하도록 수정 예정입니다.

++ model training process 에서 jit script를 사용하여서 불러와 사용하는데에 문제 없게 하였습니다.
++ training process에서 task파일 안에 model class가 존재하게되면 model 사용에 에러가 발생해 분리하였습니다.

* Add redis at mnist prediction

redis 사용에 있어서 pickle로 밀어넣을 수 없어서 save_to_buffer 로 넣고 bytes로 읽어옵니다.

* Add redis connection pool
redis의 connection pool을 도입하였습니다.

* Fix data load & cleanup

하드코딩되어있던 부분을 조금 수정했습니다. & 불러오는 로직을 재사용성을 높이기위해 수정했습니다.

* Fix data path

data 를 storage에서 읽어오는것으로 변경하였습니다.

* Modify atmos predict api
1. redis에서 ML모델을 불러오는 코드가 중첩되어 있던 부분을 삭제하였습니다.

* Test performance according to method
1. postgres DB에서 모델 로드, redis에서 모델 로드, serialize, deserialize 각각에서 시간이 얼마나 걸리는지 테스트 하기 위하여 시간 측정 코드를 추가해 두었습니다.

* Modify caching algorithm
1. 모델을 redis나 기타 DB에 caching하지 않는 방법으로 변경하였습니다. model_timer class를 만들어 instance variable로 저장하고 일정 시간이 지나면 삭제되게 하였습니다. 모델을 직렬화할 필요성이 없어져 inference 속도가 크게 개선되었습니다.
2. caching되지 않은 모델을 mlflow에서 처음 불러오는 부분은 여전히 느립니다.

* Modify cache class
1. cache class를 좀 더 범용적인 이름으로 변경하였습니다.
2. 기능상의 변경은 없습니다.
3. predict method를 삭제하였습니다.

* Fix mnist model caching

redis 사용해서 읽어오던 상황에서 코드를 개선했습니다.
redis를 사용해 모델을 캐싱하려면 serialize, deserialize 하는 과정이 필요합니다.
그 과정에서 시간이  너무 오래 걸리기 때문에 개선의 필요성을 느껴 직접 관리하는 코드를 사용합니다.
모델을 읽어오고 캐싱하는 부분을 class로 묶어서 모아놓았습니다.

* Fix load model load logic

model load하고 캐싱하는 부분에서 lock을 걸고 진행합니다.
data도 임시로 캐싱해서 사용하도록 하였습니다.

* Remove unnecessary code

test를 위한 코드를 삭제하였습니다.

* Delete redis
레디스 삭제하였습니다.
비동기함수 잘못 작성된 부분 수정하였습니다.

* Update README

* Add prefect working directory
1. prefect agent의 실행 위치에 따라 경로를 찾을 수 없는 문제를 해결하였습니다.
2. flow에 working directory를 명시해주면 됩니다.

* Add more metrics to Mnist

Mnist train과정에서 class별 accuracy를 측정할 수 있도록 추가하였습니다.

* Add pipeline option

여러 파이프라인이 동시에 적용될수있도록 insurance pipeline도 업데이트하였습니다.

* Update insurance save logic

insurance 부분도 다른것들과 마찬가지로 run_id를 저장하는 형식으로 업데이트 하였습니다.

* Add insuracne predict

insurance predict하는것도 다른것과 동일한 방법으로 진행합니다.

* Fix task decorator

test와중에 주석처리되었던 @task를 주석해제하였습니다.

* Remove resource_per_trial

resource_per_trial을 명시했을때 계속 pending상태에 머무르는 현상이 있어 삭제하였습니다.

* Fix mlflow-url

mlflow default url을 수정하였습니다.

* Add git action to build docker containers
1. 도커파일을 빌드하기 위한 깃 액션을 추가하였습니다.
2. 일부 잘못 설정되어있던 호스트 이름을 수정하였습니다.
3. requirements.txt를 최신화 하였습니다.

* Remove PR condition

Build API server container 하는 부분에서 PR일때 상황은 제거하였습니다.

* Modify mnist prediction

Mnist prediction 부분을 input을 받아서 예측하도록 바꾸었습니다.
run_in_threadpool로 predict부분을 묶어두었습니다.
return을 세분화 하였습니다.

* Add kubernetes yaml files
1. 쿠버네티스 설정파일들을 추가하였습니다.
2. prefect_Dockerfile을 수정하였습니다.

* Remove testing branch

test용으로 on push 트리거 브랜치에  feature/kubernetes 도 포함시켜두어서 제외하였습니다.

* Add deprecated folder

deprecated 된 것들은 폴더에 모아서 기록하였습니다.

* Delete experiments and import train code
1. 현재 사용하지 않는 experiments 폴더를 삭제하였습니다.
2. main.py에서 train api를 import하여 에러가 나던 부분을 수정하였습니다.

* Modify Dockerfile name
1. 기존 xxx_Dockerfile 형식으로 되어있던 파일 이름을 Dockerfile.xxx 형식으로 바꾸었습니다. 이러한 형식으로 저장하면 파일의 목록을 출력했을 때 도커파일이 모여있게 되므로 가독성이 올라갈 것으로 생각됩니다.

* Add load type

Model load 방식을 최고성능 모델을 가져오는것 외에
production으로 등록된 모델을 가져오는 방식을 추가하였습니다.
mlflow로 모델 production 및 staging을 관리할 경우에 이 방법이 사용될 수 있습니다.

* Fix data load logic

data로드를 환경변수에서 경로를 읽어서 하지않고 db에서 버전별
그리고 실험별로 읽을 수 있도록 하였습니다.

* Modify Mnist Train

출력층 이전 레이어에서 64개의 특징을 추출합니다.
model의 xai 특징을 만들때 출력층을 제거하지 않은 모델 예측도 함께 특징으로 사용해서 74개의 특징을 knn으로 학습합니다.

* Modify Mnist Predict

train부분의 변화에 대응해 수정되었습니다.

* Add mnist avg metadata

pixel 평균값에 대한 정보를 추가하였습니다.

prediction 때에도 input으로 들어온 데이터에대한 pixel 평균값을 로깅합니다.

* Add is_cloud parameter

is_cloud parameter가 누락되어있어 추가하였습니다.

* Add Continuous deploy process
- rollout 해주는 부분을 추가함으로써 무중단 배포할 수 있게 되었습니다.

* Modify Git Action workflows
-  continuous deployments를 위해 필요한 deployments만 재시작 되도록 분리하였습니다.

* Modify Git Action workflows
-  continuous deployments를 위해 필요한 kubernetes deployments만 재시작 되도록 분리하였습니다.

* Feature/readme (#57)

Update README

* Feature/data load (#58)

update data load

* Update phase2.md

* Feature/readme (#59)

Update README

* Feature/readme (#61)

* Update README

README에내용을 추가하였습니다.

* Add phase1 info

Phase1 에 대한 내용을 조금 추가하였습니다.

* Add phase1 info

* Modify phase1 info

* Add phase2-local

* Add logos
로고추가

* Resize images

* Add requirements.sh

* Modify main readme page
그림도 넣구 구조도 조금 바꾸었습니당

* Delete phase2.PNG

* Add figure
phase2 그림 추가하였습니다.

* Modify readme file
내용의 순서를 변경하였습니다.

* Remove local.md

* Add info phase2

* Add frontend link

* Modify readme
프로젝트 소개부분 수정하였습니다.

* Modify readme
도커 컨테이너를 실행하는 부분에 대한 설명을 추가하였습니다.

* Add readme

* Update phase2.md

* Update README.md

* Delete kubernetes nodeselector
1. deployments에 nodeSelector가 설정되어있던 부분을 삭제했습니다.

* Modify readme.md
1. readme file의 전체적인 구조를 변경하였습니다.
2. Phase2의 세부내용을 변경된 구조에 맞추어 다시 작성하였습니다.
3. 아직 작성중입니다ㅜㅜ

* Modify README.md
1. readme 에서 phase2 프로젝트를 설명한 부분을 수정하였습니다.

Co-authored-by: ehddnr301 <[email protected]>

Co-authored-by: ehddnr301 <[email protected]>

.github/workflows/build_apiserver.yaml

@@ -0,0 +1,52 @@
+name: Build API server container
+on:
+  push:
+    branches: [ main ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check Out Repo 
+        uses: actions/checkout@v2
+
+      - name: Login to Docker Hub
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          password: ${{ secrets.DOCKER_HUB_ACCESS_TOKEN }}
+
+      - name: Set up Docker Buildx
+        id: buildx
+        uses: docker/setup-buildx-action@v1
+
+      - name: Build and push api-server
+        id: api-server
+        uses: docker/build-push-action@v2
+        with:
+          context: ./
+          file: ./Dockerfile.fastapi
+          push: true
+          tags: ${{ secrets.DOCKER_HUB_USERNAME }}/mlops-project:api-server-1.0
+
+      - name: Build and push prefect-worker
+        id: prefect-worker
+        uses: docker/build-push-action@v2
+        with:
+          context: ./
+          file: ./Dockerfile.prefect
+          push: true
+          tags: ${{ secrets.DOCKER_HUB_USERNAME }}/mlops-project:prefect-worker-1.0
+
+      - name: Image digest
+        run: echo ${{ steps.docker_build.outputs.digest }}
+
+      - name: Deploy
+        uses: appleboy/ssh-action@master
+        with:
+          host: ${{ secrets.REMOTE_IP }}
+          username: ${{ secrets.REMOTE_SSH_ID }}
+          port: ${{ secrets.REMOTE_SSH_PORT }}
+          key: ${{ secrets.REMOTE_SSH_KEY }}
+          script: |
+            kubectl rollout restart -f ./MLOps/k8s/prepi_deployments.yaml

.gitignore

@@ -4,3 +4,8 @@ __pycache__
 tf_model/**/*
 log.txt
 experiments/**/temp/
+.ssl/
+prefect/atmos_tmp_pipeline/ray_mlflow
+prefect/atmos_tmp_pipeline/*.sh
+mlruns
+exp_models

.pre-commit-config.yaml

@@ -8,4 +8,5 @@ repos:
     rev: 5.6.4
     hooks:
       - id: isort
-        language_version: python3
+        language_version: python3
+        args: ["--profile", "black"]

Dockerfile.baseimage

@@ -0,0 +1,8 @@
+FROM python:3.8
+
+COPY requirements.txt /requirements.txt
+
+RUN pip install --upgrade pip &&\
+    pip install --no-cache-dir -r requirements.txt &&\
+    pip uninstall -y tensorflow==2.6 &&\
+    pip install --no-cache-dir tensorflow-cpu==2.4

Dockerfile.fastapi

@@ -0,0 +1,7 @@
+FROM hl8469/mlops-project:base-image-1.0
+
+COPY . /
+
+EXPOSE 8000
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "3"]

Dockerfile.prefect

@@ -0,0 +1,8 @@
+FROM hl8469/mlops-project:base-image-1.0
+
+COPY ./prefect /prefect
+COPY ./set_prefect.sh /
+
+RUN prefect backend cloud
+
+CMD /set_prefect.sh

app/api/schemas.py → app/api/data_class.py

@@ -29,3 +29,7 @@ class ModelCorePrediction(BaseModel):
 class ModelCore(ModelCoreBase):
     class Config:
         orm_mode = True
+
+
+class MnistData(BaseModel):
+    mnist_num: str

app/api/router/predict.py

@@ -1,18 +1,34 @@
 # -*- coding: utf-8 -*-
+import ast
+import asyncio
+import os
 from typing import List
 
+import mlflow
 import numpy as np
+import pandas as pd
+import torchvision.transforms as transforms
+import xgboost as xgb
+from dotenv import load_dotenv
 from fastapi import APIRouter
 from starlette.concurrency import run_in_threadpool
 
-from app import models
-from app.api.schemas import ModelCorePrediction
+from app import schema
+from app.api.data_class import MnistData, ModelCorePrediction
 from app.database import engine
-from app.utils import ScikitLearnModel, my_model
+from app.query import SELECT_BEST_MODEL
+from app.utils import CachingModel, VarTimer, load_data, softmax
 from logger import L
 
-models.Base.metadata.create_all(bind=engine)
+load_dotenv()
 
+schema.Base.metadata.create_all(bind=engine)
+
+host_url = os.getenv("MLFLOW_HOST")
+mlflow.set_tracking_uri(host_url)
+reset_sec = 5
+CLOUD_STORAGE_NAME = os.getenv("CLOUD_STORAGE_NAME")
+CLOUD_VALID_MNIST = os.getenv("CLOUD_VALID_MNIST")
 
 router = APIRouter(
     prefix="/predict",
@@ -21,66 +37,136 @@
 )
 
 
-@router.put("/insurance")
-async def predict_insurance(info: ModelCorePrediction, model_name: str):
-    """
-    정보를 입력받아 보험료를 예측하여 반환합니다.
-
-    Args:
-        info(dict): 다음의 값들을 입력받습니다. age(int), sex(int), bmi(float), children(int), smoker(int), region(int)
-
-    Returns:
-        insurance_fee(float): 보험료 예측값입니다.
-    """
-
-    def sync_call(info, model_name):
-        """
-        none sync 함수를  sync로 만들어 주기 위한 함수이며 입출력은 부모 함수와 같습니다.
-        """
-        model = ScikitLearnModel(model_name)
-        model.load_model()
-
-        info = info.dict()
-        test_set = np.array([*info.values()]).reshape(1, -1)
-
-        pred = model.predict_target(test_set)
-        return {"result": pred.tolist()[0]}
+mnist_model = CachingModel("pytorch", 600)
+knn_model = CachingModel("sklearn", 600)
+data_lock = asyncio.Lock()
+train_df = VarTimer(600)
+
+
+@router.put("/mnist")
+async def predict_mnist(item: MnistData):
+    global train_df
+    global mnist_model, knn_model
+
+    item2 = np.array(ast.literal_eval(item.mnist_num)).astype(np.uint8)
+    model_name = "mnist"
+    model_name2 = "mnist_knn"
+    is_cloud = False
+    data_version = 1
+    exp_name = 'mnist'
+
+    if not isinstance(train_df._var, pd.DataFrame):
+        async with data_lock:
+            if not isinstance(train_df._var, pd.DataFrame):
+                df, _ = load_data(is_cloud, data_version, exp_name)
+                train_df.cache_var(df)
+
+    transform = transforms.Compose(
+        [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
+    )
+    reshaped_input = item2.reshape(28, 28)
+    transformed_input = transform(reshaped_input)
+    transformed_input = transformed_input.view(1, 1, 28, 28)
+
+    await mnist_model.get_model(model_name, load_type="production")
+    await knn_model.get_model(model_name2, load_type="production")
+
+    def sync_call(mnist_model, knn_model, train_df):
+        # Net1
+        result = mnist_model.predict(transformed_input)
+        p_res = softmax(result.detach().numpy()) * 100
+        percentage = np.around(p_res[0], 2).tolist()
+        # Net2
+        result = mnist_model.predict(transformed_input, True)
+        result = np.concatenate((result.detach().numpy(), np.array(percentage).reshape(1,-1) / 10), axis=1)
+        # KNN
+        knn_result = knn_model.predict(result)
+        xai_result = train_df.get_var().iloc[knn_result, 1:].values[0].tolist()
+        return {
+            "result": {
+                "percentage": percentage,
+                "answer": percentage.index(max(percentage)),
+                "xai_result": xai_result,
+            },
+            "error": None,
+        }
 
     try:
-        result = await run_in_threadpool(sync_call, info, model_name)
+        result = await run_in_threadpool(
+            sync_call, mnist_model, knn_model, train_df
+        )
         L.info(
-            f"Predict Args info: {info}\n\tmodel_name: {model_name}\n\tPrediction Result: {result}"
+            f"Predict Args info: {item.mnist_num}\n\tmodel_name: {model_name}\n\tPrediction Result: {result}\n\tcolor_avg_{result['result']['answer']}: {np.round(np.mean(item2), 2)}"
         )
         return result
-
     except Exception as e:
         L.error(e)
         return {"result": "Can't predict", "error": str(e)}
 
 
-@router.put("/atmos")
-async def predict_temperature(time_series: List[float]):
+insurance_model = CachingModel("xgboost", 30)
+
+
+@router.put("/insurance")
+async def predict_insurance(info: ModelCorePrediction):
+    info = info.dict()
+    test_set = xgb.DMatrix(np.array([*info.values()]).reshape(1, -1))
+
+    model_name = "insurance"
+    await insurance_model.get_model(model_name, load_type="production")
+    result = insurance_model.predict(test_set)
+
+    result = float(result[0])
+    return {
+        "result": result,
+        "error": None,
+    }
+
+
+lock = asyncio.Lock()
+atmos_model_cache = VarTimer()
+
+
+@router.put("/atmos_temperature")
+async def predict_temperature_(time_series: List[float]):
     """
     온도 1시간 간격 시계열을 입력받아 이후 24시간 동안의 온도를 1시간 간격의 시계열로 예측합니다.
-
     Args:
         time_series(List): 72시간 동안의 1시간 간격 온도 시계열 입니다. 72개의 원소를 가져야 합니다.
-
     Returns:
         List[float]: 입력받은 시간 이후 24시간 동안의 1시간 간격 온도 예측 시계열 입니다.
     """
+
+    global lock
+
     if len(time_series) != 72:
         L.error(f"input time_series: {time_series} is not valid")
         return {"result": "time series must have 72 values", "error": None}
 
+    model_name = "atmos_tmp"
+
+    if not atmos_model_cache.is_var:
+        async with lock:
+            if not atmos_model_cache.is_var:
+                run_id = engine.execute(
+                    SELECT_BEST_MODEL.format(model_name)
+                ).fetchone()[0]
+                print("start load model from mlflow")
+                atmos_model_cache.cache_var(
+                    mlflow.keras.load_model(f"runs:/{run_id}/model")
+                )
+                print("end load model from mlflow")
+
     def sync_pred_ts(time_series):
         """
         none sync 함수를  sync로 만들어 주기 위한 함수이며 입출력은 부모 함수와 같습니다.
         """
-        time_series = np.array(time_series).reshape(1, -1, 1)
-        result = my_model.predict_target(time_series)
+
+        time_series = np.array(time_series).reshape(1, 72, 1)
+        result = atmos_model_cache.get_var().predict(time_series)
+        atmos_model_cache.reset_timer()
         L.info(
-            f"Predict Args info: {time_series.flatten().tolist()}\n\tmodel_name: {my_model.model_name}\n\tPrediction Result: {result.tolist()[0]}"
+            f"Predict Args info: {time_series.flatten().tolist()}\n\tmodel_name: {model_name}\n\tPrediction Result: {result.tolist()[0]}"
         )
 
         return {"result": result.tolist(), "error": None}