经典机器学习 | 如何做到预流失与流失挽回？

# 正负样本均衡import randompos_vs_neg = 1.0pos_data_count = train_df.filter(train_df['ilabel'] == 1).count()neg_data_count = train_df.filter(train_df['ilabel'] == 0).count()gap = pos_data_count-neg_data_count*pos_vs_negprint('from ', pos_data_count, neg_data_count, 'to', pos_data_count, neg_data_count*pos_vs_neg, 'gap', gap)

if gap>0: # 正样本多，取样负样本   data_add = train_df.filter(train_df['ilabel'] == 0).sample(True, gap/neg_data_count, random.randint(1, 1000))   train_df = train_df.union(data_add)elif gap<0: # 负样本多，取样正样本   data_add = train_df.filter(train_df['ilabel'] == 1).sample(True, -gap/pos_data_count, random.randint(1, 1000))   train_df = train_df.union(data_add)    print(train_df.filter(train_df['ilabel'] == 1).count(), train_df.filter(train_df['ilabel'] == 0).count())print('balancing data finish')

aggregationDepth: suggested depth for treeAggregate (>= 2). (default: 2)elasticNetParam: the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty. (default: 0.0)family: The name of family which is a description of the label distribution to be used in the model. Supported options: auto, binomial, multinomial (default: auto)featuresCol: features column name. (default: features)fitIntercept: whether to fit an intercept term. (default: True)labelCol: label column name. (default: label, current: ilabel)maxIter: max number of iterations (>= 0). (default: 100, current: 1000)predictionCol: prediction column name. (default: prediction)probabilityCol: Column name for predicted class conditional probabilities. Note: Not all models output well-calibrated probability estimates! These probabilities should be treated as confidences, not precise probabilities. (default: probability, current: score)rawPredictionCol: raw prediction (a.k.a. confidence) column name. (default: rawPrediction)regParam: regularization parameter (>= 0). (default: 0.0, current: 0.03)standardization: whether to standardize the training features before fitting the model. (default: True)threshold: Threshold in binary classification prediction, in range [0, 1]. If threshold and thresholds are both set, they must match.e.g. if threshold is p, then thresholds must be equal to [1-p, p]. (default: 0.5, current: 0.6)thresholds: Thresholds in multi-class classification to adjust the probability of predicting each class. Array must have length equal to the number of classes, with values > 0, excepting that at most one value may be 0. The class with largest value p/t is predicted, where p is the original probability of that class and t is the class's threshold. (undefined)tol: the convergence tolerance for iterative algorithms (>= 0). (default: 1e-06)

'''模型评估'''# 模型评估## 训练数据的AUCprint("train auc is %.6f" %lr_model.summary.areaUnderROC)

## 方法一，使用pyspark.mllib.evaluation.BinaryClassificationMetrics来计算AUC# BinaryClassificationMetrics的参数是 RDD of (score, label) pairs，其中score表示预测为1的概率binary_metrics = BinaryClassificationMetrics(   result_df.selectExpr(['ilabel', 'score']) \   .rdd.map(lambda x: (float(x['score'][1]), float(x['ilabel']))))print("test auc is %.6f" %binary_metrics.areaUnderROC)

## 方法二，使用pyspark.ml.evaluation.BinaryClassificationEvaluator来计算AUCfrom pyspark.ml.evaluation import BinaryClassificationEvaluatorbinary_evaluator = BinaryClassificationEvaluator(rawPredictionCol='rawPrediction', labelCol='ilabel', metricName='areaUnderROC')print("test auc is %.6f" %binary_evaluator.evaluate(result_df))

## 方法三，使用BinaryLogisticRegressionSummary来计算AUC，注意，直接传入test data，不需要提前预测print("test auc is %.6f" %lr_model.evaluate(test_df).areaUnderROC)

tn = result_df.filter(result_df['ilabel'] == 0).filter(result_df['prediction'] == 0).count()fp = result_df.filter(result_df['ilabel'] == 0).filter(result_df['prediction'] == 1).count()fn = result_df.filter(result_df['ilabel'] == 1).filter(result_df['prediction'] == 0).count()tp = result_df.filter(result_df['ilabel'] == 1).filter(result_df['prediction'] == 1).count()print(tn, fp, fn, tp)precision = tp*1.0 / (fp+tp)recall = tp*1.0 / (fn+tp)f1 = 2*precision*recall / (precision+recall)print("test precision is %.6f" %(precision))print("test recall is %.6f" %(recall))print("test f1 is %.6f" %(f1))

from pyspark.sql import SparkSessionfrom pyspark.ml import Pipelinefrom pyspark.ml.classification import LogisticRegressionfrom pyspark.ml.feature import StandardScaler, OneHotEncoder, HashingTF, Tokenizer, VectorAssemblerfrom pyspark.ml.evaluation import BinaryClassificationEvaluator

# 训练数据# 预测数据# 预测结果表

# 需要做缺失值处理的特征missing_value_zero_features = ['lilastactdate' ...]missing_value_max_features = ['dilastactdate' ...]# z_scorez_score_features = ['lilastactdate' ...]# one_hot_dataone_hot_features = ['idayact14_loss' ...]print('args init finish')   def get_data():     # 获得训练数据   session = SparkSession \           .builder \           .appName("prelost-lr") \           .getOrCreate()   train_df = ...

   # 缺失值过滤   train_df = train_df.dropna(subset=['sopenid', 'iuin']) \       .na.fill(0, missing_value_zero_features) \       .na.fill(9999, missing_value_max_features)          # 测试集和训练集   (raw_train_data, raw_test_data) = train_df.randomSplit([0.7, 0.3], seed=43)   print('train data count is', raw_train_data.count())   print('test data count is', raw_test_data.count())   return raw_train_data, raw_test_data

def features_clear():   # Configure an ML pipeline, which consists of some stages   # 将所有的z_score_features特征合并起来一起做归一化   z_score_vector_assember = VectorAssembler(inputCols=z_score_features, outputCol="z_score_features")   z_score_standard_scaler = StandardScaler(inputCol=z_score_vector_assember.getOutputCol(), outputCol="z_score_features_scaled")

   # 组合所有特征到features字段里   features = [z_score_standard_scaler.getOutputCol()] + one_hot_features   features_vector_assember = VectorAssembler(inputCols=features, outputCol="features")      lr = LogisticRegression(maxIter=1000, regParam=0.03, threshold=0.6, labelCol="ilabel", probabilityCol="score")

   model_pipeline = Pipeline(stages=[z_score_vector_assember, z_score_standard_scaler,                                        features_vector_assember, lr])   print('features clear pipeline config finish')   return model_pipeline

def model_train(model_pipeline, raw_train_data):   model = model_pipeline.fit(raw_train_data)   print('model train finish')   return model   def model_evaluator(model, data):   result_data = model.transform(data)

   binary_evaluator = BinaryClassificationEvaluator(rawPredictionCol='rawPrediction', labelCol='ilabel', metricName='areaUnderROC')   print("auc is %.6f" %binary_evaluator.evaluate(result_data))

   tn = result_data.filter(result_data['ilabel'] == 0).filter(result_data['prediction'] == 0).count()   fp = result_data.filter(result_data['ilabel'] == 0).filter(result_data['prediction'] == 1).count()   fn = result_data.filter(result_data['ilabel'] == 1).filter(result_data['prediction'] == 0).count()   tp = result_data.filter(result_data['ilabel'] == 1).filter(result_data['prediction'] == 1).count()   print(tn, fp, fn, tp)   precision = tp*1.0 / (fp+tp)   recall = tp*1.0 / (fn+tp)   f1 = 2*precision*recall / (precision+recall)   print("precision is %.6f" %(precision))   print("recall is %.6f" %(recall))   print("f1 is %.6f" %(f1))

def model_predict():   result_data = lr_model.transform(test_data)   tdw_util = TDWUtil(tdw_user_name, tdw_user_pwd, tdw_db)   tdw_util.dropPartition(tdw_predict_result_tbl, tdw_predict_result_tbl_pri_partition, level=0)   tdw_util.createListPartition(tdw_predict_result_tbl, tdw_predict_result_tbl_pri_partition, tdw_predict_result_tbl_pri_date, 0)   tdw.saveToTable(result_data, tdw_predict_result_tbl, tdw_predict_result_tbl_pri_partition)   print('save data into tdw finish')   

raw_train_data, raw_test_data = get_data()model_pipeline = features_clear()model = model_train(model_pipeline, raw_train_data)print('------------------ train model ------------------')model_evaluator(model, raw_train_data)print('------------------ predict data ------------------')model_evaluator(model, raw_test_data)