[MLflow] MLflow란? + MLflow 사용법

1. MLflow란 무엇이고 왜 쓰는 것인가?

MLflow는 통상적인 머신러닝 워크플로우에서 모델링 및 훈련, 평가, 배포 부분을 도와주는 툴이라고 생각하면 쉽다.

아래의 그림에서 형광색 사각형의 부분을 주로 도와준다고 볼 수 있다.

2. MLflow에서 제공하는 기능

MLflow는 4가지 컴포넌트로 구성되어 있고 각각의 컴포넌트는 다음과 같은 기능을 제공한다.

1. Tracking : 트래킹은 모델 트레이닝 실험 과정에서 나오는 메타데이터, 데이터, 결과물등을 저장할 수 있는 기능이다. 저장 된 내용을 통해 실험을 추적 할 수 있게 도와준다.
2. Projects : MLflow를 사용한 프로젝트를 yaml파일을 통해 패키징 하여 다른 플랫폼에서도 쉽게 사용할 수 있도록 도와준다.
3. Models : 다양한 ML라이브러리를 통해 학습된 모델을 쉽게 배포할 수 있도록 도와준다.
4. Model Registry : 학습된 모델을 등록하여 버저닝 및 라이프사이클을 관리 할 수 있도록 도와준다.

 

위의 네가지 컴포넌트가 제공하는 기능이 조금 이해하기 어려울 수 있다.

 

간단하게 말하면 MLflow는 학습된 ML Model을 저장 할 수 있고(+ 학습중에 나오는 각종 데이터)

저장된 모델을 API를 통해 Serving할 수 있게 도와준다.

 

 

3. MLflow 사용 예제

사용 시나리오

1. ML framework Keras를 통해 간단하게 모델을 학습한다.

2. 학습 된 모델을 MLflow를 통해 저장한다.

3. 저장된 모델을 MLflow를 통해 Serving 한다.

 

1. ML framework Keras를 통해 간단하게 모델을 학습한다.

먼저 mlflow를 설치해준다.

pip install mlflow

keras와 keggle의 titanic train.csv 데이터를 이용하여 모델을 학습시킨다.

(https://www.kaggle.com/c/titanic/data) 여기서 train.csv를 다운받으면 된다.

 

이러한 column을 갖는 tarin.csv 파일을 ./data/train.csv에 저장한 뒤 keras를 이용하여 학습시켜준다.

 

2. 학습 된 모델을 MLflow를 통해 저장한다.

학습이 완료된 후 MLflow API를 통해 model을 저장하면 프로젝트내에 mlruns가 생긴다

 

학습이 완료되면 local에 model이 저장되고 mlflow ui를 통해 학습의 결과 및 metric등을 확인할 수 있다.

터미널에서 mlflow ui 실행 후 http://127.0.0.1:5000 로 접속하면 아래와 같이 실험내역을 볼 수 있다.

 

3. 저장된 모델을 MLflow를 통해 Serving 한다.

학습 된 모델을 서빙할 때에는 run id를 이용하여 다음과 같이 터미널에 입력한 뒤 curl을 보내면 된다.

 

모델 서빙 (5001번 포트 사용)

mlflow models serve -m ./mlruns/0/40428715c14446be866f06b973987af6/artifacts/tf_keras_model -p 5001

 

Prediction 요청

curl -X POST -H "Content-Type:application/json; format=pandas-split" --data '{"columns": ["Sex", "Age_band", "Pclass"], "data": [[1,1,1],[1,1,1]]}' http://localhost:5001/invocations

 

Prediction 결과

[{"0": 0.9737244248390198}, {"0": 0.9737244248390198}]

 

 

예제 코드

import mlflow
import os
import pandas as pd
import numpy as np

from mlflow import keras as ml_keras
from mlflow import log_metrics, log_params
from sklearn.ensemble import RandomForestClassifier

from tensorflow.keras.layers import Dense, Dropout, Input
from tensorflow.keras.models import Model

os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'


def _get_keras_model():
    inp = Input(shape=(3,), name='inp_layer')
    dense_layer_1 = Dense(32, activation='relu', name="dense_1")
    dense_layer_2 = Dense(16, activation='relu', name="dense_2")
    predict_layer = Dense(1, activation='sigmoid', name='predict_layer')

    dense_vector_1 = dense_layer_1(inp)
    dense_vector_2 = dense_layer_2(dense_vector_1)
    predict_vector = predict_layer(dense_vector_2)

    model = Model(inputs=inp, outputs=predict_vector)
    model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['acc'])
    return model


def _get_rf_model(n_estimator):
    return RandomForestClassifier(n_estimators=n_estimator, max_depth=5)


def run_keras_modeling(X, y):
    model = _get_keras_model()
    model.fit(X, y, epochs=20, batch_size=10)

    model_info = {
        'score': {
            'model_score': np.float64(round(model.evaluate(X, y)[1], 2))
        },
        'params': {'epochs': 20, 'batch_size': 10}
    }

    return model, model_info


class TitanicML():
    def __init__(self):
        pass

    def run(self):
        project_path = os.getcwd()
        titanic_path = f"{project_path}/data/"

        data = self._get_data_from_csv(titanic_path)
        data = self.run_preprocessing(data)
        X, y = self._get_X_y(data)

        model, model_info = run_keras_modeling(X, y)

        return model, model_info

    def _get_data_from_csv(self, path):
        return pd.read_csv(f'{path}/train.csv')

    def _get_X_y(self, data):
        X = data[data.columns[1:]]
        X = X[['Sex', 'Age_band', 'Pclass']]
        y = data['Survived']

        return X, y

    def run_preprocessing(self, data):
        data = self._set_initial(data)
        data = self._set_fill_na(data)
        data = self._set_feature(data)
        data = self._set_replace(data)

        return data

    def _set_fill_na(self, data):
        data.loc[(data['Age'].isnull()) & (data['Initial'] == 'Mr'), 'Age'] = 33
        data.loc[(data['Age'].isnull()) & (data['Initial'] == 'Master'), 'Age'] = 5
        data.loc[(data['Age'].isnull()) & (data['Initial'] == 'Mrs'), 'Age'] = 36
        data.loc[(data['Age'].isnull()) & (data['Initial'] == 'Miss'), 'Age'] = 22
        data.loc[(data['Age'].isnull()) & (data['Initial'] == 'Other'), 'Age'] = 46
        data['Embarked'].fillna('S', inplace=True)

        return data

    def _set_initial(self, data):
        data['Initial'] = 0
        data['Initial'] = data['Name'].str.extract('([A-Za-z]+)\.')
        data['Initial'].replace(
            ['Mlle', 'Mme', 'Ms', 'Dr', 'Major', 'Lady', 'Countess', 'Jonkheer', 'Col', 'Rev', 'Capt', 'Sir', 'Don','Dona'],
            ['Miss', 'Miss', 'Miss', 'Mr', 'Mr', 'Mrs', 'Mrs', 'Other', 'Other', 'Other', 'Mr', 'Mr', 'Mr', 'Other'],
            inplace=True)

        return data

    def _set_feature(self, data):
        data['Fare'] = data["Fare"].map(lambda i: np.log(i) if i > 0 else 0)
        data['Age_band'] = 0
        data['Alone'] = 0
        data['Family_Size'] = 0

        data.loc[data['Age'] <= 16, 'Age_band'] = 0
        data.loc[(data['Age'] > 16) & (data['Age'] <= 32), 'Age_band'] = 1
        data.loc[(data['Age'] > 32) & (data['Age'] <= 48), 'Age_band'] = 2
        data.loc[(data['Age'] > 48) & (data['Age'] <= 64), 'Age_band'] = 3
        data.loc[data['Age'] > 64, 'Age_band'] = 4

        data['Family_Size'] = data['Parch'] + data['SibSp']

        data.loc[data.Family_Size == 0, 'Alone'] = 1

        return data

    def _set_replace(self, data):
        data['Sex'].replace(['male','female'],[0,1],inplace=True)
        data['Embarked'].replace(['S','C','Q'],[0,1,2],inplace=True)
        data['Initial'].replace(['Mr','Mrs','Miss','Master','Other'],[0,1,2,3,4],inplace=True)
        data.drop(['Name', 'Age', 'Ticket', 'Cabin', 'PassengerId'], axis=1, inplace=True)

        return data


if __name__ == "__main__":
    titanic = TitanicML()
    tf_model, model_info = titanic.run()
    log_metrics(model_info['score'])
    log_params(model_info['params'])
    ml_keras.log_model(tf_model, 'tf_keras_model')
    print("Model saved in run id : %s" % mlflow.active_run().info.run_uuid)