Spark機器學習庫之線性回歸

前言：ML是Spark裡面的機器學習庫。它使得基礎的一些機器學習演算法變得簡單易用。Spark提供了常規的機器學習演算法，比如說分類、回歸、聚類、協同過濾演算法。還包含了類似於sk-learn中的一些特徵提取以及特徵選擇。

現在我們基於Spark2.21這個版本，主要是介紹基於DataFrame的機器學習該如何實現，基本參照官網上的例子。來實現一遍，以便學習。在介紹一下相關的演算法。

首先我們先介紹一些什麼是線性回歸，是根據自變數x1,x2...xn以及應變數y的實際樣本，去估計兩者之間近似的函數關係f，類似於我們有三個點(1,1) (2,2) (3,3)代表的是xy，我們需要找到一個函數來滿足這三個點的一個f,很明顯我們可以找到是y=x。當自變數x只有一個時，我們稱之為一元線性回歸，多個就是多元線性回歸。

回歸是一種監督學習，監督學習分兩種，一種是回歸一種就是分類，回歸主要是用於預測連續值的比如說根據特徵預測房價呀，分類就是類別判斷，比如說垃圾郵件識別。

線性回歸的表示方式是：

f(x) = w1X1 + w2X2 + ...+wnXn+b

其實就是我們初中學習的一元方程式。w1是斜率，b是截距。

我們需要做的就是找出最優的w值和b值，使得我們的預測值誤差最小。

舉個簡單的列子：

房價預測，我們收集了一些房價的信息，包括房子的大小(hd)、房子的地理位置(hs)、房子的房間個數(hr)、房子的樓層(hl)等等。然後我們收集了它對應的房價。

我們就需要找到一個函數，通過這些已知的數據。

f(x) = w1*hd + w2*hs + w3*hr + w4*hl +..+ b

通過已知數據慢慢擬合找到最合適的w組以及b。

然後就是優化，一般的優化我們經常才有梯度下降的方式去優化。最小二乘法。

大概就是這個樣子。然後我們就是用Spark來簡單的演練一下吧。

Spark的實現

我們先看看源碼：這是Spark的一個優化W和b的過程

/**

* Add a new training instance to this LeastSquaresAggregator, and update the loss and gradient

* of the objective function.

*

*@param instanceThe instance of data point to be added.

*@returnThis LeastSquaresAggregator object.

*/

defadd(instance: Instance):this.type= {

instancematch{caseInstance(label,weight,features) =>

if(weight ==0.0)return this

valdiff =dot(features,effectiveCoefficientsVector) - label / labelStd +offset

if(diff !=) {

vallocalGradientSumArray =gradientSumArray

vallocalFeaturesStd =featuresStd

features.foreachActive { (index,value) =>

if(localFeaturesStd(index) !=0.0&& value !=0.0) {

localGradientSumArray(index) += weight * diff * value / localFeaturesStd(index)

}

}

lossSum+= weight * diff * diff /2.0

}

totalCnt+=1

weightSum+= weight

this

}

}

然後還有個預測：

override protected defpredict(features: Vector):Double= {

dot(features,coefficients) + intercept

}

就是w*x + b

保存：

override protected defsaveImpl(path:String):Unit= {

// Save metadata and Params

DefaultParamsWriter.saveMetadata(instance,path,sc)

// Save model data: intercept, coefficients

valdata =Data(instance.intercept,instance.coefficients)

valdataPath =newPath(path,"data").toString

sparkSession.createDataFrame(Seq(data)).repartition(1).write.parquet(dataPath)

}

}

保存的就是w和b兩個變數，所以說我們在外面看著很陌生，一旦深入進去，就覺得很簡單了，這就是在求一個變數的過程。訓練結果也就是我們所謂的模型其實就是這些個變數。

然後我們用Spark的例子來實現一下吧。

packageorg.apache.spark.examples

importorg.apache.spark.sql.SparkSession

importorg.apache.spark.ml.regression.LinearRegression

importorg.apache.spark.ml.evaluation.RegressionEvaluator

objectSparkDemo {

defmain(args: Array[String]) {

valspark = SparkSession

.builder

.appName("LinearRegression").master("local")

.getOrCreate()

// Load training data

valtraining = spark.read.format("libsvm").load("file:\D:\sample_linear_regression_data.txt")

valArray(trainData,testData) = training.randomSplit(Array(0.8,0.2))

//模型設置參數

vallr =newLinearRegression().setMaxIter(10)

.setRegParam(0.3)

.setElasticNetParam(0.8)

//建立模型

vallrModel = lr.fit(trainData)

//列印學習到的參數

println(s"Coefficients:$Intercept:$")

//列印一些模型的參數

valtrainingSummary = lrModel.summary

println(s"numIterations:$")

println(s"objectiveHistory: [$]")

trainingSummary.residuals.show()

println(s"RMSE:$")

println(s"r2:$")

//預測數據

valpredict = lrModel.transform(testData)

predict.show(3)

//顯示測試數據的指標

println("test rmse:"+newRegressionEvaluator().setMetricName("rmse").evaluate(predict))

println("test mse:"+newRegressionEvaluator().setMetricName("mse").evaluate(predict))

println("test r2:"+newRegressionEvaluator().setMetricName("r2").evaluate(predict))

println("test mae:"+newRegressionEvaluator().setMetricName("mae").evaluate(predict))

spark.stop()

}

}

// scalastyle:on println

我們來看看裡面對應的輸出：

差不多這樣就在Spark上實現了一個線性回歸。但是呢其中還是有很多參數是需要我們進一步去調試跟進的。

下面我在介紹一下該演算法在Spark中的一些參數吧。

setMaxIter()設置迭代次數。默認是100.

@Since("1.3.0")

defsetMaxIter(value:Int):this.type= set(maxIter,value)

setDefault(maxIter->100)

setTol() 設置容錯，當我們隨著迭代次數的增加，誤差值會越來越小，當誤差值下於該設置的值時，就會停止迭代。默認1E-6

@Since("1.4.0")

defsetTol(value: Double):this.type= set(tol,value)

setDefault(tol ->1E-6)

setRegParam() 設置正則化項係數，正則化項主要用於防止過擬合現象，當數據集比較小，特徵維數又比較多的時候，易出現過擬合，此時可以考慮增大正則化項係數 默認0.0

@Since("1.3.0")

defsetRegParam(value:Double):this.type= set(regParam,value)

setDefault(regParam->0.0)

setElasticNetParam() 正則化範式比(默認0.0)，正則化一般有兩種範式：L1和L2正則。L1一般用於特徵的稀疏化，L2一般用於防止過擬合。這裡的參數即設置L1範式的佔比，默認0.0即只使用L2範式

@Since("1.4.0")

defsetElasticNetParam(value:Double):this.type= set(elasticNetParam,value)

setDefault(elasticNetParam->0.0)

setFeaturesCol() 設置特徵列 默認是「features」

setLabelCol() 設置標籤列 默認是「label」

setWeightCol（） 設置權重，這裡是設個各個參數的權重比，默認都為1。

@Since("1.6.0")

defsetWeightCol(value:String):this.type= set(weightCol,value)

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！