fixing mlflow code and add functionalities

Author: jrzaurin

initialize.py

@@ -2,12 +2,12 @@
 import pandas as pd
 import lightgbm as lgb
 import pickle
-import pdb
 import warnings
+import argparse
 import os
+import pdb
 
 from pathlib import Path
-from train.train_hyperopt import LGBOptimizer
 from utils.preprocess_data import build_train
 
 
@@ -50,24 +50,32 @@ def download_data():
 	df_test.to_csv(test_path)
 
 
-
 def create_data_processor():
 	print("creating preprocessor...")
 	dataprocessor = build_train(TRAIN_PATH/'train.csv', DATAPROCESSORS_PATH)
 
 
-
-def create_model():
+def create_model(hyper):
 	print("creating model...")
 	init_dataprocessor = 'dataprocessor_0_.p'
 	dtrain = pickle.load(open(DATAPROCESSORS_PATH/init_dataprocessor, 'rb'))
+	if hyper == "hyperopt":
+		# from train.train_hyperopt import LGBOptimizer
+		from train.train_hyperopt_mlflow import LGBOptimizer
+	elif hyper == "hyperparameterhunter":
+		# from train.train_hyperparameterhunter import LGBOptimizer
+		from train.train_hyperparameterhunter_mlfow import LGBOptimizer
 	LGBOpt = LGBOptimizer(dtrain, MODELS_PATH)
-	LGBOpt.optimize(maxevals=10)
-	# LGBOpt = LGBOptimizer(dtrain, str(MODELS_PATH))
-	# LGBOpt.optimize('f1_score', StratifiedKFold, n_splits=3, maxevals=10)
+	LGBOpt.optimize(maxevals=2)
+
 
 if __name__ == '__main__':
-	# create_folders()
-	# download_data()
+
+	parser = argparse.ArgumentParser()
+
+	parser.add_argument("--hyper", type=str, default="hyperopt")
+	args = parser.parse_args()
+	create_folders()
+	download_data()
 	create_data_processor()
-	# create_model()
+	create_model(args.hyper)

predictor.py

@@ -13,7 +13,7 @@
 MODELS_PATH = PATH/'models'
 DATAPROCESSORS_PATH = PATH/'dataprocessors'
 MESSAGES_PATH = PATH/'messages'
-RETRAIN_EVERY = 20
+RETRAIN_EVERY = 25
 EXTRA_MODELS_TO_KEEP = 1
 
 column_order = pickle.load(open(DATAPROCESSORS_PATH/'column_order.p', 'rb'))

train/train_hyperopt.py

@@ -88,7 +88,7 @@ def __init__(self, trainDataset, out_dir):
 			categorical_feature = self.categorical_columns,
 			free_raw_data=False)
 
-	def optimize(self, maxevals=50, model_id=0):
+	def optimize(self, maxevals=200, model_id=0):
 
 		param_space = self.hyperparameter_space()
 		objective = self.get_objective(self.lgtrain)

train/train_hyperopt_mlflow.py

@@ -91,7 +91,7 @@ def __init__(self, trainDataset, out_dir):
 			categorical_feature = self.categorical_columns,
 			free_raw_data=False)
 
-	def optimize(self, maxevals=50, model_id=0):
+	def optimize(self, maxevals=200, model_id=0, reuse_experiment=False):
 
 		param_space = self.hyperparameter_space()
 		objective = self.get_objective(self.lgtrain)
@@ -106,22 +106,17 @@ def optimize(self, maxevals=50, model_id=0):
 		best['num_leaves'] = int(best['num_leaves'])
 		best['verbose'] = -1
 
-		# The next few lines are the only ones related to mlflow. One
-		# "annoying" behaviour of mlflow is that when you instantiate a client
-		# it creates the 'mlruns' dir by default as well as the first
-		# experiment and there does not seem to be a way I can change this
-		# behaviour without changing the source code. The solution is the
-		# following hack:
+		# The next few lines are the only ones related to mlflow.
 		if not Path('mlruns').exists():
-			client = MlflowClient()
-		else:
-			client = MlflowClient()
-			n_experiments = len(client.list_experiments())
-			experiment_name = 'experiment_' + str(n_experiments)
-			client.create_experiment(name=experiment_name)
-		experiments = client.list_experiments()
-		with mlflow.start_run(experiment_id=experiments[-1].experiment_id) as run:
-		# with mlflow.start_run() as run:
+            # here set the tracking_uri. If None then http://localhost:5000
+		    client = MlflowClient()
+		    n_experiments=0
+		elif not reuse_experiment:
+		    client = MlflowClient()
+		    n_experiments = len(client.list_experiments())
+		    experiment_name = 'experiment_' + str(n_experiments)
+		    client.create_experiment(name=experiment_name)
+		with mlflow.start_run(experiment_id=n_experiments):
 			model = lgb.LGBMClassifier(**best)
 			model.fit(self.lgtrain.data,
 				self.lgtrain.label,

train/train_hyperparameterhunter.py

@@ -36,26 +36,29 @@ def __init__(self, trainDataset, out_dir):
             Path to the output directory
         """
 
-        self.PATH = out_dir
+        self.PATH = str(out_dir)
         self.data = trainDataset.data
         self.data['target'] = trainDataset.target
         self.colnames = trainDataset.colnames
         self.categorical_columns = trainDataset.categorical_columns + trainDataset.crossed_columns
 
-    def optimize(self, metrics, cv_type, n_splits, maxevals=200, do_predict_proba=None):
+    def optimize(self, metrics='f1_score', n_splits=3, cv_type=StratifiedKFold,
+        maxevals=200, do_predict_proba=None, model_id=0):
 
         params = self.hyperparameter_space()
         extra_params = self.extra_setup()
 
         env = Environment(
             train_dataset=self.data,
-            results_path=self.PATH,
+            results_path='HyperparameterHunterAssets',
+            # results_path=self.PATH,
             metrics=[metrics],
             do_predict_proba = do_predict_proba,
             cv_type=cv_type,
             cv_params=dict(n_splits=n_splits),
         )
 
+        # optimizer = opt.GradientBoostedRegressionTreeOptimization(iterations=maxevals)
         optimizer = opt.BayesianOptimization(iterations=maxevals)
         optimizer.set_experiment_guidelines(
             model_initializer=lgb.LGBMClassifier,
@@ -66,8 +69,7 @@ def optimize(self, metrics, cv_type, n_splits, maxevals=200, do_predict_proba=No
         # there are a few fixes on its way and the next few lines will soon be
         # one. At the moment, to access to the best parameters one has to read
         # from disc and access them
-        best_experiment = self.PATH+\
-            '/HyperparameterHunterAssets/Experiments/Descriptions/'+\
+        best_experiment = 'HyperparameterHunterAssets/Experiments/Descriptions/'+\
             optimizer.best_experiment+'.json'
         with open(best_experiment) as best:
             best = json.loads(best.read())['hyperparameters']['model_init_params']
@@ -77,10 +79,11 @@ def optimize(self, metrics, cv_type, n_splits, maxevals=200, do_predict_proba=No
             feature_name=self.colnames,
             categorical_feature=self.categorical_columns
             )
-        pickle.dump(model, open(self.PATH+'/HHmodel.p', 'wb'))
-        pickle.dump(optimizer, open(self.PATH+'/HHoptimizer.p', 'wb'))
+        model_fname = 'model_{}_.p'.format(model_id)
+        best_experiment_fname = 'best_experiment_{}_.p'.format(model_id)
+        pickle.dump(model, open('/'.join([self.PATH,model_fname]), 'wb'))
+        pickle.dump(optimizer, open('/'.join([self.PATH,best_experiment_fname]), 'wb'))
 
-        return
 
     def hyperparameter_space(self, param_space=None):
 
@@ -114,10 +117,3 @@ def extra_setup(self, extra_setup=None):
             return extra_setup
         else:
             return extra_params
-
-# if __name__ == '__main__':
-
-#     MD_PATH = Path('data/models/')
-#     dtrain = pickle.load(open(MD_PATH/'preprocessor_0_.p', 'rb'))
-#     HHOpt = HHOptimizer(dtrain, str(MD_PATH))
-#     optimizer = HHOpt.optimize('f1_score', StratifiedKFold, n_splits=3, maxevals=3)

train/train_hyperparameterhunter_mlfow.py

@@ -40,13 +40,14 @@ def __init__(self, trainDataset, out_dir):
             Path to the output directory
         """
 
-        self.PATH = out_dir
+        self.PATH = str(out_dir)
         self.data = trainDataset.data
         self.data['target'] = trainDataset.target
         self.colnames = trainDataset.colnames
         self.categorical_columns = trainDataset.categorical_columns + trainDataset.crossed_columns
 
-    def optimize(self, metrics, cv_type, n_splits, maxevals=200, do_predict_proba=None):
+    def optimize(self, metrics='f1_score', n_splits=3, cv_type=StratifiedKFold,
+        maxevals=200, do_predict_proba=None, model_id=0, reuse_experiment=False):
 
         params = self.hyperparameter_space()
         extra_params = self.extra_setup()
@@ -78,20 +79,17 @@ def optimize(self, metrics, cv_type, n_splits, maxevals=200, do_predict_proba=No
         with open(best_experiment) as best:
             best = json.loads(best.read())['hyperparameters']['model_init_params']
 
-        # The next few lines are the only ones related to mlflow. One
-        # "annoying" behaviour of mlflow is that when you instantiate a client
-        # it creates the 'mlruns' dir by default as well as the first
-        # experiment and there does not seem to be a way I can change this
-        # behaviour without changing the source code. The solution is the
-        # following hack:
+        # The next few lines are the only ones related to mlflow
         if not Path('mlruns').exists():
+            # here set the tracking_uri. If None then http://localhost:5000
             client = MlflowClient()
-        else:
+            n_experiments=0
+        elif not reuse_experiment:
             client = MlflowClient()
             n_experiments = len(client.list_experiments())
-            client.create_experiment(name=str(n_experiments))
-        experiments = client.list_experiments()
-        with mlflow.start_run(experiment_id=experiments[-1].experiment_id) as run:
+            experiment_name = 'experiment_' + str(n_experiments)
+            client.create_experiment(name=experiment_name)
+        with mlflow.start_run(experiment_id=n_experiments):
             model = lgb.LGBMClassifier(**best)
             X, y = self.data.drop('target',axis=1), self.data.target
             model.fit(X,y,
@@ -103,10 +101,11 @@ def optimize(self, metrics, cv_type, n_splits, maxevals=200, do_predict_proba=No
             mlflow.log_metric('f1_score', -optimizer.optimizer_result.fun)
             mlflow.sklearn.log_model(model, "model")
 
-        pickle.dump(model, open(self.PATH+'/HHmodel.p', 'wb'))
-        pickle.dump(optimizer, open(self.PATH+'/HHoptimizer.p', 'wb'))
+        model_fname = 'model_{}_.p'.format(model_id)
+        best_experiment_fname = 'best_experiment_{}_.p'.format(model_id)
+        pickle.dump(model, open('/'.join([self.PATH,model_fname]), 'wb'))
+        pickle.dump(optimizer, open('/'.join([self.PATH,best_experiment_fname]), 'wb'))
 
-        return
 
     def hyperparameter_space(self, param_space=None):
 
@@ -139,10 +138,3 @@ def extra_setup(self, extra_setup=None):
             return extra_setup
         else:
             return extra_params
-
-# if __name__ == '__main__':
-
-#     MD_PATH = Path('data/models/')
-#     dtrain = pickle.load(open(MD_PATH/'preprocessor_0_.p', 'rb'))
-#     HHOpt = HHOptimizer(dtrain, str(MD_PATH))
-#     optimizer = HHOpt.optimize('f1_score', StratifiedKFold, n_splits=3, maxevals=3)

trainer.py

@@ -2,15 +2,13 @@
 import json
 import pandas as pd
 import pickle
+import argparse
 
 from pathlib import Path
 from kafka import KafkaConsumer
 
 from utils.messages_utils import publish_traininig_completed
 from utils.preprocess_data import build_train
-from train.train_hyperopt import LGBOptimizer
-# uncomment line below to use hyperparameterhunter
-# from train.train_hyperparameterhunter import LGBOptimizer
 
 
 KAFKA_HOST = 'localhost:9092'
@@ -22,15 +20,21 @@
 MESSAGES_PATH = PATH/'messages'
 
 
-def train(model_id, messages):
+def train(model_id, messages, hyper):
 	print("RETRAINING STARTED (model id: {})".format(model_id))
 	dtrain = build_train(TRAIN_DATA, DATAPROCESSORS_PATH, model_id, messages)
+	if hyper == "hyperopt":
+		# from train.train_hyperopt import LGBOptimizer
+		from train.train_hyperopt_mlflow import LGBOptimizer
+	elif hyper == "hyperparameterhunter":
+		# from train.train_hyperparameterhunter import LGBOptimizer
+		from train.train_hyperparameterhunter_mlfow import LGBOptimizer
 	LGBOpt = LGBOptimizer(dtrain, MODELS_PATH)
-	LGBOpt.optimize(maxevals=10, model_id=model_id)
+	LGBOpt.optimize(maxevals=2, model_id=model_id)
 	print("RETRAINING COMPLETED (model id: {})".format(model_id))
 
 
-def start():
+def start(hyper):
 	consumer = KafkaConsumer(RETRAIN_TOPIC, bootstrap_servers=KAFKA_HOST)
 
 	for msg in consumer:
@@ -41,9 +45,14 @@ def start():
 			message_fname = 'messages_{}_.txt'.format(batch_id)
 			messages = MESSAGES_PATH/message_fname
 
-			train(model_id, messages)
+			train(model_id, messages, hyper)
 			publish_traininig_completed(model_id)
 
 
 if __name__ == '__main__':
-	start()
+	parser = argparse.ArgumentParser()
+
+	parser.add_argument("--hyper", type=str, default="hyperopt")
+	args = parser.parse_args()
+
+	start(args.hyper)