← posts

MLFlow로 모델 실험 트래킹

,

들어가기 전에

이 글을 보고 얻을 수 있는 정보는:

이 글에서 사용한 것은:

import seaborn as sns
penguins = sns.load_dataset('penguins')
penguins.head(3)
import matplotlib.pyplot as plt
import pandas as pd
import mlflow
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import LabelEncoder
from sklearn.metrics import roc_auc_score, plot_roc_curve, confusion_matrix
penguins = penguins.dropna(axis=0, how = 'any')
le = LabelEncoder()
penguins['species_encoded']= le.fit_transform(penguins['species'])
penguins['island_encoded']= le.fit_transform(penguins['island'])
penguins['sex_encoded']= le.fit_transform(penguins['sex'])

MLflow Tracking 기본

⛳ 나머지 정보들로 수컷인지 암컷인지 예측

features=['bill_length_mm', 'bill_depth_mm','flipper_length_mm', 'body_mass_g','island_encoded', 'species_encoded']
X = penguins[features]
y = penguins['sex_encoded']
x_train, x_test, y_train, y_test = train_test_split(X, y, test_size = 0.2)
sc = StandardScaler()
scaled_x_train = sc.fit_transform(x_train)
scaled_x_test = sc.transform(x_test)
def train(model, x_train, y_train):
  model = model.fit(x_train, y_train)
  train_acc = model.score(x_train, y_train)
  mlflow.log_metric("train_acc", train_acc)
  print(f"Train Accuracy: {train_acc:.3%}")
  
  
def evaluate(model, x_test, y_test):
  eval_acc = model.score(x_test, y_test)
  preds = model.predict(x_test)
  auc_score = roc_auc_score(y_test, preds, multi_class='ovr')
  mlflow.log_metric("eval_acc", eval_acc)
  mlflow.log_metric("auc_score", auc_score)
  print(f"Auc Score: {auc_score:.3%}")
  print(f"Eval Accuracy: {eval_acc:.3%}")
  roc_plot = plot_roc_curve(model, x_test, y_test, name = 'ROC Curve')
  plt.savefig("roc_plot.png")
  plt.show()
  plt.clf()
  
  conf_matrix = confusion_matrix(y_test, preds)
  ax = sns.heatmap(conf_matrix, annot=True, fmt='g')
  plt.title('Confusion Matrix')
  plt.savefig('conf_matrix.png')
  
  
  mlflow.log_artifact("roc_plot.png")
  mlflow.log_artifact("conf_matrix.png")
model = LogisticRegression()
mlflow.set_experiment("/Users/your_id/penguin_logistic_regression")
with mlflow.start_run():
  train(model, scaled_x_train, y_train)
  evaluate(model, scaled_x_test, y_test)
  mlflow.sklearn.log_model(model, "logistic_regression")
  print("Model run: ", mlflow.active_run().info.run_uuid)
mlflow.end_run()

저장된 걸 확인하고 모델 불러오기

loaded_model = mlflow.sklearn.load_model("dbfs:/databricks/mlflow-tracking/3683903273715082/18e2bcc166cf4e259fb45944295208a3/artifacts/logistic_regression")
loaded_model.score(scaled_x_test, y_test)

파라미터 서치

⛳ 나머지 정보들로 몸무게 예측, 근데 파라미터를 그리드 서치 해서 최적의 모델을 찾고 싶다.

from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import GridSearchCV

features=['bill_length_mm', 'bill_depth_mm','flipper_length_mm','sex_encoded','species_encoded','island_encoded']

X = penguins[features].values
y = penguins['body_mass_g'].values
mlflow.sklearn.autolog()

parameters = {'alpha':[0.05, 0.1, 0.5, 1, 2, 5]}
model = Ridge()
ridge_reg = GridSearchCV(model, parameters, scoring='neg_mean_squared_error',cv=5)

mlflow.set_experiment("/Users/your_id/penguin_ridge_regression")
with mlflow.start_run() as run:
  ridge_reg.fit(X, y)
  
mlflow.end_run()

Pyspark ver.

⛳ 나머지 정보들로 아델리인지 젠투인지 턱끈인지 예측, 근데 pyspark 로 하고 싶다.

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression

penguins_sdf = spark.createDataFrame(penguins)
features = ['bill_length_mm', 'bill_depth_mm','flipper_length_mm', 'body_mass_g','island_encoded', 'sex_encoded']

va = VectorAssembler(inputCols = features, outputCol = 'features')
data = penguins_sdf.select(F.col('species_encoded').alias('label'), *feature_list)
trainData, testData = data.randomSplit([0.8, 0.2])

trainData = va.transform(trainData)
testData = va.transform(testData)

def train(model, train_data):
  model = model.fit(train_data)
  trainingSummary = model.summary
  mlflow.log_metric("train_acc", trainingSummary.accuracy)
  print("Training Accuracy: ", trainingSummary.accuracy)
  return model

def evaluate(model, test_data):
  evaluationSummary = model.evaluate(test_data)
  mlflow.log_metric("eval_acc", evaluationSummary.accuracy)
  print("Evaluation Accuracy: ", evaluationSummary.accuracy)

lr = LogisticRegression(featuresCol = 'features', labelCol = 'label', maxIter = 10)
mlflow.set_experiment('/Users/your_id/penguin_logistic_regression_pyspark')
with mlflow.start_run():
  lr = train(lr, trainData)
  evaluate(lr, testData)
  mlflow.spark.log_model(lr, "logistic_regression_pyspark")
mlflow.end_run()

올?

같은 실험인데 그룹별로 나눠서 할래

⛳ 라벨을 쓰지 않고 클러스터링을 할 건데 파라미터를 실험해보고 싶다. 근데 데이터에 나오는 섬이 3개인데 섬마다 딱 맞는 파라미터가 다를 것 같아서 섬 별로 나눠서 하고 싶다.

from pyspark.sql.functions import pandas_udf, PandasUDFType
from sklearn.cluster import DBSCAN
from sklearn.metrics import silhouette_score, calinski_harabasz_score, davies_bouldin_score
from s_dbw import S_Dbw

return_schema = T.StructType([
  T.StructField('island', T.StringType()), 
  T.StructField('eps', T.FloatType()),
  T.StructField('silhouette', T.FloatType()),
  T.StructField('calinski-harabasz', T.FloatType()),
  T.StructField('davies-bouldin', T.FloatType()),
  T.StructField('s-dbw', T.FloatType()),
  T.StructField('n_clusters', T.IntegerType()),
  T.StructField('n_clustered_records', T.IntegerType()),
  T.StructField('model_path', T.StringType())
])
def clustering(pdf: pd.DataFrame) -> pd.DataFrame:
  island = pdf['island'].iloc[0]
  run_id = pdf['run_id'].iloc[0] 
  feature_list = ['bill_length_mm', 'bill_depth_mm','flipper_length_mm', 'body_mass_g']
  X = pdf[feature_list].values
  scaler = StandardScaler()
  X_scaled = scaler.fit_transform(X)
	
  n_clusters_list = []
  n_clustered_records_list = []
  slh_list = []
  island_list = []
  model_path_list = []
  
  for eps_val in eps_list:
	island_list.append(island)
	
	with mlflow.start_run(run_id=run_id):
	  with mlflow.start_run(run_name=str(island)+'-eps'+str(eps_val), nested=True) as run:
		mlflow.log_param("eps", eps_val)
		model = DBSCAN(eps=eps_val, min_samples=3)
		model.fit(X_scaled)
		labels = model.labels_
		num_labels = len(set(labels[labels>=0]))
	  
		if num_labels -1 >= 1 :
		  clustered_label = labels[labels>=0]
		  clustered_data = X[np.where(labels>=0)[0]]
		  clustered_data_cnt = len(clustered_data)
		  
		  #scores
		  slh = silhouette_score(clustered_data, clustered_label)
		  ch = calinski_harabasz_score(clustered_data, clustered_label)
		  db = davies_bouldin_score(clustered_data, clustered_label)
		  s_dbw = S_Dbw(X_scaled, labels, alg_noise='filter')
		  
		  
		else:
		  slh = -1
		  ch = -1
		  db = -1
		  s_dbw = -1
		  clustered_data_cnt = 0

		mlflow.sklearn.log_model(model, str(island)+str('-')+str(eps_val))
		mlflow.log_metric("silhouette", slh)
		mlflow.log_metric("calinski-harabasz", ch)
		mlflow.log_metric("davies-bouldin", db)
		mlflow.log_metric("s-dbw", s_dbw)
		mlflow.log_metric("n_clusters", num_labels)
		mlflow.log_metric("n_clustered_records", clustered_data_cnt)
		
		artifact_uri = f"runs:/{run.info.run_id}/{island}" + str(eps_val)
		
		model_path_list.append(artifact_uri)
		n_clustered_records_list.append(clustered_data_cnt)
		n_clusters_list.append(num_labels)
		
		slh_list.append(slh)
		ch_list.append(ch)
		db_list.append(db)
		sdbw_list.append(s_dbw)
		
  
  returnDF = pd.DataFrame({'island':island_list, 'eps':eps_list, 'silhouette': slh_list, 'calinski-harabasz':ch_list, 'davies-bouldin': db_list, 's-dbw':sdbw_list, 'n_clusters':n_clusters_list, 'n_clustered_records':n_clusters_list, 'model_path':model_path_list})

  return returnDF
with mlflow.start_run(run_name="섬 별 클러스터링") as run:
  run_id = run.info.run_uuid
  eps_list = [0.1, 0.2, 0.3, 0.5, 0.7, 0.9, 1.1, 1.3, 1.5]
  modelDF = (penguins_sdf
			 .withColumn("run_id", F.lit(run_id))
			 .groupby("island").applyInPandas(clustering, schema = return_schema)
			)

display(modelDF)
display(modelDF.filter(F.col('island')=='Dream').orderBy('eps'))

참고한 책과 글

Comments