refactor: 减少代码冗余，将生成libsvm文件的职责集中到LibsvmAdapter类；对变量名进行修改，添加注释

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

@@ -23,6 +23,8 @@ import scala.collection.JavaConverters._
 @Component
 class FitTestData extends SparkConnect{
 
+  @Autowired
+  val libsvmAdapter: LibsvmAdapter = null
   @Autowired
   val fileInfoDao: FileInfoDao = null
 
@@ -102,59 +104,11 @@ class FitTestData extends SparkConnect{
     //添加部分到此结束
 
 
-
-    val featuresDF = dataFrameForPredict.select(keyFeature,fieldNames:_*).persist()
-    featuresDF.schema.foreach(println)
-
-    val strFeatures = featuresDF.schema.filter(_.dataType.equals(StringType)).map(_.name)
-
-
-    val notNullFeatureDF = featuresDF.na.fill("NA", strFeatures)
-
-
-    val indexers = strFeatures.map(field => {
-      new StringIndexer()
-        .setInputCol(field)
-        .setOutputCol(field+"_index")
-        .fit(notNullFeatureDF)
-    })
-
-//    transform String to Indexer
-    var tmpDF:DataFrame = null
-    if (indexers.nonEmpty) {
-      println("Indexers: " + indexers.size)
-      val pipeline = new Pipeline().setStages(indexers.toArray[StringIndexerModel])
-      println("Pipeline: " + pipeline.getStages.length)
-      tmpDF = pipeline.fit(notNullFeatureDF).transform(notNullFeatureDF)
-    } else {
-      tmpDF = notNullFeatureDF
-    }
-    tmpDF.show()
-
-    val transformDF = tmpDF.drop(strFeatures: _*)
-    val keyFeatureIdx  = if(strFeatures.nonEmpty && strFeatures.head.equals(keyFeature)) transformDF.columns.length - strFeatures.size else 0
-
-
-    val labeledData = transformDF.collect().map(row => {
-      val seq = row.toSeq
-      val key = seq(keyFeatureIdx)
-      val tmp = seq.splitAt(keyFeatureIdx)
-      val tail = tmp._1 ++ tmp._2.tail
-      val features = tail.zipWithIndex.map(pair =>
-        if (pair._1 == null) null else (pair._2 + 1) + ":" + pair._1).filter(_ != null)
-      if (key == null) null else key + " " + features.mkString(" ")
-    }).filter(_ != null)
-
-    val tmpFileLoc = hdfsUri + config.getId + "_ver2.libsvm"
-
-    //hdfsDao.deleteFileInHdfs(tmpFileLoc, true)
-    val labeledDS = sparkSession.createDataset(labeledData)
-
-    labeledDS.repartition(1).write.mode(SaveMode.Overwrite).text(tmpFileLoc)
-    log.info("Write to the file: " + tmpFileLoc)
+    //libsvm文件统一资源定位符。
+    val libsvmFileUri = libsvmAdapter.dataFrame2Libsvm(dataFrameForPredict,fileId,keyFeature,fieldNames)
 
     // Load the data stored in LIBSVM format as a DataFrame.
-    val data = sparkSession.read.format("libsvm").load(tmpFileLoc)
+    val data = sparkSession.read.format("libsvm").load(libsvmFileUri)
     // Index labels, adding metadata to the label column.
     // Fit on whole dataset to include all labels in index.
 
@@ -204,7 +158,6 @@ class FitTestData extends SparkConnect{
           (columns, pair._1.toString)
         }).toMap.asJava
       }).toList.asJava
-
       res
     }
   }

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

@@ -29,36 +29,22 @@ class LibsvmAdapter extends SparkConnect{
 
   //val spark: SparkSession = sparkSession
 
-  /**
-    *
-    * @param fileId
-    * @param keyFeature
-    * @param trainingFeature
-    * @throws java.lang.Exception
-    * @return libSVM{String} file location in HDFS
-    * keyFeature means the classification feature
-    * tainingFeature means the training Items(This one show be continuous value<or Integer>)
-    * isInteger means the feature is int or other dataType
-    * @since 1.0.0
-    */
-  @throws(classOf[Exception])
-  def parquet2Libsvm(fileId: Int, keyFeature: String,
-                     trainingFeature: Array[String]): String = {
-
-    val fileLocs = fileLocDao.findById(fileId)
-    println(fileLocs)
-    val filepath = fileLocs.getLocation
-    val filename = fileLocs.getFilename
-
-    log.info("file loc: " + filepath)
-
-    val spark = sparkSession
-    val parquetFile = sparkSession.read.parquet(filepath)
-
-
-    val featuresDF = parquetFile.select(keyFeature, trainingFeature:_*).persist()
-    featuresDF.schema.foreach(println)
 
+  /**
+   *
+   * @author   fanyanpeng
+   * @date 2023/3/4 8:06
+   * @param dataFrame 完整的数据集，包括数据所有列。对于待预测数据，也已添加标签列，并置为标签可选值的第一个。
+   * @param fileFlag  文件标志，可以是文件Id，也可以是文件名。用于标志libsvm文件。
+   * @param label 标签
+   * @param features  特征数组，包括选取的特征
+   * @return java.lang.String
+   */
+  def dataFrame2Libsvm(dataFrame: DataFrame,fileFlag:Any, label: String,
+                       features: Array[String]): String = {
+
+    val featuresDF = dataFrame.select(label, features: _*).persist()
+    featuresDF.show()
 
     val strFeatures = featuresDF.schema.filter(_.dataType.equals(StringType)).map(_.name)
 
@@ -69,12 +55,12 @@ class LibsvmAdapter extends SparkConnect{
     val indexers = strFeatures.map(field => {
       new StringIndexer()
         .setInputCol(field)
-        .setOutputCol(field+"_index")
+        .setOutputCol(field + "_index")
         .fit(notNullFeatureDF)
     })
 
     //transform String to Indexer
-    var tmpDF:DataFrame = null
+    var tmpDF: DataFrame = null
     if (indexers.nonEmpty) {
       println("Indexers: " + indexers.size)
       val pipeline = new Pipeline().setStages(indexers.toArray[StringIndexerModel])
@@ -87,7 +73,7 @@ class LibsvmAdapter extends SparkConnect{
 
     val transformDF = tmpDF.drop(strFeatures: _*)
     val keyFeatureIdx =
-      if(strFeatures.nonEmpty && strFeatures.head.equals(keyFeature))
+      if (strFeatures.nonEmpty && strFeatures.head.equals(label))
         transformDF.columns.length - strFeatures.size
       else 0
 
@@ -102,15 +88,47 @@ class LibsvmAdapter extends SparkConnect{
       if (key == null) null else key + " " + features.mkString(" ")
     }).filter(_ != null)
 
-    val tmpFileLoc = hdfsUri + filename + "123.libsvm"
+    val libsvmFileUri = hdfsUri + fileFlag + "-2023.libsvm"
 
     //hdfsDao.deleteFileInHdfs(tmpFileLoc, true)
-    implicit val encoder=org.apache.spark.sql.Encoders.STRING//添加字符串类型编码器
+    implicit val encoder = org.apache.spark.sql.Encoders.STRING //添加字符串类型编码器
     val labeledDS = sparkSession.createDataset(labeledData)
 
-    labeledDS.repartition(1).write.mode(SaveMode.Overwrite).text(tmpFileLoc)
+    labeledDS.repartition(1).write.mode(SaveMode.Overwrite).text(libsvmFileUri)
+
+    log.info("Write to the file: " + libsvmFileUri)
+    libsvmFileUri
 
-    log.info("Write to the file: " + tmpFileLoc)
-    tmpFileLoc
   }
+  /**
+    *
+    * @param fileId
+    * @param keyFeature
+    * @param trainingFeature
+    * @throws java.lang.Exception
+    * @return libSVM{String} file location in HDFS
+    * keyFeature means the classification feature
+    * tainingFeature means the training Items(This one show be continuous value<or Integer>)
+    * isInteger means the feature is int or other dataType
+    * @since 1.0.0
+    */
+  @throws(classOf[Exception])
+  def parquet2Libsvm(fileId: Int, label: String,
+                     features: Array[String]): String = {
+
+    val fileLocs = fileLocDao.findById(fileId)
+    println(fileLocs)
+    val filepath = fileLocs.getLocation
+    val filename = fileLocs.getFilename
+
+    log.info("file loc: " + filepath)
+
+    val spark = sparkSession
+    val parquetFile = sparkSession.read.parquet(filepath)
+
+
+    dataFrame2Libsvm(parquetFile,fileId,label,features)
+  }
+
+
 }