Spark SVM分類器

package Spark_MLlib

import java.util.Properties

import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.classification.SVMWithSGD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
/**
 * SGD:一種優化方法,目的使目標函數的值最小.SGD指stochastic gradient descent,即隨機梯度下降.是梯度下降的batch版本.
 * 這么做的好處在於：
 * 當訓練數據太多時,利用整個數據集更新往往時間上不顯示.batch的方法可以減少機器的壓力,並且可以更快地收斂.
 * 當訓練集有很多冗余時(類似的樣本出現多次),batch方法收斂更快.以一個極端情況為例,若訓練集前一半和后一半梯度相同.那么如果前一半作為一個batch,
 * 后一半作為另一個batch,那么在一次遍歷訓練集時,batch的方法向最優解前進兩個step,而整體的方法只前進一個step.
  */
object SVM{
  def main(args: Array[String]): Unit = {
    val spark=SparkSession.builder().master("local").getOrCreate()
    val train_data=spark.sparkContext.textFile("file:///home/soyo/下載/Hadoop+Spark+Hbase/Spark訓練數據/train_after.csv")
    val test_data=spark.sparkContext.textFile("file:///home/soyo/下載/Hadoop+Spark+Hbase/Spark訓練數據/test_after.csv")
   val train=train_data.map{x=>
       val parts=x.split(",")
         LabeledPoint(parts(4).toDouble,Vectors.dense(parts(0).toDouble,parts(1).toDouble,parts(2).toDouble,parts(3).toDouble))
        }
  //  train.foreach(println)
    val test=test_data.map{x=>
      val y=x.split(",")
      LabeledPoint(y(4).toDouble,Vectors.dense(y(0).toDouble,y(1).toDouble,y(2).toDouble,y(3).toDouble))
    }
    //numIterations:迭代次數,默認是100
    val numIterations = 600
    val model=SVMWithSGD.train(train,numIterations)
    //val model=new SVMWithSGD().run(train)    //兩種求model都行
    //清除默認閾值,這樣會輸出原始的預測評分,即帶有確信度的結果。
    model.clearThreshold()
    val scoreAndLabels=test.map{x=>
      val score=model.predict(x.features)
      score+" "+x.label
    }
    scoreAndLabels.foreach(println)

    val rebuyRDD=scoreAndLabels.map(_.split(" "))
    //設置模式信息
    val schema=StructType(List(StructField("score",StringType,true),StructField("label",StringType,true)))
   //創建Row對象,每個Row對象都是rowRDD中的一行
    val rowRDD=rebuyRDD.map(x=>Row(x(0).trim,x(1).trim))
    //建立模式和數據之間的關系
    val rebuyDF=spark.createDataFrame(rowRDD,schema)
    //prop變量保存JDBC連接參數
    val prop=new Properties()
    prop.put("user","root")
    prop.put("password","密碼")
    prop.put("driver","com.mysql.jdbc.Driver")  //表示驅動程序是com.mysql.jdbc.Driver
    rebuyDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/數據庫名","數據庫表名",prop)

  }

}

Spark 機器學習庫從 1.2 版本以后被分為兩個包：

• spark.mllib 包含基於RDD的原始算法API。Spark MLlib 歷史比較長，在1.0 以前的版本即已經包含了，提供的算法實現都是基於原始的RDD。
• spark.ml 則提供了基於DataFrames 高層次的API,可以用來構建機器學習工作流（PipeLine).ML Pipeline 彌補了原始 MLlib 庫的不足，向用戶提供了一個基於 DataFrame 的機器學習工作流式 API 套件。

從Spark2.0開始，Spark機器學習庫基於RDD的API進入維護模式（即不增加任何新的特性),很有可能於3.0以后的版本的時候會移除出MLLib