refactor: 模型代码复用

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

@@ -16,6 +16,7 @@ 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 lombok.Data;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +25,9 @@ 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;
@@ -224,7 +227,7 @@ public class ModelService {
      * @param model
      * @return
      */
-    public boolean updateResOfModel(int modelId, HashMap<String, String> resOfModel, Object model) throws Exception {
+    public boolean updateResOfModel(int modelId, Map<String, String> resOfModel, Object model) throws Exception {
 
         // 将模型持久化（上传到hdfs）
         utils.storeModel(model,modelId);
@@ -274,12 +277,16 @@ public class ModelService {
                  regParam: Float,
                  elasticNetParam: Float
             * */
-                    PipelineModel lrModel = lrClassification.lrTraining(libsvm,
-                            Integer.parseInt(modelArguments.get("MaxIter")),
-                            Float.parseFloat(modelArguments.get("RegParam")),
-                            Float.parseFloat(modelArguments.get("ElasticNetParam")));
-                    log.info("\n\nstore Model result\n\n");
-                    HashMap<String,String> lrRst = lrClassification.dtResult(lrModel,modelId);
+
+                    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);

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

@@ -0,0 +1,14 @@
+package com.example.util;
+
+/**
+ * @author fanyanpeng
+ * @date 2023/3/22 16:25
+ */
+public class Constant {
+
+    public static final String TRAIN_PARAM_MAX_CATEGORY = "Category";
+    public static final String TRAIN_PARAM_TRAINING_DATASET_OCCUPY="TrainingDataSetOccupy";
+    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";
+}

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

@@ -0,0 +1,114 @@
+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, LogisticRegression, LogisticRegressionModel}
+import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
+import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
+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 scala.collection.mutable
+
+/**
+ * 分类方法的默认实现
+ * @author   fanyanpeng
+ * @date 2023/3/22 16:15
+ */
+class DefaultClassification extends SparkConnect {
+
+
+  @Autowired
+  val utils: Utils = null
+
+  @throws(classOf[Exception])
+  def training(libsvmFileUri: String,
+               params: util.Map[String,Object]): (PipelineModel,util.Map[String, String]) = {
+
+    val trainSet = sparkSession.read.format("libsvm").load(libsvmFileUri)
+    trainSet.show()
+
+    val maxCategories = params.get(Constant.TRAIN_PARAM_MAX_CATEGORY).asInstanceOf[Int]
+    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 classifier = buildClassifier(params)
+    /*******************************************/
+
+    val pipeline = new Pipeline()
+      .setStages(Array(featureIndexer, classifier))
+
+    // Train model. This also runs the indexers.
+    val pipelineModel:PipelineModel = pipeline.fit(subTrainSet)
+
+
+    val resultMap:util.Map[String,String] = new util.HashMap[String,String]()
+    // Make predictions.
+    val predictions = pipelineModel.transform(subTestSet)
+
+    // Select (prediction, true label) and compute test error.
+    //为了简便，使用了多分类
+    val evaluator = new MulticlassClassificationEvaluator()
+      .setLabelCol("label")
+      .setPredictionCol("prediction")
+
+
+    val weightedRecall = evaluator.setMetricName("weightedRecall").evaluate(predictions);
+    val accuracy = evaluator.setMetricName("accuracy").evaluate(predictions)
+    log.info("Test Error = " + (1.0 - accuracy))
+    val weightedPrecision = evaluator.setMetricName("weightedPrecision").evaluate(predictions);
+    val f1 = evaluator.setMetricName("f1").evaluate(predictions);
+
+
+    val model = pipelineModel.stages(1).asInstanceOf[LogisticRegressionModel]
+    //println("Learned classification tree model:\n" + treeModel.toDebugString)
+    log.info("\n\nLearned classification Logistic model:\n" + model.toString());
+
+
+    resultMap.put("Accuracy: ", accuracy.toString)
+    resultMap.put("weightedPrecision", weightedPrecision.toString)
+    resultMap.put("weightedRecall", weightedRecall.toString)
+    resultMap.put("f1", f1.toString)
+
+    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
+  }
+
+
+}

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

@@ -13,113 +13,11 @@ import org.springframework.stereotype.Component
 
 import collection.JavaConverters._
 
-/**
-  * Created by lucas on 2016/12/1.
-  * Logistic回归
-  */
-@Component
-class LRClassification extends SparkConnect{
-
-  var trainingSet: Dataset[Row] = _
-  var testData: Dataset[Row] = _
-  @Autowired
-  val utils: Utils = null
-
-  /*
-  * initial params
-  * */
-
-  /**
-   *
-   * @param libsvmFile
-   * @param maxIter
-   * @param regParam
-   * @param elasticNetParam
-   * @throws java.lang.Exception
-   * @return
-   */
-  @throws(classOf[Exception])
-  def lrTraining(libsvmFile: String,
-                 maxIter: Int,
-                 regParam: Float,
-                 elasticNetParam: Float): PipelineModel = {
-    val data = sparkSession.read.format("libsvm").load(libsvmFile)
-    data.printSchema()
-    data.show()
-
-
-    // Automatically identify categorical features, and index them.
-    val featureIndexer = new VectorIndexer()
-      .setInputCol("features")
-      .setOutputCol("indexedFeatures")
-      .setMaxCategories(5)
-      .fit(data)
-
-    data.show()
-
-    // Split the data into training and test sets (30% held out for testing).
-    val Array(trainingSet, testSet) = data.randomSplit(Array(0.7, 1 - 0.7))
-    testData = testSet
-
-    val lr = new LogisticRegression()
-      .setLabelCol("label")
-      .setFeaturesCol("indexedFeatures")
-      .setMaxIter(maxIter)
-      .setRegParam(regParam)
-      .setElasticNetParam(elasticNetParam)
-
-
-
-    // Chain indexers and tree in a Pipeline.
-    val pipeline = new Pipeline()
-      .setStages(Array(featureIndexer, lr))
-
-    // Train model. This also runs the indexers.
-    val model = pipeline.fit(trainingSet)
-
-    model
-  }
-
-  /**
-    *
-    * @param lrModel
-    * @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.
-    // Make predictions.
-    val predictions = pipelineModel.transform(testData)
-
-
-    // Select (prediction, true label) and compute test error.
-    val evaluator = new MulticlassClassificationEvaluator()
-      .setLabelCol("label")
-      .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 model = pipelineModel.stages(1).asInstanceOf[LogisticRegressionModel]
-    //println("Learned classification tree model:\n" + treeModel.toDebugString)
-    log.info("\n\nLearned classification Logistic model:\n" + model.toString());
 
+@Component
+class LRClassification extends DefaultClassification {
 
-    rslMap.put("Accuracy: ", accuracy.toString)
-    rslMap.put("weightedPrecision", weightedPrecision.toString)
-    rslMap.put("weightedRecall", weightedRecall.toString)
-    rslMap.put("f1", f1.toString)
 
 
-    return rslMap
-  }
 
 }