chores: 修改hdfsServerAddress->hdfsUri（统一资源标志，为本机对应路径）; 添加个人操作空间，易于管理

java_compile/src/main/java/com/example/config/AppConfig.java

@@ -4,6 +4,8 @@ import lombok.Data;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
+import java.util.Map;
+
 /**
  *
  * Created by lucas on 2016/10/10.
@@ -22,11 +24,18 @@ public class AppConfig {
     private String executorMemory;
     @Value("${com.jtang.picture.uri}")
     private String picPath;
+    @Value("${com.jtang.hdfs.serverAddr}user-space/${user.name}/")
+    private String hdfsUri;
+
+
+
+    public String getHdfsUri() {
+        return hdfsUri;
+    }
 
     public String getAppName() {
         return appName;
     }
-
     public String getSparkServer() {
         return sparkServerAddr;
     }

java_compile/src/main/java/com/example/dao/HdfsDao.java

@@ -11,6 +11,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
+import java.util.Map;
 
 /**
  * this file is used to complete hdfs operates and mongodb operates
@@ -24,19 +25,42 @@ public class HdfsDao {
     //connect to hdfs
     private FileSystem hdfs;
     private Configuration conf;
+
+    private String hdfsServerAddr;
     private String hdfsUri;  //get hdfs Address
 
+
     @Autowired
-    public HdfsDao(AppConfig appConfig) {
+    public HdfsDao(AppConfig appConfig) {   //在初始化方法中使用到的bean需要以注入初始化方法的形式导入，否则读取到的引用可能是null。
         this.appConfig = appConfig;
+        //hdfs主机地址 hdfs://ip:port/
+        hdfsServerAddr= appConfig.getHdfsServerAddr();
+        //hdfs存储位置 hdfs://ip:port/user-space/*/
+        hdfsUri= appConfig.getHdfsUri();
         conf = new Configuration();
-        hdfsUri = appConfig.getHdfsServerAddr();
+
+        setUserSpace();
         setConf();
     }
 
+    //保证hdfs存储位置可以访问
+    private void setUserSpace(){
+        //先根据主机ip设置配置
+        conf.set("fs.defaultFS", hdfsServerAddr);
+        try {
+            hdfs = FileSystem.get(conf);
+            //若不存在文件夹，新建
+            //若文件夹已存在，新建失败
+            hdfs.mkdirs(new Path(hdfsUri));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
     //set the configuration and get connection
     private void setConf(){
         conf.set("fs.defaultFS", hdfsUri);
+        log.info("当前主机下的hdfs目录是："+hdfsUri);
         //System.out.println(conf);
         try {
             hdfs = FileSystem.get(conf);

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

@@ -20,7 +20,7 @@ trait SparkConnect {
 
   var sparkSession: SparkSession = _
   var sparkContext: SparkContext = _
-  var hdfsServer: String = _
+  var hdfsUri: String = _
 
   /**
     * initial of spark connection info
@@ -29,7 +29,7 @@ trait SparkConnect {
   def Init() = {
     val serverAddr = conf.getSparkServer
     val appName = conf.getAppName
-    hdfsServer = conf.getHdfsServerAddr
+    hdfsUri = conf.getHdfsUri
     val executorMemory = conf.getExecutorMemory
     log.info("Get sparkServer " + serverAddr)
     val sconf = new SparkConf().setMaster(serverAddr).setAppName(appName)

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

@@ -86,11 +86,11 @@ class CsvAdapter extends SparkConnect{
     }
     res.show()*/
     //3. now simply write to a parquet file
-    df.coalesce(1).write.mode(SaveMode.Overwrite).parquet(hdfsServer + outFile)
+    df.coalesce(1).write.mode(SaveMode.Overwrite).parquet(hdfsUri + outFile)
     //4. delete the csv file
     hdfsDao.deleteFileInHdfs(csvHdfsLoc, true)
     //5. Store filename to db
-    val fileId = fileInfoDao.save(new FileInfo(hdfsServer + outFile, userId, outFile)).getId
+    val fileId = fileInfoDao.save(new FileInfo(hdfsUri + outFile, userId, outFile)).getId
     //6. 将header存入数据库中
     df.schema.fields.map(field => {
       headerInfoDao.save(new HeaderInfo(fileId, field.name, field.dataType.catalogString))

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

@@ -77,8 +77,10 @@ class FitTestData extends SparkConnect{
     import spark.implicits._
     val parquetFile = sparkSession.read.parquet(filepath)
 
-//    parquetFile.select()
-    val featuresDF = parquetFile.select(keyFeature, fieldNames: _*).persist()
+
+    //这一段是直接复制的Lib，获得数据时，不需要处理label数据，全部删除
+    parquetFile.select()
+    val featuresDF = parquetFile.select(keyFeature,fieldNames:_*).persist()
     featuresDF.schema.foreach(println)
 
     val strFeatures = featuresDF.schema.filter(_.dataType.equals(StringType)).map(_.name)
@@ -94,7 +96,7 @@ class FitTestData extends SparkConnect{
         .fit(notNullFeatureDF)
     })
 
-    //transform String to Indexer
+//    transform String to Indexer
     var tmpDF:DataFrame = null
     if (indexers.nonEmpty) {
       println("Indexers: " + indexers.size)
@@ -120,13 +122,12 @@ class FitTestData extends SparkConnect{
       if (key == null) null else key + " " + features.mkString(" ")
     }).filter(_ != null)
 
-    val tmpFileLoc = hdfsServer + config.getId + "_ver2.libsvm"
+    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)
 
     // Load the data stored in LIBSVM format as a DataFrame.

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

@@ -102,7 +102,7 @@ class LibsvmAdapter extends SparkConnect{
       if (key == null) null else key + " " + features.mkString(" ")
     }).filter(_ != null)
 
-    val tmpFileLoc = hdfsServer + filename + "123.libsvm"
+    val tmpFileLoc = hdfsUri + filename + "123.libsvm"
 
     //hdfsDao.deleteFileInHdfs(tmpFileLoc, true)
     implicit val encoder=org.apache.spark.sql.Encoders.STRING//添加字符串类型编码器

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

@@ -3,7 +3,6 @@ package com.example.model.classification
 import java.util
 
 import com.example.SparkConnect
-import com.example.dao.HdfsDao
 import com.example.model.helper.Utils
 import org.apache.spark.ml.{Pipeline, PipelineModel}
 import org.apache.spark.ml.classification.{GBTClassificationModel, GBTClassifier}
@@ -21,8 +20,6 @@ class GDBTClassification extends SparkConnect{
 
   var trainingSet:Dataset[Row]  = _
   var testData:Dataset[Row] = _
-  @Autowired
-  val hdfsDao: HdfsDao = null
 
   @Autowired
   val utils: Utils = null

java_compile/src/main/scala/com/example/model/helper/Utils.scala

@@ -23,7 +23,7 @@ class Utils extends SparkConnect{
 
   @throws(classOf[Exception])
   def storeModel(anyModel: Any, modelId: Int): Boolean = {
-    val modelPath = hdfsServer  + "_" + modelId + ".model"
+    val modelPath = hdfsUri  + "_" + modelId + ".model"
 
     hdfsDao.deleteFileInHdfs(modelPath, true)