refactor: 增加枚举类，进一步减少代码重复

java_compile/src/main/java/com/example/services/ModelService.java

@@ -16,20 +16,16 @@ import com.example.model.regression.RFRegression;
 import com.example.services.pojo.DLGeneralModelPojo;
 import com.example.services.pojo.DLModelPojo;
 import com.example.services.pojo.ModelPojo;
-import com.example.util.Constant;
 import com.example.util.Json2Object;
+import com.example.util.TrainParamLoader;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.spark.ml.PipelineModel;
-import org.apache.spark.ml.classification.LogisticRegressionModel;
-import org.apache.spark.ml.clustering.KMeansModel;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import scala.Tuple2;
 
-import javax.persistence.Tuple;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -267,108 +263,16 @@ public class ModelService {
 
 
         String modelDetailName = modelType.getModelDetailName();
-        HashMap<String,String> modelArguments = new Json2Object().Json2HashMap(trainingModel.getArguments());
+        Map<String,String> modelArguments = new Json2Object().Json2HashMap(trainingModel.getArguments());
         boolean rsl = false;
+        Map<String,Object> params = TrainParamLoader.getParams(modelArguments,modelDetailName);
         try {
-            switch (modelDetailName) {
-                case "LogisticRegression":
-            /*
-            * maxIter: Int,
-                 regParam: Float,
-                 elasticNetParam: Float
-            * */
-
-                    Map<String,Object> params = new HashMap<>();
-                    params.put(Constant.TRAIN_PARAM_MAX_CATEGORY,4);
-                    params.put(Constant.TRAIN_PARAM_MAX_ITER,Integer.parseInt(modelArguments.get("MaxIter")));
-                    params.put(Constant.TRAIN_PARAM_REG_PARAM,Float.parseFloat(modelArguments.get("RegParam")));
-                    params.put(Constant.TRAIN_PARAM_ELASTICNET_PARAM,Float.parseFloat(modelArguments.get("ElasticNetParam")));
-                    Tuple2<PipelineModel,Map<String,String>> trainResult =lrClassification.training(libsvm,params);
-
-                    PipelineModel lrModel = trainResult._1;
-                    Map<String,String> lrRst = trainResult._2;
-
-                    log.info("\n\nupdate result of model\n\n");
-                    rsl = updateResOfModel(modelId, lrRst, lrModel);
-                    break;
-                case "DecisionTree":
-            /*
-            * maxIter: Int,
-                 regParam: Float,
-                 elasticNetParam: Float
-            * */
-                    PipelineModel dtModel = dtClassification.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("Category")),
-                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
-                    HashMap<String,String> dtResult = dtClassification.dtResult(dtModel, modelId);
-
-                    rsl = updateResOfModel(modelId, dtResult, dtModel);
-                    break;
-                case "RandomForest":
-                    /*
-                * maxIter: Int,
-                 regParam: Float,
-                 elasticNetParam: Float
-            * */
-
-                    PipelineModel rfModel = rfClassification.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("Category")),
-                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
-                    HashMap<String,String> rfResult = rfClassification.dtResult(rfModel, modelId);
-
-                    rsl = updateResOfModel(modelId,rfResult,rfModel);
-                    break;
-                case "GBDT":
-                    /*
-                * maxIter: Int,
-                 regParam: Float,
-                 elasticNetParam: Float
-            * */
-
-                    PipelineModel gbtModel = gdbtClassification.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("Category")),
-                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
-                    HashMap<String,String> gbtResult = gdbtClassification.dtResult(gbtModel, modelId);
-
-                    rsl = updateResOfModel(modelId,gbtResult,gbtModel);
-                    break;
-
-                case "K-Means":
-                    KMeansModel kmeansModel = kmeansCluster.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("NumOfCluster")),
-                            Integer.parseInt(modelArguments.get("Seed")));
-                    HashMap<String,String> kmeansResult = kmeansCluster.dtResult(kmeansModel,modelId);
-                    rsl = updateResOfModel(modelId,kmeansResult,kmeansModel);
-                    break;
-
-                case "MultilayerPerceptronClassifier":
-                    PipelineModel mlpc = mulPerClassification.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("Category")),
-                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")), featureNum);
-                    HashMap<String,String> mlpcResult = mulPerClassification.dtResult(mlpc, modelId);
-                    rsl = updateResOfModel(modelId,mlpcResult,mlpc);
-                    break;
-
-                case "NaiveBayes":
-                    PipelineModel nby = naiveBayesClassification.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("Category")),
-                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
-                    HashMap<String,String> nbyResult = naiveBayesClassification.dtResult(nby, modelId);
-                    rsl = updateResOfModel(modelId,nbyResult,nby);
-                    break;
-
-                case "RandomForestRegression":
-                    PipelineModel rfReg = rfRegression.dtTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("Category")),
-                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy"))
-                            );
-                    HashMap<String,String> rfRegResult = rfRegression.dtResult(rfReg,modelId);
-
-                    rsl = updateResOfModel(modelId,rfRegResult,rfReg);
-                    break;
-                default:
-                    rsl = false;
-            }
+            Tuple2<PipelineModel,Map<String,String>> trainResult = trainClassifier(libsvm,params,modelDetailName);
+            PipelineModel lrModel = trainResult._1;
+            Map<String,String> lrRst = trainResult._2;
+
+            log.info("\n\nupdate result of model\n\n");
+            rsl = updateResOfModel(modelId, lrRst, lrModel);
         } catch (Exception e) {
             e.printStackTrace();
             changeStatusOfTrained(modelId, false);
@@ -400,4 +304,39 @@ public class ModelService {
     }
 
 
+    /**
+     * 统一的算法访问接口，仅仅包括分类算法
+     * @author   fanyanpeng
+     * @date 2023/3/22 21:15
+     * @param libsvmUrl
+     * @param params
+     * @param modelType
+     * @return scala.Tuple2<org.apache.spark.ml.PipelineModel,java.util.Map<java.lang.String,java.lang.String>>
+     */
+    private Tuple2<PipelineModel,Map<String,String>> trainClassifier(String libsvmUrl, Map<String,Object> params, String modelType) throws Exception {
+        Tuple2<PipelineModel,Map<String,String>> result=null;
+        DefaultClassification defaultClassification=null;
+        switch (modelType){
+            case "LogisticRegression":
+                defaultClassification = lrClassification;break;
+            case "DecisionTree":
+                defaultClassification = dtClassification;break;
+            case "RandomForest":
+                defaultClassification = rfClassification;break;
+            case "GBDT":
+                defaultClassification = gdbtClassification;break;
+            case "MultilayerPerceptronClassifier":
+                defaultClassification = mulPerClassification;break;
+            case "NaiveBayes":
+                defaultClassification = naiveBayesClassification;break;
+            default:
+                break;
+
+        }
+        return defaultClassification.training(libsvmUrl,params);
+    }
+
+
+
+
 }

java_compile/src/main/java/com/example/util/Constant.java

@@ -11,4 +11,6 @@ public class Constant {
     public static final String TRAIN_PARAM_ELASTICNET_PARAM =  "ElasticNetParam";
     public static final String TRAIN_PARAM_REG_PARAM = "RegParam";
     public static final String TRAIN_PARAM_MAX_ITER = "MaxIter";
+
+    public static final String TRAIN_PARAM_FEATURE_NUM = "FeatureNUm";
 }

java_compile/src/main/java/com/example/util/TrainParamEnum.java

@@ -0,0 +1,63 @@
+package com.example.util;
+
+/**
+ * @author fanyanpeng
+ * @date 2023/3/22 19:32
+ */
+
+public enum TrainParamEnum {
+
+
+    FEATURE_NUM("feature_num","Int","特征数目，没有默认值",-1),
+    MAX_CATEGORIES("Category","Int","特征取值高于该数值的会被视为连续，不会改变；小于等于该数值，被视为类型数据",5),
+    TRAIN_DATA_SET_OCCUPY("TrainingDataSetOccupy","Double","选定为训练集的比例，取值应该在(0.0,1.0)",0.7),
+
+    //LogisticRegression, from https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/ml/LogisticRegressionExample.scala
+    REG_PARAM("RegParam","Double","regularization parameter",0.0),
+    MAXI_ITER("MaxIter","Int","maximum number of iterations",100),
+    ELASTIC_NET_PARAM("ElasticNetParam","Double","ElasticNet mixing parameter",0.0),
+
+
+    //DecisionTree,from https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/ml/DecisionTreeExample.scala
+    MAX_DEPTH("maxDepth","Int","max depth of the tree",5),
+    MAX_BINS("maxBins","Int","max number of bins",32),
+    MIN_INSTANCES_PER_NODE("minInstancesPerNode","Int","min number of instances required at child nodes to create the parent split",1),
+    MIN_INFO_GAIN("minInfoGain","Double","min info gain required to create a split",0.0),
+    ;
+
+
+    private final String paramName;
+    private final String paramType;
+    private final String description;
+
+    private Object defaultValue;
+
+    TrainParamEnum(String paramName, String paramType, String description) {
+        this.paramName = paramName;
+        this.paramType = paramType;
+        this.description = description;
+    }
+
+    TrainParamEnum(String paramName, String paramType, String description, Object defaultValue) {
+        this.paramName = paramName;
+        this.paramType = paramType;
+        this.description = description;
+        this.defaultValue = defaultValue;
+    }
+
+    public String getParamName() {
+        return paramName;
+    }
+
+    public String getParamType() {
+        return paramType;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public Object getDefaultValue() {
+        return defaultValue;
+    }
+}

java_compile/src/main/java/com/example/util/TrainParamLoader.java

@@ -0,0 +1,58 @@
+package com.example.util;
+
+
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author fanyanpeng
+ * @date 2023/3/22 20:43
+ */
+public class TrainParamLoader {
+
+    static private Map<String,TrainParamEnum[]> modelTrainParamMap = new HashMap<String,TrainParamEnum[]>();
+
+    static {
+        modelTrainParamMap.put("LogisticRegression",new TrainParamEnum[]{
+                TrainParamEnum.REG_PARAM,
+                TrainParamEnum.MAX_CATEGORIES,
+                TrainParamEnum.ELASTIC_NET_PARAM,
+                TrainParamEnum.MAXI_ITER,
+                TrainParamEnum.TRAIN_DATA_SET_OCCUPY});
+    }
+
+    public static Map<String,Object> getParams(Map<String,String> modelArguments,String modelType){
+        Map<String,Object> params = new HashMap<>();
+        TrainParamEnum[] toFill = modelTrainParamMap.get(modelType);
+        for(TrainParamEnum trainParamEnum : toFill){
+            String paramName = trainParamEnum.getParamName();
+            String paramType = trainParamEnum.getParamType();
+            String stringValue = modelArguments.get(paramName);
+            Object value = null;
+
+            //取出值为空
+            if(stringValue == null){
+                value = trainParamEnum.getDefaultValue();
+            }
+            else {  //取出值不为空时，才可以进行转换
+                switch (paramType){
+                    case "Int":
+                        value = Integer.valueOf(stringValue);
+                        break;
+                    case "Double":
+                        value = Double.valueOf(stringValue);
+                        break;
+                    default:
+                        throw new  UnsupportedOperationException("参数不支持Int、Double以外的类型");
+                }
+            }
+
+            params.put(paramName,value);
+
+        }
+        return params;
+    }
+
+
+}

java_compile/src/main/scala/com/example/model/classification/DTClassification.scala

@@ -1,129 +1,27 @@
 package com.example.model.classification
 
 import java.util
-
 import com.example.SparkConnect
 import com.example.entity.Model
 import com.example.model.helper.Utils
 import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
 import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
 import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
-import org.apache.spark.ml.{Pipeline, PipelineModel}
+import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
 import org.apache.spark.sql.{Dataset, Row}
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
+import com.example.util.Constant
 
 
-/**
-  * Decision Tree Dataset
-  * Created by lucas on 2016/11/30.
-  */
 @Component
-class DTClassification extends SparkConnect{
-
-  var trainingSet:Dataset[Row]  = _
-  var testData:Dataset[Row] = _
-  @Autowired
-  val utils: Utils = null
-  /**
-    *
-    * @param libsvmFile
-    * @param category
-    * @param trainingSetOccupy
-    * @throws java.lang.Exception
-    * @return decision tree model{PipelineModel}
-    */
-  @throws(classOf[Exception])
-  def dtTraining(libsvmFile: String,
-                 category: Int,
-                 trainingSetOccupy: Float): PipelineModel = {
-
-
-    // Load the data stored in LIBSVM format as a DataFrame.
-    val data = sparkSession.read.format("libsvm").load(libsvmFile)
-    log.info("\n\nhas getten data\n\n")
-
-    data.show()
-    // Index labels, adding metadata to the label column.
-    // Fit on whole dataset to include all labels in index.
-    val labelIndexer = new StringIndexer()
-      .setInputCol("label")
-      .setOutputCol("indexedLabel")
-      .fit(data)
-
-    // Automatically identify categorical features, and index them.
-    val featureIndexer = new VectorIndexer()
-      .setInputCol("features")
-      .setOutputCol("indexedFeatures")
-      .setMaxCategories(category) // features with > 4 distinct values are treated as continuous.
-      .fit(data)
+class DTClassification extends DefaultClassification {
 
-    data.show()
-    log.info("\n\n data split\n\n")
-    // Split the data into training and test sets (30% held out for testing).
-    val Array(trainingSet, testSet) = data.randomSplit(Array(trainingSetOccupy, 1 - trainingSetOccupy))
-    testData = testSet
-
-    log.info("\n\nready to train\n\n")
-    // Train a DecisionTree model.
+  override def buildClassifier(params: util.Map[String, Object]): PipelineStage = {
     val dt = new DecisionTreeClassifier()
       .setLabelCol("indexedLabel")
       .setFeaturesCol("indexedFeatures")
-
-    // Convert indexed labels back to original labels.
-    val labelConverter = new IndexToString()
-      .setInputCol("prediction")
-      .setOutputCol("predictedLabel")
-      .setLabels(labelIndexer.labels)
-
-    // Chain indexers and tree in a Pipeline.
-    val pipeline = new Pipeline()
-      .setStages(Array(labelIndexer, featureIndexer, dt, labelConverter))
-
-    // Train model. This also runs the indexers.
-    val model = pipeline.fit(trainingSet)
-
-    model
+    dt
   }
 
-  /**
-    *
-    * @param pipelineModel
-    * @throws java.lang.Exception
-    * @return result of decision tree structure{java.util.Map[String, String]}
-    */
-  @throws(classOf[Exception])
-  def dtResult(pipelineModel: PipelineModel, modelId: Int): util.HashMap[String, String] = {
-    val rslMap = new util.HashMap[String, String]()
-    // Make predictions.
-    val predictions = pipelineModel.transform(testData)
-
-
-
-    // Select (prediction, true label) and compute test error.
-    val evaluator = new MulticlassClassificationEvaluator()
-      .setLabelCol("indexedLabel")
-      .setPredictionCol("prediction")
-
-    val weightedRecall = evaluator.setMetricName("weightedRecall").evaluate(predictions);
-    val accuracy = evaluator.setMetricName("accuracy").evaluate(predictions)
-    //println("Test Error = " + (1.0 - accuracy))
-    log.info("\n\nTest Error = " + (1.0 - accuracy));
-    val weightedPrecision = evaluator.setMetricName("weightedPrecision").evaluate(predictions);
-
-    val f1 = evaluator.setMetricName("f1").evaluate(predictions);
-
-    val treeModel = pipelineModel.stages(2).asInstanceOf[DecisionTreeClassificationModel]
-    //println("Learned classification tree model:\n" + treeModel.toDebugString)
-    log.info("\n\nLearned classification tree model:\n" + treeModel.toDebugString);
-
-    rslMap.put("Accuracy: ", accuracy.toString)
-    rslMap.put("weightedPrecision", weightedPrecision.toString)
-    rslMap.put("weightedRecall", weightedRecall.toString)
-    rslMap.put("f1", f1.toString)
-
-
-    return rslMap
-
-  }
 }

java_compile/src/main/scala/com/example/model/classification/DefaultClassification.scala

@@ -10,7 +10,7 @@ import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
 import org.apache.spark.sql.{Dataset, Row}
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
-import com.example.util.Constant
+import com.example.util.{Constant, TrainParamEnum, TrainParamReader}
 
 import scala.collection.mutable
 
@@ -32,18 +32,22 @@ class DefaultClassification extends SparkConnect {
     val trainSet = sparkSession.read.format("libsvm").load(libsvmFileUri)
     trainSet.show()
 
-    val maxCategories = params.get(Constant.TRAIN_PARAM_MAX_CATEGORY).asInstanceOf[Int]
+    //获取最大种类，若读取失败，获取默认值
+    val maxCategories = TrainParamReader.readParam(params,TrainParamEnum.MAX_CATEGORIES).asInstanceOf[Int]
+    log.info("maxCategories="+maxCategories)
+
     val featureIndexer = new VectorIndexer()
       .setInputCol("features")
       .setOutputCol("indexedFeatures")
       .setMaxCategories(maxCategories)
-      .fit(trainSet)
 
     trainSet.show()
 
-    // Split the data into training and test sets (30% held out for testing).
-    //为了对训练效果进行评估，将训练集进一步划分为子训练集与子测试集
-    var Array(subTrainSet:Dataset[Row], subTestSet:Dataset[Row]) = trainSet.randomSplit(Array(0.7, 1 - 0.7))
+    //读取训练集占比
+    val trainingDataSetOccupy: Double = TrainParamReader.readParam(params,TrainParamEnum.TRAIN_DATA_SET_OCCUPY).asInstanceOf[Double]
+
+    //训练集切分
+    var Array(subTrainSet:Dataset[Row], subTestSet:Dataset[Row]) = trainSet.randomSplit(Array(trainingDataSetOccupy, 1 - trainingDataSetOccupy ))
 
     /*******************************************/
     //需要由子类实现
@@ -88,26 +92,15 @@ class DefaultClassification extends SparkConnect {
     return (pipelineModel, resultMap)
   }
 
-
   /**
-   * 默认实现：逻辑回归
    * @author   fanyanpeng
    * @date 2023/3/22 16:44
    * @param params 参数列表
    * @return org.apache.spark.ml.PipelineStage
    */
   def buildClassifier(params:util.Map[String,Object]):PipelineStage  = {
-    val maxIter = params.get(Constant.TRAIN_PARAM_MAX_ITER).asInstanceOf[Int]
-    val regParam = params.get(Constant.TRAIN_PARAM_REG_PARAM).asInstanceOf[Float]
-    val elasticNetParam = params.get(Constant.TRAIN_PARAM_ELASTICNET_PARAM).asInstanceOf[Float]
-
-    val lr = new LogisticRegression()
-      .setLabelCol("label")
-      .setFeaturesCol("indexedFeatures")
-      .setMaxIter(maxIter)
-      .setRegParam(regParam)
-      .setElasticNetParam(elasticNetParam)
-    lr
+    //由子类实现
+    null
   }
 
 

java_compile/src/main/scala/com/example/model/classification/GDBTClassification.scala

@@ -1,123 +1,20 @@
 package com.example.model.classification
-
 import java.util
-
-import com.example.SparkConnect
-import com.example.model.helper.Utils
-import org.apache.spark.ml.{Pipeline, PipelineModel}
-import org.apache.spark.ml.classification.{GBTClassificationModel, GBTClassifier}
-import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
-import org.apache.spark.sql.{Dataset, Row}
-import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
+import org.apache.spark.ml.{PipelineStage}
+import org.apache.spark.ml.classification.{GBTClassifier}
 import org.springframework.stereotype.Component
-import org.springframework.beans.factory.annotation.Autowired
-/**
-  * Created by tangwb on 2017/3/24.
-  * 梯度下降决策树
-  */
-@Component
-class GDBTClassification extends SparkConnect{
 
-  var trainingSet:Dataset[Row]  = _
-  var testData:Dataset[Row] = _
 
-  @Autowired
-  val utils: Utils = null
-  /**
-    *
-    * @param libsvmFile
-    * @param category
-    * @param trainingSetOccupy
-    * @throws java.lang.Exception
-    * @return gdbt model{PipelineModel}
-    */
 
-  @throws(classOf[Exception])
-  def dtTraining(libsvmFile: String,
-                 category: Int,
-                 trainingSetOccupy: Float
-                ): PipelineModel ={
-    // Load the data stored in LIBSVM format as a DataFrame.
-    val data = sparkSession.read.format("libsvm").load(libsvmFile)
-    // Index labels, adding metadata to the label column.
-    // Fit on whole dataset to include all labels in index.
-    val labelIndexer = new StringIndexer()
-      .setInputCol("label")
-      .setOutputCol("indexedLabel")
-      .fit(data)
-    // Automatically identify categorical features, and index them.
-    val featureIndexer = new VectorIndexer()
-      .setInputCol("features")
-      .setOutputCol("indexedFeatures")
-      .setMaxCategories(category) // features with > 4 distinct values are treated as continuous.
-      .fit(data)
-    // Split the data into training and test sets (30% held out for testing).
-    val Array(trainingSet, testSet) = data.randomSplit(Array(trainingSetOccupy, 1 - trainingSetOccupy))
-    testData = testSet
 
+@Component
+class GDBTClassification extends DefaultClassification{
+  override def buildClassifier(params: util.Map[String, Object]): PipelineStage = {
     // Train a gdbt model.
     val gbt = new GBTClassifier().setLabelCol("indexedLabel")
       .setFeaturesCol("indexedFeatures")
-
-    // Convert indexed labels back to original labels.
-    val labelConverter = new IndexToString()
-      .setInputCol("prediction")
-      .setOutputCol("predictedLabel")
-      .setLabels(labelIndexer.labels)
-
-    // Chain indexers and tree in a Pipeline.
-    val pipeline = new Pipeline()
-      .setStages(Array(labelIndexer, featureIndexer, gbt, labelConverter))
-
-    // Train model. This also runs the indexers.
-    val model = pipeline.fit(trainingSet)
-
-    model
-
+    gbt
   }
 
-  /**
-    *
-    * @param pipelineModel
-    * @throws java.lang.Exception
-    * @return
-    */
-
-  @throws(classOf[Exception])
-  def dtResult(pipelineModel: PipelineModel, modelId: Int): util.HashMap[String, String] = {
-    val rslMap = new util.HashMap[String, String]()
-    // Make predictions.
-    val predictions = pipelineModel.transform(testData)
-
-
-
-
-    // Select example rows to display.
-    val matrixs = predictions.select("predictedLabel", "label")
-    matrixs.show()
-    matrixs.createOrReplaceTempView("Matrixs")
-    println("12121"+matrixs.dtypes)
-    // Select (prediction, true label) and compute test error.
-    val evaluator = new MulticlassClassificationEvaluator()
-      .setLabelCol("indexedLabel")
-      .setPredictionCol("prediction")
-    //.setMetricName("accuracy")
-
-    val accuracy = evaluator.evaluate(predictions)
-    println("Test Error = " + (1.0 - accuracy))
-    val treeModel = pipelineModel.stages(2).asInstanceOf[GBTClassificationModel]
-
-    println("Learned classification tree model:\n" + treeModel.toDebugString)
-    val weightedRecall = evaluator.setMetricName("weightedRecall").evaluate(predictions)
-    val weightedPrecision = evaluator.setMetricName("weightedPrecision").evaluate(predictions);
-
-    val f1 = evaluator.setMetricName("f1").evaluate(predictions);
-    rslMap.put("Accuracy: ", accuracy.toString)
-    rslMap.put("weightedPrecision", weightedPrecision.toString)
-    rslMap.put("weightedRecall", weightedRecall.toString)
-    rslMap.put("f1", f1.toString)
-
-    return rslMap
-  }
 
 }

java_compile/src/main/scala/com/example/model/classification/LRClassification.scala

@@ -3,7 +3,7 @@ package com.example.model.classification
 import java.util
 import com.example.SparkConnect
 import com.example.model.helper.Utils
-import org.apache.spark.ml.{Pipeline, PipelineModel}
+import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
 import org.apache.spark.ml.classification.{BinaryLogisticRegressionSummary, DecisionTreeClassificationModel, GBTClassifier, LogisticRegression, LogisticRegressionModel}
 import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
 import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
@@ -12,12 +12,26 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
 
 import collection.JavaConverters._
-
+import com.example.util.{Constant, TrainParamEnum, TrainParamReader}
 
 @Component
 class LRClassification extends DefaultClassification {
 
 
+  override def buildClassifier(params: util.Map[String, Object]): PipelineStage = {
+
+    val maxIter = TrainParamReader.readParam(params,TrainParamEnum.MAXI_ITER).asInstanceOf[Int]
+    val regParam = TrainParamReader.readParam(params,TrainParamEnum.REG_PARAM).asInstanceOf[Double]
+    val elasticNetParam = TrainParamReader.readParam(params,TrainParamEnum.ELASTIC_NET_PARAM).asInstanceOf[Double]
+
+    val lr = new LogisticRegression()
+      .setLabelCol("label")
+      .setFeaturesCol("indexedFeatures")
+      .setMaxIter(maxIter)
+      .setRegParam(regParam)
+      .setElasticNetParam(elasticNetParam)
+    lr
+  }
 
 
 }

java_compile/src/main/scala/com/example/model/classification/MulPerClassification.scala

@@ -1,58 +1,28 @@
 package com.example.model.classification
 
 import java.util
-
 import com.example.SparkConnect
 import com.example.model.helper.Utils
 import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier, MultilayerPerceptronClassificationModel, MultilayerPerceptronClassifier}
 import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
-import org.apache.spark.ml.{Pipeline, PipelineModel}
+import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
 import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
 import org.apache.spark.sql.{Dataset, Row}
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
+import com.example.util.{Constant, TrainParamEnum, TrainParamReader}
+
 
-/**
-  * Created by twb on 2017/4/26.
-  */
-@Component
-class MulPerClassification extends SparkConnect{
-  var trainingSet:Dataset[Row]  = _
-  var testData:Dataset[Row] = _
 
-  @Autowired
-  val utils: Utils = null
 
-  @throws(classOf[Exception])
-  def dtTraining(libsvmFile: String,
-                 category: Int,
-                 trainingSetOccupy: Float, featureNum: Int): PipelineModel = {
-    // Load the data stored in LIBSVM format as a DataFrame.
-    val data = sparkSession.read.format("libsvm").load(libsvmFile)
-    // Index labels, adding metadata to the label column.
-    // Fit on whole dataset to include all labels in index.
-    val labelIndexer = new StringIndexer()
-      .setInputCol("label")
-      .setOutputCol("indexedLabel")
-      .fit(data)
 
-    // Automatically identify categorical features, and index them.
-    val featureIndexer = new VectorIndexer()
-      .setInputCol("features")
-      .setOutputCol("indexedFeatures")
-      .setMaxCategories(category) // features with > 4 distinct values are treated as continuous.
-      .fit(data)
-    // Split the data into training and test sets (30% held out for testing).
-    val Array(trainingSet1, testSet) = data.randomSplit(Array(trainingSetOccupy, 1 - trainingSetOccupy))
-    testData = testSet
-    trainingSet = trainingSet1
+@Component
+class MulPerClassification extends DefaultClassification{
+  override def buildClassifier(params: util.Map[String, Object]): PipelineStage = {
 
-    // specify layers for the neural network:
-    // input layer of size 4 (features), two intermediate of size 5 and 4
-    // and output of size 3 (classes)
+    val featureNum = TrainParamReader.readParam(params,TrainParamEnum.FEATURE_NUM).asInstanceOf[Int]
     val layers = Array[Int](2, 5, 4, 5)
     layers(0) = featureNum
-    // Train a MLPC  model.
     val mlpc = new MultilayerPerceptronClassifier()
       .setLabelCol("indexedLabel")
       .setFeaturesCol("indexedFeatures")
@@ -60,60 +30,7 @@ class MulPerClassification extends SparkConnect{
       .setBlockSize(128)
       .setSeed(1234L)
       .setMaxIter(100)
-
-    // Convert indexed labels back to original labels.
-    val labelConverter = new IndexToString()
-      .setInputCol("prediction")
-      .setOutputCol("predictedLabel")
-      .setLabels(labelIndexer.labels)
-
-    // Chain indexers and tree in a Pipeline.
-    val pipeline = new Pipeline()
-      .setStages(Array(labelIndexer, featureIndexer, mlpc, labelConverter))
-
-    // Train model. This also runs the indexers.
-    val model = pipeline.fit(trainingSet)
-
-    model
-  }
-
-
-  @throws(classOf[Exception])
-  def dtResult(pipelineModel: PipelineModel, modelId: Int): util.HashMap[String, String] = {
-    val rslMap = new util.HashMap[String, String]()
-    // Make predictions.
-    val predictions = pipelineModel.transform(testData)
-
-    // Select example rows to display.
-    val metrixs = predictions.select("predictedLabel", "label")//.show(100)
-    //metrixs.filter(tmp=>tmp==0).count()
-    // Select (prediction, true label) and compute test error.
-    val evaluator = new MulticlassClassificationEvaluator()
-      .setLabelCol("indexedLabel")
-      .setPredictionCol("prediction")
-
-
-    val weightedRecall = evaluator.setMetricName("weightedRecall").evaluate(predictions);
-    val accuracy = evaluator.setMetricName("accuracy").evaluate(predictions)
-
-    println("Test Error = " + (1.0 - accuracy))
-    val weightedPrecision = evaluator.setMetricName("weightedPrecision").evaluate(predictions);
-
-    val f1 = evaluator.setMetricName("f1").evaluate(predictions);
-
-    val mlpcModel = pipelineModel.stages(2).asInstanceOf[MultilayerPerceptronClassificationModel]
-    println("Learned classification tree model:\n" + mlpcModel.toString())
-
-    rslMap.put("Accuracy: ", accuracy.toString)
-    rslMap.put("weightedPrecision", weightedPrecision.toString)
-    rslMap.put("weightedRecall", weightedRecall.toString)
-    rslMap.put("f1", f1.toString)
-
-    return rslMap
-
+    mlpc
   }
 
-
-
-
 }

java_compile/src/main/scala/com/example/model/classification/NaiveBayesClassification.scala

@@ -1,110 +1,28 @@
 package com.example.model.classification
 
 import java.util
-
 import com.example.SparkConnect
 import com.example.model.helper.Utils
-import org.apache.spark.ml.{Pipeline, PipelineModel}
+import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
 import org.apache.spark.ml.classification.{MultilayerPerceptronClassificationModel, MultilayerPerceptronClassifier, NaiveBayes, NaiveBayesModel}
 import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
 import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
 import org.apache.spark.sql.{Dataset, Row}
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
+import com.example.util.Constant
 
-/**
-  * Created by twb on 2017/4/27.
-  */
-@Component
-class NaiveBayesClassification extends SparkConnect{
-  var trainingSet:Dataset[Row]  = _
-  var testData:Dataset[Row] = _
-
-  @Autowired
-  val utils: Utils = null
-
-  @throws(classOf[Exception])
-  def dtTraining(libsvmFile: String,
-                 category: Int,
-                 trainingSetOccupy: Float): PipelineModel = {
-    // Load the data stored in LIBSVM format as a DataFrame.
-    val data = sparkSession.read.format("libsvm").load(libsvmFile)
-    // Index labels, adding metadata to the label column.
-    // Fit on whole dataset to include all labels in index.
-    val labelIndexer = new StringIndexer()
-      .setInputCol("label")
-      .setOutputCol("indexedLabel")
-      .fit(data)
-
-    // Automatically identify categorical features, and index them.
-    val featureIndexer = new VectorIndexer()
-      .setInputCol("features")
-      .setOutputCol("indexedFeatures")
-      .setMaxCategories(category) // features with > 4 distinct values are treated as continuous.
-      .fit(data)
-    // Split the data into training and test sets (30% held out for testing).
-    val Array(trainingSet1, testSet) = data.randomSplit(Array(trainingSetOccupy, 1 - trainingSetOccupy))
-    testData = testSet
-    trainingSet = trainingSet1
-
-
-    // Train a NBC  model.
-    val nbc = new NaiveBayes().setLabelCol("indexedLabel").setFeaturesCol("indexedFeatures")
-    // Convert indexed labels back to original labels.
-    val labelConverter = new IndexToString()
-      .setInputCol("prediction")
-      .setOutputCol("predictedLabel")
-      .setLabels(labelIndexer.labels)
-
-    // Chain indexers and tree in a Pipeline.
-    val pipeline = new Pipeline()
-      .setStages(Array(labelIndexer, featureIndexer, nbc, labelConverter))
-
-    // Train model. This also runs the indexers.
-    val model = pipeline.fit(trainingSet)
-
-    model
-  }
-
-
-  @throws(classOf[Exception])
-  def dtResult(pipelineModel: PipelineModel, modelId: Int): util.HashMap[String, String] = {
-    val rslMap = new util.HashMap[String, String]()
-    // Make predictions.
-    val predictions = pipelineModel.transform(testData)
-
-    predictions.printSchema()
-    predictions.show()
-    // Select example rows to display.
-    //predictions.select("predictedLabel", "label", "features").show(100)
-
-    // Select (prediction, true label) and compute test error.
-    val evaluator = new MulticlassClassificationEvaluator()
-      .setLabelCol("indexedLabel")
-      .setPredictionCol("prediction")
-
-
-    val weightedRecall = evaluator.setMetricName("weightedRecall").evaluate(predictions);
-    val accuracy = evaluator.setMetricName("accuracy").evaluate(predictions)
 
-    println("Test Error = " + (1.0 - accuracy))
-    val weightedPrecision = evaluator.setMetricName("weightedPrecision").evaluate(predictions);
 
-    val f1 = evaluator.setMetricName("f1").evaluate(predictions);
 
-    val mlpcModel = pipelineModel.stages(2).asInstanceOf[NaiveBayesModel]
-    println("Learned classification tree model:\n" + mlpcModel.toString())
-
-    rslMap.put("Accuracy: ", accuracy.toString)
-    rslMap.put("weightedPrecision", weightedPrecision.toString)
-    rslMap.put("weightedRecall", weightedRecall.toString)
-    rslMap.put("f1", f1.toString)
+@Component
+class NaiveBayesClassification extends DefaultClassification{
 
-    return rslMap
+  override def buildClassifier(params: util.Map[String, Object]): PipelineStage = {
+    val nbc = new NaiveBayes().setLabelCol("indexedLabel")
+      .setFeaturesCol("indexedFeatures")
+    nbc
   }
 
-
-
-
 }
 

java_compile/src/main/scala/com/example/model/classification/RFClassification.scala

@@ -1,113 +1,30 @@
 package com.example.model.classification
 
 import java.util
-
 import com.example.SparkConnect
 import com.example.model.helper.Utils
-import org.apache.spark.ml.{Pipeline, PipelineModel}
+import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
 import org.apache.spark.ml.classification.{DecisionTreeClassifier, RandomForestClassificationModel, RandomForestClassifier}
 import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
 import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
 import org.apache.spark.sql.{Dataset, Row}
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
+import com.example.util.Constant
+
+
+
 
-/**
-  * Created by tangwb on 2017/3/20.
-  */
 @Component
-class RFClassification extends SparkConnect{
-  var trainingSet:Dataset[Row]  = _
-  var testData:Dataset[Row] = _
+class RFClassification extends  DefaultClassification{
 
-  @Autowired
-  val utils: Utils = null
-  /**
-    *
-    * @param libsvmFile
-    * @param category
-    * @param trainingSetOccupy
-    * @throws java.lang.Exception
-    * @return
-    */
-  @throws(classOf[Exception])
-  def dtTraining(libsvmFile: String,
-                 category: Int,
-                 trainingSetOccupy: Float): PipelineModel ={
-    // Load the data stored in LIBSVM format as a DataFrame.
-    val data = sparkSession.read.format("libsvm").load(libsvmFile)
-    // Index labels, adding metadata to the label column.
-    // Fit on whole dataset to include all labels in index.
-    val labelIndexer = new StringIndexer()
-      .setInputCol("label")
-      .setOutputCol("indexedLabel")
-      .fit(data)
-    // Automatically identify categorical features, and index them.
-    val featureIndexer = new VectorIndexer()
-      .setInputCol("features")
-      .setOutputCol("indexedFeatures")
-      .setMaxCategories(category) // features with > 4 distinct values are treated as continuous.
-      .fit(data)
-    // Split the data into training and test sets (30% held out for testing).
-    val Array(trainingSet, testSet) = data.randomSplit(Array(trainingSetOccupy, 1 - trainingSetOccupy))
-    testData = testSet
-    // Train a RandomForest model.
+  override def buildClassifier(params: util.Map[String, Object]): PipelineStage = {
     val rf = new RandomForestClassifier()
       .setLabelCol("indexedLabel")
       .setFeaturesCol("indexedFeatures")
 
-    // Convert indexed labels back to original labels.
-    val labelConverter = new IndexToString()
-      .setInputCol("prediction")
-      .setOutputCol("predictedLabel")
-      .setLabels(labelIndexer.labels)
-
-    // Chain indexers and tree in a Pipeline.
-    val pipeline = new Pipeline()
-      .setStages(Array(labelIndexer, featureIndexer, rf, labelConverter))
-
-    // Train model. This also runs the indexers.
-    val model = pipeline.fit(trainingSet)
-
-    model
-
+    rf
   }
 
 
-  /**
-    *
-    * @param pipelineModel
-    * @throws java.lang.Exception
-    * @return
-    */
-  @throws(classOf[Exception])
-  def dtResult(pipelineModel: PipelineModel, modelId: Int): util.HashMap[String, String] = {
-    val rslMap = new util.HashMap[String, String]()
-    // Make predictions.
-    val predictions = pipelineModel.transform(testData)
-
-    // Select example rows to display.
-    predictions.select("predictedLabel", "label", "features").show(100)
-
-    // Select (prediction, true label) and compute test error.
-    val evaluator = new MulticlassClassificationEvaluator()
-      .setLabelCol("indexedLabel")
-      .setPredictionCol("prediction")
-      //.setMetricName("accuracy")
-    val accuracy = evaluator.evaluate(predictions)
-    println("Test Error = " + (1.0 - accuracy))
-
-    val treeModel = pipelineModel.stages(2).asInstanceOf[RandomForestClassificationModel]
-    println("Learned classification tree model:\n" + treeModel.toDebugString)
-    val weightedRecall = evaluator.setMetricName("weightedRecall").evaluate(predictions)
-    val weightedPrecision = evaluator.setMetricName("weightedPrecision").evaluate(predictions);
-    println(weightedPrecision)
-    println(weightedRecall)
-    val f1 = evaluator.setMetricName("f1").evaluate(predictions);
-    rslMap.put("Accuracy: ", accuracy.toString)
-    rslMap.put("weightedPrecision", weightedPrecision.toString)
-    rslMap.put("weightedRecall", weightedRecall.toString)
-    rslMap.put("f1", f1.toString)
-    return rslMap
-  }
 }

java_compile/src/main/scala/com/example/util/TrainParamReader.scala

@@ -0,0 +1,11 @@
+package com.example.util
+
+import java.util
+
+object TrainParamReader {
+
+  def readParam(params: util.Map[String, Object],trainParamEnum: TrainParamEnum): Any = {
+    return params.getOrDefault(trainParamEnum.getParamName, trainParamEnum.getDefaultValue)
+  }
+
+}