import logging import multiprocessing as mp import os import time import traceback import uuid from multiprocessing import Process, Manager import mlflow import mlflow.sklearn import numpy as np import pandas as pd from pebble import ProcessPool, ProcessExpired from sklearn.decomposition import PCA from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.preprocessing import LabelEncoder from sklearn.preprocessing import MinMaxScaler from sklearn.utils import shuffle from .. import mlflow_client from ..tasks.plot_metrics import draw_feature_importance from ..tasks.trainers import scikit_dbscan, scikit_knn, scikit_medoids, hdbscan_algo from ..utils.aws_connectors import s3_to_pandas logging.getLogger().setLevel(logging.INFO) # https://stackoverflow.com/questions/38393269/fill-up-a-dictionary-in-parallel-with-multiprocessing manager = Manager() def run_random_forest(df, cols, target): """ This is only used to retrieve feature importance for clusters """ train_data = shuffle(df) x = train_data[cols] y = train_data[target] train_data = None # Free some RAM x = x.fillna(-1) y = y.fillna(-1) X_train, X_test, y_train, y_test = train_test_split(x, y, test_size=0.02, random_state=75844) # We want to overfit model here to discover feature importance. Don't care about train/test split hyperparams = {'n_estimators': 300, 'oob_score': True, 'max_features': None, 'n_jobs': -1} rf = RandomForestClassifier(**hyperparams) model = rf.fit(X_train, y_train, sample_weight=None) # y_train feature_importance = model.feature_importances_ feature_importance_plot_location = draw_feature_importance(feature_importance, cols) return feature_importance_plot_location, feature_importance def train_models(schema, collection_ids, bucket, file_key, modelling_config): feature_vector = s3_to_pandas(bucket, file_key, header='infer', file_format='csv') if 'fan_id' not in feature_vector.columns: return {'missing_required_attrs': 'fan_id', 'schema': schema, 'file_key': file_key } """ Drop fan_id and other _id columns""" # We drop every column that has "_id" in it, because it most likely is not useful for us and messes with quality fan_id = feature_vector['fan_id'] feature_vector = feature_vector[feature_vector.columns.drop(list(feature_vector.filter(regex='_id')))] # TODO! This belongs to data-processor # Drop time columns try: feature_vector = feature_vector[feature_vector.columns.drop(['51', '52', '53'])] except Exception as e: pass """ Drop fields which don't have at least 2 unique values. """ #logging.info( # f"Removing low info features. Column count before low info removal: {len(feature_vector.columns)} ") #feature_vector = remove_low_information_features(feature_vector, features=None) #logging.info(f"Column count after low info removal: {len(feature_vector.columns)} ") """ Try to fix data types. The more numeric encoding, the better for the models. """ feature_vector = feature_vector.apply(pd.to_numeric, errors='ignore') numeric_types = ['float64', 'float32', 'float16', 'int64', 'int32', 'int16', 'int8'] categorical_types = ['object', 'category'] cat_cols = feature_vector.select_dtypes(categorical_types).columns.tolist() # Calculate list of columns that have low cardinality (less than 50 unique values). cardinality = feature_vector[cat_cols].apply(pd.Series.nunique) low_cardinality = cardinality[cardinality < 50].index.tolist() del cardinality # Create correlation matrix. This only works with numeric fields. corr_threshold = modelling_config.get('drop_corr_threshold') if corr_threshold: corr_matrix = feature_vector.corr().abs() # Select upper triangle of correlation matrix upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(np.bool)) # Find index of feature columns with correlation greater than corr_threshold to_drop = [column for column in upper.columns if any(upper[column] > corr_threshold)] feature_vector = feature_vector.drop(feature_vector[to_drop], axis=1) logging.info( f"Dropped {len(to_drop)} highly correlated (>{corr_threshold}) columns to improve model quality: {to_drop}") del corr_matrix del upper # TODO! we should not be doing this here when our data types are fixed and handled properly numeric_cols = feature_vector.select_dtypes(numeric_types).columns.tolist() # We only include categorical columns with low cardinality and numeric columns columns = low_cardinality + numeric_cols logging.info(f"Training clusters with input dataset shape {feature_vector[columns].shape}") """ Transforming categorical fields to numeric to make it edible for models """ feature_vector_transformed = transform_categorical(feature_vector, cat_cols) del feature_vector # Add fan_id back to feature vector for mapping cluster back to fan_id feature_vector_transformed['fan_id'] = fan_id """ Do a final transformation, missing all empty values with -1. Would want to assume feature interpolation happens earlier, during preprocessing and feature engineering. Fill in some bigger value, because the distance between values plays more role in clustering, and -1 is quite close to any actual values than -1000 for that matter. """ feature_vector_transformed = feature_vector_transformed.fillna(-1000) ml_runs = train_clusters(schema, feature_vector_transformed, columns, collection_ids, modelling_config) del feature_vector_transformed best_model = get_best_model(ml_runs) return best_model def get_best_model(ml_runs): """ Calculates the best clustering algorithm, processing individual runs in parallel - https://fansifter.atlassian.net/browse/FSM-341 """ def process_mlrun(run_id, algo_name, algos): """ Processes individual mlflow run.""" algo_attributes = {} run_data = mlflow_client.get_run(run_id) artifact_uri = run_data.info.artifact_uri cluster_output_file = run_data.data.params['cluster_output_file'] cluster_output_location = artifact_uri + '/' + cluster_output_file cluster_output_location = cluster_output_location.replace('s3://fansifter-model-data/', '') # download S3 csv to dataframe bucket = 'fansifter-model-data' cluster_output = s3_to_pandas(bucket, cluster_output_location, header='infer')[ ['fan_id', 'cluster']] clusters_size = cluster_output.groupby(['cluster']).count() nr_of_elements = min(clusters_size['fan_id']) algo_attributes['min_cluster_size'] = nr_of_elements metrics_data = run_data.data.metrics algo_attributes['run_id'] = run_id algo_attributes['algo'] = algo_name algo_attributes['clusters'] = metrics_data.get('clusters') algo_attributes['clusters_size'] = clusters_size algo_attributes['silhouette_coefficient'] = metrics_data.get('silhouette_coefficient') algo_attributes['davies_bouldin_index'] = metrics_data.get('davies_bouldin_index') algo_attributes['calinski_harabasz_index'] = metrics_data.get('calinski_harabasz_index') algo_attributes['s_dbw'] = run_data.data.metrics.get('s_dbw') algo_attributes['validity_index'] = run_data.data.metrics.get('validity_index') if algo_attributes['clusters'] < 7: algos.append(algo_attributes) return algos = manager.list() procs = [] for algo_name, run_id in ml_runs.items(): proc = Process(target=process_mlrun, args=(run_id, algo_name, algos)) procs.append(proc) proc.start() for proc in procs: proc.join() algos_df = pd.DataFrame(list(algos)) # convert from multiprocessing.managers.ListProxy class to list algos_df = algos_df.round({'silhouette_coefficient': 6, 'davies_bouldin_index': 6, 'calinski_harabasz_index': 6, 's_dbw': 6, 'validity_index': 6 }) # TODO! validate if NaN is fine in the conditionals below (otherwise fillna) # substantiality: check if clusters have too few fan's size_treshold = 25 algos_df['no_treshold_violation'] = np.where(algos_df['min_cluster_size'] < size_treshold, 0, 1) # identifiability: how unique clusters are- sihlouette coefficient best_silhouette_y_n = algos_df.silhouette_coefficient == round(algos_df.silhouette_coefficient.max(), 6) algos_df['best_silhouette'] = np.where(best_silhouette_y_n, 1, 0) # identifiability: how unique clusters are- calinski_harabasz_index best_calinski_harabasz_y_n = algos_df.calinski_harabasz_index == round(algos_df.calinski_harabasz_index.max(), 6) algos_df['best_calinski_harabasz'] = np.where(best_calinski_harabasz_y_n, 1, 0) # identifiability: how unique clusters are- davies_bouldin_index best_davies_bouldin_y_n = algos_df.davies_bouldin_index == round(algos_df.davies_bouldin_index.min(), 6) algos_df['best_davies_bouldin'] = np.where(best_davies_bouldin_y_n, 1, 0) # validate clustering assignments on non-globular, arbitrarily shaped clusters best_validity_index_y_n = algos_df.validity_index == round(algos_df.validity_index.max(), 6) algos_df['best_validity_index'] = np.where(best_validity_index_y_n, 1, 0) columns = ['no_treshold_violation', 'best_silhouette', 'best_calinski_harabasz', 'best_davies_bouldin', 'best_validity_index'] algos_df['total_score'] = algos_df[columns].sum(axis=1) # in case of draw use silhouette index to find best algo # if still draw we select the one having largest cluster size # finally if still we sort by algo name and pick first cond = algos_df.total_score == algos_df.total_score.max() best_algorithm = algos_df[cond] if best_algorithm.shape[0] > 1: cond2 = best_algorithm.silhouette_coefficient == round(best_algorithm.silhouette_coefficient.max(), 6) best_algorithm = best_algorithm[cond2] if best_algorithm.shape[0] > 1: cond3 = best_algorithm.min_cluster_size == best_algorithm.min_cluster_size.max() best_algorithm = best_algorithm[cond3] best_algorithm.sort_values(by=['algo'], inplace=True) best_algorithm_name = best_algorithm['algo'].iloc[0] best_run_id = best_algorithm['run_id'].iloc[0] logging.info(f"Best model is {best_algorithm_name} with run_id {best_run_id}") #logging.info(best_algorithm) return {'algo': best_algorithm_name, 'run_id': best_run_id} def dimensionality_reduction(feature_vector, cols, pca_component): """ Apply dimensionality reduction algorithms. Currently only PCA """ # TODO! Add PCA evaluation metrics # TODO! pick best PCA nbr dynamically (see from evaluation chart quality what pct of info loss is acceptable) # TODO! add extra dimension in PCA as "size" in plots (and/or shape) reduced_data = PCA(n_components=pca_component, random_state=0, svd_solver='arpack').fit_transform(feature_vector[cols]) reduced_data = pd.DataFrame(reduced_data) # Our columns are changing column_names = {x: f"PCA_{x}" for x in reduced_data.columns} reduced_data = reduced_data.rename(columns=column_names) cols = reduced_data.columns.tolist() # Attach fan_id to the feature vector so we could map output back later """ feature_vector = reduced_data.join(feature_vector['fan_id']) return feature_vector, cols def run_algo_async(algorithm, collection_ids, modelling_config, feature_vector, original_feature_vector, original_cols, **kwargs): # TODO! do something better with the original_feature_vector, lose it altogether run_pca_component = modelling_config.get('pca_components', [-1]) # Try because we don't want to fail whole return when only one model fails but rest succeed for pca_component in run_pca_component: """ We make a copy of the dataframe, otherwise we keep operating on the same df. If we don't do this, then every iteration in loop will add another "cluster" column to dataframe """ start_time = time.time() algo_name = algorithm['name'] algo_func = algorithm['func'] algo_params = algorithm['algo_params'] algo_dtype = algorithm['dtype'] """ If we don't get pca_component, we just run models with all features. If we do get, we do dimensionality reduction using different pca_component settings (e.g. a list of PCA component nbr to train different models) """ """ If we get -1 from modelling config or if pca_component is >= nbr of columns, we don't do dimensionality reduction. PCA component must always strictly be len(cols) minus 1. """ if pca_component == -1 or pca_component >= len(original_cols): df, cols = original_feature_vector, original_cols else: df, cols = dimensionality_reduction(feature_vector, original_cols, pca_component) algo_name = algo_name + ' PCA ' + str(pca_component) training_set = df[cols] fan_ids = df['fan_id'].to_frame() # Just one col is ndarray, but we need dataframe in later steps """ Scale features to treat features of different scales equally in distance calculations. """ training_set = MinMaxScaler().fit_transform(training_set) if algo_dtype != np.float64: training_set = training_set.astype(algo_dtype) logging.info(f"Started running algorithm {algo_name} ...") with mlflow.start_run(run_name=algo_name, nested=True): try: mlflow.set_tag('model_name', algo_name) model, eval_metrics, hyperparams, cluster_output, graph_locations = algo_func(training_set, fan_ids, modelling_config, algo_params) if modelling_config.get('importance'): """ Add feature importance. We calculate feature importance by running random forest, and targeting the cluster label we get from clustering model """ # Our output index should match with input index df['cluster'] = cluster_output['cluster'] original_feature_vector['cluster'] = cluster_output['cluster'] target = 'cluster' feature_importance_plot_location, feature_importance = run_random_forest(original_feature_vector, original_cols, target) # TODO! This belongs to data-processor instead. We want to artifically always increase the importance of Purchase Monetary # get index of element 74 and replace the current feature_importance for that value with something that surely is significant feature_importance = feature_importance.tolist() try: i = original_cols.index('74') feature_importance[i] = 0.5 except Exception as e: pass mlflow.log_artifact(feature_importance_plot_location) mlflow.log_param('feature_importance', feature_importance) os.remove(feature_importance_plot_location) df_shape = df[cols].shape original_feature_vector_shape = original_feature_vector[original_cols].shape del df # Store cluster output file location to mlflow unique_id = uuid.uuid4() cluster_output_file = f'{algo_name}_output_{unique_id}.csv' mlflow.log_param('cluster_output_file', cluster_output_file) """ Store cluster output file csv """ current_path = os.path.dirname(__file__) file_location = f'{current_path}/resources/{cluster_output_file}' cluster_output.to_csv(file_location, sep=',', encoding='utf-8') mlflow.log_artifact(file_location) # delete file from local container, we don't need to retain it os.remove(file_location) del cluster_output mlflow_time = time.time() # ALTER TABLE params ALTER COLUMN value SET DATA TYPE varchar; mlflow.log_param('collection_ids', collection_ids) mlflow.log_param('data_params', { "data_shape": df_shape, "original_data_shape": original_feature_vector_shape, "features": cols, # make sure this is always a list "pca_component": pca_component }) mlflow.log_param('modelling_config', modelling_config) mlflow.log_param('original_features', original_cols) mlflow.log_param('model_params', { 'algorithm': algo_name, 'hyperparams': hyperparams }) mlflow.log_metrics(eval_metrics) # For speeding up, we remove graphs for location in graph_locations: # logging.info(f"storing {location}") mlflow.log_artifact(location) # delete file from local container, we don't need to retain it os.remove(location) # We don't need to store models themselves yet because we don't run inference #if modelling_config.get('save_model'): # mlflow.sklearn.log_model(model, schema) logging.info(f"Completed training for {algo_name} in {time.time() - start_time}s. Logging mlflow artifacts took {time.time() - mlflow_time}s.") # Return a mlflow run id so we can query modelling run artifacts elsewhere return dict(run_id=mlflow.active_run().info.run_id, algo_name=algo_name) except MemoryError as e: logging.error(f"Could not train {algo_name} with params {algo_params}: {e}") raise e except Exception as e: logging.error(f"Could not train {algo_name} with params {algo_params}: {e}") traceback.print_exc() return def get_algorithm_config(dataset_length): """ Declare a list of algorithms and their configurations. All of algorithms in async_algorithms list will be trained in parallel and results stored in Mlflow. """ # TODO! try mahalonobis metric ('metric_params': {'V': np.cov(X)}) # TODO! try rbf type for spectral algo affinity param # Notice that all algos have different params available. knn_params = {'init': 'k-means++', 'max_iter': 300} # dask algorithm does not support float16 input, # and sklearn's KMeans with float16 distance matrix for 100k rows occupies 20GB RAM - too much for our 30GB Fargate knn_algorithm = 'dask' if dataset_length < 25000 else ('sklearn' if dataset_length < 75000 else 'sklearn_minibatch') knn_dtype = np.float64 if knn_algorithm == 'dask' else np.float16 kmedoids_params = {'init': 'k-medoids++', 'max_iter': 0, 'method': 'pam', 'metric': 'euclidean'} dbscan_params = {'min_samples': 5, 'metric': 'euclidean'} hdbscan_params = {'metric': 'euclidean'} spectral_params = {'affinity': 'nearest_neighbors'} affinity_propa_params = {'affinity': 'euclidean', 'max_iter': 2000, 'convergence_iter': 15} """ Instead static modelling config, we should do hyper-param fine-tuning for each model """ async_algorithms = [ # The order of algorithms is important: the last ones will not be executed for large files # as the first ones will much likely perform better. # The order is based on ML Engine CloudWatch Logs between 02.2021 and 08.2021. # TODO! add t-SNE # TODO! add OPTICS model {'name': 'KMedoids 6 clusters', 'func': scikit_medoids.run_algo, 'dtype': np.float16, 'algo_params': {'n_clusters': 6, 'hyperparams': kmedoids_params}}, {'name': 'KMeans 6 clusters', 'func': scikit_knn.run_algo, 'dtype': knn_dtype, 'algo_params': {'n_clusters': 6, 'algorithm': knn_algorithm, 'hyperparams': knn_params}}, {'name': 'KMedoids 5 clusters', 'func': scikit_medoids.run_algo, 'dtype': np.float16, 'algo_params': {'n_clusters': 5, 'hyperparams': kmedoids_params}}, {'name': 'KMedoids 4 clusters', 'func': scikit_medoids.run_algo, 'dtype': np.float16, 'algo_params': {'n_clusters': 4, 'hyperparams': kmedoids_params}}, {'name': 'KMeans 5 clusters', 'func': scikit_knn.run_algo, 'dtype': knn_dtype, 'algo_params': {'n_clusters': 5, 'algorithm': knn_algorithm, 'hyperparams': knn_params}}, {'name': 'DBSCAN', 'func': scikit_dbscan.run_algo, 'dtype': np.float16, 'algo_params': {'hyperparams': dbscan_params}}, {'name': 'KMeans 4 clusters', 'func': scikit_knn.run_algo, 'dtype': knn_dtype, 'algo_params': {'n_clusters': 4, 'algorithm': knn_algorithm, 'hyperparams': knn_params}}, {'name': 'KMeans 3 clusters', 'func': scikit_knn.run_algo, 'dtype': knn_dtype, 'algo_params': {'n_clusters': 3, 'algorithm': knn_algorithm, 'hyperparams': knn_params}}, # TODO! DBSCAN kills processing with bigger data and requires similar performance analysis as was done for Kmedoids {'name': 'HDBSCAN 6 cluster approach', 'func': hdbscan_algo.run_algo, 'dtype': np.float16, 'algo_params': {'n_clusters': 6, 'hyperparams': hdbscan_params}}, {'name': 'HDBSCAN 4 cluster approach', 'func': hdbscan_algo.run_algo, 'dtype': np.float16, 'algo_params': {'n_clusters': 4, 'hyperparams': hdbscan_params}}, {'name': 'HDBSCAN 5 cluster approach', 'func': hdbscan_algo.run_algo, 'dtype': np.float16, 'algo_params': {'n_clusters': 5, 'hyperparams': hdbscan_params}}, # TODO! Spectral needs a speedup, do some analysis with dask_ml # {'name': 'Spectral 3 clusters', 'func': scikit_spectral.run_algo, 'algo_params': {'n_clusters': 3, 'hyperparams': spectral_params}}, # {'name': 'Spectral 4 clusters', 'func': scikit_spectral.run_algo, 'algo_params': {'n_clusters': 4, 'hyperparams': spectral_params}}, # {'name': 'Spectral 5 clusters', 'func': scikit_spectral.run_algo, 'algo_params': {'n_clusters': 5, 'hyperparams': spectral_params}}, # {'name': 'Spectral 5 clusters', 'func': scikit_spectral.run_algo, 'algo_params': {'n_clusters': 6, 'hyperparams': spectral_params}}, # TODO! For time-being we exclude AP - needs more testing as is quite resource hungry. # {'name': 'Affinity Propagation damping 0.5', 'func': scikit_affinityprop.run_algo, 'algo_params': {'damping': 0.5, 'hyperparams: affinity_propa_params}}, # {'name': 'Affinity Propagation damping 0.7', 'func': scikit_affinityprop.run_algo, 'algo_params': {'damping': 0.7, 'hyperparams: affinity_propa_params}}, # {'name': 'Affinity Propagation damping 0.9', 'func': scikit_affinityprop.run_algo, 'algo_params': {'damping': 0.9, 'hyperparams: affinity_propa_params}}, ] """The dataset_length - execution time dependency is quadratic, so we can't afford running all models when datasets have 50k+ rows. So, the above list of models was prioritized by the chance of performing the best. The number of models used for training depends on the length of the dataset in a quadratic manner. For example, all 11 models (as of 08.2021) are executed for files with less than 15k rows, 8 models are executed for files with 40k rows, and only 2 models are executed starting from 75k rows. """ # Empirical -(kx)^2 + b function based on current performance observations # TODO: optimize this function by observing real ML engine performance time number_of_algos_to_use = int(np.round(-(0.00004 * dataset_length)**2 + len(async_algorithms))) number_of_algos_to_use = max(number_of_algos_to_use, 2) # always train at least two models async_algorithms = async_algorithms[:number_of_algos_to_use] """Currently we can't perform clustering 50k+ rows with any algorithm except KMeans, since AWS Fargate has only 30 GB RAM. We use MiniBatchKMeans in this case to generate "good", but not "the best" result. """ if dataset_length > 75000: # TODO fix this temporary solution async_algorithms = [algo for algo in async_algorithms if algo['name'].startswith("KMeans")] logging.info(f"Using {len(async_algorithms)} models for training: {', '.join([algo['name'] for algo in async_algorithms])}") """ Depending on dataset row count, decide whether to execute high memory consuming models or not. We don't want to run those model individually even in synchronous manner, because they could still eat up memory. 42k rows requires more than 10gb of memory during run time. We are stuck with this until it is solved in library. - See this issue - https://github.com/scikit-learn-contrib/scikit-learn-extra/issues/23 """ # We're now pointing to big_ml_engine when row count is >10k and should have more than enough RAM #if len_features < 10000: # async_algorithms.extend(additional_algos_dicts) return async_algorithms def run_subprocess_training(params_per_run, max_workers=mp.cpu_count() - 1): modelling_runs = {} out_of_memory_run_params = [] # using pebble.ProcessPool, which (unlike multiprocessing.Pool or concurrent.futures.ProcessPoolExecutor) # is able to handle out-of-memory errors in workers well without any impact on the main process and other workers. with ProcessPool(max_workers=max_workers, max_tasks=1) as pool: futures = [] for params in params_per_run: futures.append(pool.schedule(run_algo_async, args=params)) for i, future in enumerate(futures): algorithm = params_per_run[i][0] try: result = future.result() if result is not None: modelling_runs[result['algo_name']] = result['run_id'] except (ProcessExpired, MemoryError) as error: logging.warning(f"Process '{algorithm['name']}' has probably run out of memory: {error}") # storing the params of unsuccessful workers to rerun them sequentially later out_of_memory_run_params.append(params_per_run[i]) except Exception as error: logging.warning("function raised %s" % error) logging.warning(error.traceback) # Python's traceback of remote process return modelling_runs, out_of_memory_run_params def train_clusters(schema, feature_vector, columns, collection_ids, modelling_config): # TODO! handle this duplication of data somehow more smarter way (this consumes memory) original_feature_vector = feature_vector.copy() original_cols = columns modelling_runs = {} try: """ Even though 'set_experiment' creates new when not exists, we explicitly create it to have control over artifact_location (set_experiment) doesn't expose this option """ mlflow.create_experiment(schema) mlflow.set_experiment(schema) logging.info("Created new experiment {}".format(schema)) except Exception as e: logging.info("Using existing Mlflow experiment {}".format(schema)) mlflow.set_experiment(schema) # Creates a new one if doesn't exist with same name algorithms = get_algorithm_config(dataset_length=len(feature_vector)) dots = '' if len(collection_ids) > 3: dots = '...' run_name = f"collections {','.join([str(x) for x in collection_ids])[0:8] + dots}" with mlflow.start_run(run_name=run_name, nested=True): # Mlflow is not threadsafe, we need to use multiprocessing logging.info(f"Training {len(algorithms)} models in parallel ({len(feature_vector)} rows)...") params_per_run = [(algorithm, collection_ids, modelling_config, feature_vector, original_feature_vector, original_cols) for algorithm in algorithms] # Start with running model training in parallel processes (3 for AWS Fargate with 4 vCPUs) modelling_runs, out_of_memory_run_params = run_subprocess_training(params_per_run, max_workers=mp.cpu_count() - 1) if len(out_of_memory_run_params) > 0: logging.warning(f"Model training was terminated by OOM error for {len(out_of_memory_run_params)} models.") logging.warning(f"Retrying sequential training ({len(feature_vector)} rows)...") # Retry unsuccessful runs in a sequential manner (but still in a separate worker process) seq_modelling_runs, _ = run_subprocess_training(out_of_memory_run_params, max_workers=1) modelling_runs.update(seq_modelling_runs) mlflow.end_run() logging.info(f"{schema} - Done modelling! {modelling_runs}") return modelling_runs def transform_categorical(df, categorical_columns): transform_cat_columns = categorical_columns.copy() # Need to convert datatypes of columns to strings if they are objects df[transform_cat_columns] = df[transform_cat_columns].astype(str) """ Do transformation on dataset outside of modelling pipeline """ if transform_cat_columns: df = MultiColumnLabelEncoder(columns=transform_cat_columns).fit_transform(df) return df class MultiColumnLabelEncoder: def __init__(self, columns=None): self.columns = columns # array of column names to encode def fit(self, X, y=None): return self # not relevant here def transform(self, X): """ Transforms columns of X specified in self.columns using LabelEncoder(). If no columns specified, transforms all columns in X. """ output = X.copy() if self.columns is not None: for col in self.columns: output[col] = LabelEncoder().fit_transform(output[col]) else: for colname, col in output.iteritems(): output[colname] = LabelEncoder().fit_transform(col) return output def fit_transform(self, X, y=None): return self.fit(X, y).transform(X) if __name__ == '__main__': ml_runs = { 'kmedoids':'eb292b657251499398e587413ce6149d', 'kmeans':'b63861bb24814683bce4aa4c5b243168', 'dbscan':'2f17bbc7397c4b4abbd66ded797fe58d', } get_best_model(ml_runs)