[MLflow] MLflow Project

 

 

MLflow Project는 주로 규칙을 기반으로 재사용 및 재현 가능한 방식으로 데이터 과학 코드를 패키징 하기 위한 형식이다.

프로젝트 구성 요소에는 프로젝트 실행을 위한 API 및 명령줄 도구가 포함되어 있어 프로젝트를 workflow와 함께 연결할 수 있다.

 

YAML 형식의 텍스트 파일을 추가하여 프로젝트를 더 자세히 설명할 수 있다. 각 프로젝트는 여러 속성을 지정할 수 있다.

 

MLproject File

MLproject File은 MLflow 프로젝트의 구성 파일로, 프로젝트의 메타데이터와 실행 단계를 정의하는 데 사용된다.

MLproject 파일은 MLflow에서 프로젝트를 실행하고 관리하는 데 필요한 정보를 제공한다.

 

일반적으로 Project Root 디렉토리에 존재하며, YAML 형식으로 작성된다. (이름, 환경, 실행 단계 등을 명시한다.)

 

name : 프로젝트의 이름을 지정한다.

 

python(or conda or docker)_env : 환경 파일의 경로를 지정, 프로젝트 실행에 필요한 패키지와 종속성을 정의, mlflow는 이 환경을 사용하여 실행환경 구성

 

entry_points : 프로젝트의 실행 단계를 정의, 각 실행 단계는 이름과 명령을 가지며, 해당 명령은 프로젝트 실행 시 실행

name: DNN_iris_Classifier
conda_env: conda.yaml
# or
# python_env : python_env.yaml
# or
# docker_env :
#     image : mlflow-docker-example

entry_points:
    main:
        parameters:
            num_epochs: { type:int, default:20 }
        command: "python3 main.py --num_epochs {num_epochs}"

 

이제 학습과 tracking 하는 코드를 작성해보자.

# train.py
import mlflow
import tensorflow as tf
import pandas as pd
import numpy as np
from mlflow.tensorflow import MLflowCallback
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split


def tf_model(train_x, train_y, val_x, val_y, run, epoch=100, batch_size=8):
    inp_layer = tf.keras.Input(shape=(train_x.shape[1],))
    d1 = tf.keras.layers.Dense(16, activation='relu')(inp_layer)
    d2 = tf.keras.layers.Dense(8, activation='relu')(d1)
    out = tf.keras.layers.Dense(3, activation='softmax')(d2)

    model = tf.keras.Model(inputs=inp_layer, outputs=out,
                           name="Iris_Classifier_Model")
    model.compile(optimizer='adam',
                  loss='categorical_crossentropy', metrics=['acc'])
    history = model.fit(train_x, train_y, epochs=epoch,
                        batch_size=batch_size, callbacks=[MLflowCallback(run=run)])
    history = pd.DataFrame(history.history)
    metrics = {
        'score': {
            'model_score': np.float64(round(model.evaluate(val_x, val_y)[1], 2))
        },
        'params': {'epochs': epoch, 'batch_size': batch_size}
    }
    return model, history, metrics


if __name__ == "__main__":
    dataset = pd.read_csv("./dataset.csv")
    X = dataset.iloc[:, 0:-3]
    y = dataset.iloc[:, -3:]

    train_x, val_x, train_y, val_y = train_test_split(
        X, y, test_size=0.2, random_state=42)
    scaler = StandardScaler()
    sxtrain = scaler.fit_transform(train_x)
    sxval = scaler.transform(val_x)

    # tf_model(sxtrain, train_y, sxval, val_y, run=run)

# main.py
from datetime import datetime
import mlflow
import mlflow.keras
from train import tf_model
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split


def train(num_epochs: int):
    dataset = pd.read_csv("./dataset.csv")
    X = dataset.iloc[:, 0:-3]
    y = dataset.iloc[:, -3:]

    train_x, val_x, train_y, val_y = train_test_split(
        X, y, test_size=0.2, random_state=42)
    scaler = StandardScaler()
    sxtrain = scaler.fit_transform(train_x)
    sxval = scaler.transform(val_x)

    # 모델을 아티팩트로 기록
    # mlflow.create_experiment("Iris Classifier")
    # mlflow.set_experiment(name)
    with mlflow.start_run() as run:

        model, history, metrics = tf_model(
            sxtrain, train_y, sxval, val_y, run=run, epoch=num_epochs)

        mlflow.log_metrics(metrics['score'])
        mlflow.log_params(metrics['params'])
        tags = {
            "Version": "0.0.1",
            "Type": "Classifier",
            "Algorithm": "Keras, Dense"
        }
        mlflow.set_tags(tags)
        # 모델 저장
        mlflow.keras.log_model(
            model, "Iris_Classifier_DNN", input_example=sxval)
    mlflow.end_run()


if __name__ == "__main__":
    # 명령줄 인수에서 num_epochs 값을 가져옴
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--num_epochs", type=int, default=10,
                        help="Number of epochs for training")

    args = parser.parse_args()
    # 훈련 실행
    train(args.num_epochs)

# conda.yaml
channels:
    - conda-forge
dependencies:
    - python=3.9.0
    - pip<=23.3.2
    - pip:
          - mlflow==2.10.0
          - cloudpickle==3.0.0
          - numpy==1.22.0
          - tensorflow-macos==2.13.0
          - tensorflow==2.13.0
name: mlflow-env

 

이제 mlflow project를 생성한다.

git repository가 없으면 프로젝트 경로를 입력한다.(예시는 프로젝트 경로)

mlflow run -e main -P num_epochs=100 --experiment-name="Iris Classifier" --run-name="iris_DNN" /Users/pupba/Documents/study/mlflowstudy/DNN_iris

• --experiment-name : 실험 이름 설정 
• --run-name : run_name 설정

 

* Mac m1에서 MLflow 와 tensorflow 간의 버전 충돌을 있으므로 잘 확인해서 사용하기