feat: 针对训练集的方法

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

@@ -5,6 +5,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.example.controller.vo.DLModelVO;
 import com.example.controller.vo.ModelVO;
 import com.example.dao.*;
+import com.example.data.DataSetHelper;
 import com.example.data.FitTestData;
 import com.example.data.LibsvmAdapter;
 import com.example.entity.*;
@@ -80,6 +81,11 @@ public class ModelService {
     private FileService fileService;
     @Autowired
     private Utils utils;
+
+
+    @Autowired
+    private DataSetHelper dataSetHelper;
+
     /**
      *
      * @param modelVO
@@ -252,7 +258,10 @@ public class ModelService {
         String keyFeature = features.get(features.size()-1);
         features.remove(features.size()-1);
         int featureNum = features.size();
-        String libsvm = libsvmAdapter.parquet2Libsvm(configInfo.getFileInfoId(), keyFeature, features.toArray(new String[0]));
+//        String libsvm = libsvmAdapter.parquet2Libsvm(configInfo.getFileInfoId(), keyFeature, features.toArray(new String[0]));
+
+        String libsvm = dataSetHelper.processTrainDataSet(modelId,configInfo.getFileInfoId(),keyFeature, features.toArray(new String[0]));
+
 
         String modelDetailName = modelType.getModelDetailName();
         HashMap<String,String> modelArguments = new Json2Object().Json2HashMap(trainingModel.getArguments());

java_compile/src/main/scala/com/example/SparkConnect.scala

@@ -19,7 +19,7 @@ trait SparkConnect {
   val log: Logger = LoggerFactory.getLogger(getClass)
 
   var sparkSession: SparkSession = _
-  var sparkContext: SparkContext = _
+
   var hdfsUri: String = _
 
   /**
@@ -35,7 +35,7 @@ trait SparkConnect {
     val sconf = new SparkConf().setMaster(serverAddr).setAppName(appName)
 
     sconf.set("spark.executor.memory", executorMemory)
-    sparkContext = SparkContext.getOrCreate(sconf)
+    val sparkContext = SparkContext.getOrCreate(sconf)
     sparkSession = SparkSession.builder().master(serverAddr).appName(appName).getOrCreate()
   }
 }

java_compile/src/main/scala/com/example/data/DataSetHelper.scala

@@ -0,0 +1,166 @@
+package com.example.data
+
+import com.example.SparkConnect
+import com.example.dao.{FileInfoDao, HdfsDao}
+import org.apache.spark.ml.{Pipeline, PipelineModel}
+import org.apache.spark.ml.feature.{StringIndexer, VectorAssembler}
+import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.{Column, DataFrame, SaveMode, SparkSession}
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Component
+
+import javax.annotation.PostConstruct
+
+
+@Component
+class DataSetHelper extends SparkConnect{
+
+  @Autowired
+  val fileLocDao: FileInfoDao = null
+
+  @Autowired
+  val hdfsDao: HdfsDao = null
+
+
+  def fillNull(dataFrame: DataFrame): DataFrame ={
+
+    val stringTypeFeatures = dataFrame.schema.filter(_.dataType.equals(StringType)).map(_.name)
+    return dataFrame.na.fill("NA",stringTypeFeatures)
+  }
+
+  def getStringTypeFeatureTransformerUri(modelId:Int): String ={
+    hdfsUri+"pipeline_StringTypeFeatureTransformer_modelId-"+modelId
+  }
+
+  def getFileLibsvmUri(fileFlag: Int): String = {
+    hdfsUri + fileFlag + "-2023.libsvm"
+  }
+
+
+  def getLabelStringIndexerUri(modelId: Int):String = {
+    hdfsUri+"LabelStringIndexer_modelId-"+modelId
+  }
+
+
+  @throws(classOf[Exception])
+  def processTrainDataSet(modelId:Int,
+                          fileId: Int,
+                          label: String,
+                          features: Array[String]): String ={
+
+    val spark = sparkSession
+    val trainDataFile = fileLocDao.findById(fileId)
+    val trainDataFileUri = trainDataFile.getLocation
+    log.info("trainDataFileUri =  "+trainDataFileUri)
+
+    //训练集读取
+    var trainSet:DataFrame = spark.read.parquet(trainDataFileUri)
+
+    //保留标签|特征
+    trainSet = trainSet.select(label,features:_*).persist()
+
+    //去除标签列为空的数据(无法进行学习）
+    trainSet = trainSet.filter("'"+ label+" ' is not null")
+
+    //填充特征空值
+    trainSet = fillNull(trainSet)
+    log.info("训练集-已填充空值")
+    trainSet.show(5)
+
+    //取出类型为string的特征
+    val stringTypeFeatures = trainSet.schema
+      .filter(_.dataType.equals(StringType))  //是string类型
+      .filter(!_.name.equals(label))   //且不是label列
+      .map(_.name)
+
+    //字符类型特征 -> 数值类型特征
+    val stringTypeFeatureIndexers = stringTypeFeatures.map(stringTypeFeature => {
+      new StringIndexer()
+        .setInputCol(stringTypeFeature)
+        .setOutputCol(stringTypeFeature + "_index")
+        .fit(trainSet)
+    })
+
+    if(stringTypeFeatures.nonEmpty){ //若存在字符类型特征
+      log.info("需要进行字符型类型转换")
+      log.info("待转换字符型特征有"+stringTypeFeatures.length+"个：")
+      log.info(stringTypeFeatures.toString())
+
+      log.info("构建pipeline进行转换")
+      val pipeline = new Pipeline().setStages(stringTypeFeatureIndexers.toArray)
+      val pipelineModel : PipelineModel = pipeline.fit(trainSet)
+
+      val pipelineUri = getStringTypeFeatureTransformerUri(modelId)
+      log.info("转换模型适配完成，正在保存到服务器，保存地址："+pipelineUri)
+      pipelineModel.write.overwrite().save(pipelineUri)
+
+      trainSet = pipelineModel.transform(trainSet)
+    }
+
+    //训练集-去除非数值的特征
+    trainSet = trainSet.drop(stringTypeFeatures:_*)
+
+    log.info("训练集-去除非数值的特征：")
+    trainSet.show(5)
+
+
+    //找到为string类型的标签（0个或者1个）
+    val stringLabel = trainSet.schema.filter(_.name.equals(label)).filter(_.dataType.equals(StringType)).map(_.name)
+    if(stringLabel.nonEmpty){ //需要转换，将label的转换方法持久化
+      val labelStringIndexer = new StringIndexer().setInputCol(label).setOutputCol("label").fit(trainSet)
+      val labelStringIndexerUri = getLabelStringIndexerUri(modelId)
+      labelStringIndexer.write.overwrite().save(labelStringIndexerUri)
+      trainSet = labelStringIndexer.transform(trainSet)
+      log.info("label转换模型已经存储到远端，uri:"+labelStringIndexerUri)
+      trainSet.show(5)
+    }
+
+    val columns = trainSet.columns
+
+    //过滤标签，取出特征列表
+    val featureList:Array[String] = columns.filter(columnName => (!columnName.equals(label) && !columnName.equals("label")))
+
+    //将特征列表合成向量
+    val assembler = new VectorAssembler()
+      .setInputCols(featureList)
+      .setOutputCol("features")
+
+    // 若label是字符串类型，那么将选择新的数值型的label列
+    var selectedLabel= label
+    if(stringLabel.nonEmpty){
+      selectedLabel = "label"
+    }
+
+    val assembledData = assembler.transform(trainSet)
+      .select(selectedLabel,"features")
+      .toDF("label","features") //重命名
+
+    log.info("写入libsvm的数据格式如下：")
+    assembledData.show(5)
+
+    val libsvmFileUri = getFileLibsvmUri(fileId)
+
+
+//    implicit val encoder = org.apache.spark.sql.Encoders.STRING
+
+
+    assembledData.write.mode(SaveMode.Overwrite) //没有上面那句声明，无法写入成功
+      .format("libsvm")
+      .save(libsvmFileUri)
+
+    log.info("libsvm写入，uri: "+libsvmFileUri)
+
+    return libsvmFileUri
+
+  }
+
+  def processTestDataSet(): String = {
+
+
+    "libsvmFileUri"
+  }
+
+
+
+
+}

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

@@ -4,7 +4,7 @@ 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.{BinaryLogisticRegressionSummary, DecisionTreeClassificationModel, LogisticRegression, LogisticRegressionModel}
+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}
 import org.apache.spark.sql.{Dataset, Row}