Spark中常用的算法

Spark中常用的算法：

分類算法屬于監督式學習，使用類标簽已知的樣本建立一個分類函數或分類模型，應用分類模型，能把資料庫中的類标簽未知的資料進行歸類。分類在資料挖掘中是一項重要的任務，目前在商業上應用最多，常見的典型應用場景有流失預測、精确營銷、客戶擷取、個性偏好等。MLlib 目前支援分類算法有：邏輯回歸、支援向量機、樸素貝葉斯和決策樹。

案例：導入訓練資料集，然後在訓練集上執行訓練算法，最後在所得模型上進行預測并計算訓練誤差。

import org.apache.spark.SparkContext

import org.apache.spark.mllib.classification.SVMWithSGD

import org.apache.spark.mllib.regression.LabeledPoint

// 加載和解析資料檔案

val data = sc.textFile("mllib/data/sample_svm_data.txt")

val parsedData = data.map { line =>

  val parts = line.split(' ')

  LabeledPoint(parts(0).toDouble, parts.tail.map(x => x.toDouble).toArray)

}

// 設定疊代次數并進行進行訓練

val numIterations = 20

val model = SVMWithSGD.train(parsedData, numIterations)

// 統計分類錯誤的樣本比例

val labelAndPreds = parsedData.map { point =>

val prediction = model.predict(point.features)

(point.label, prediction)

val trainErr = labelAndPreds.filter(r => r._1 != r._2).count.toDouble / parsedData.count

println("Training Error = " + trainErr)

回歸算法屬于監督式學習，每個個體都有一個與之相關聯的實數标簽，并且我們希望在給出用于表示這些實體的數值特征後，所預測出的标簽值可以盡可能接近實際值。MLlib 目前支援回歸算法有：線性回歸、嶺回歸、Lasso和決策樹。

案例：導入訓練資料集，将其解析為帶标簽點的RDD，使用 LinearRegressionWithSGD 算法建立一個簡單的線性模型來預測标簽的值，最後計算均方差來評估預測值與實際值的吻合度。

import org.apache.spark.mllib.regression.LinearRegressionWithSGD

val data = sc.textFile("mllib/data/ridge-data/lpsa.data")

  val parts = line.split(',')

  LabeledPoint(parts(0).toDouble, parts(1).split(' ').map(x => x.toDouble).toArray)

//設定疊代次數并進行訓練

val model = LinearRegressionWithSGD.train(parsedData, numIterations)

// 統計回歸錯誤的樣本比例

val valuesAndPreds = parsedData.map { point =>

val MSE = valuesAndPreds.map{ case(v, p) => math.pow((v - p), 2)}.reduce(_ + _)/valuesAndPreds.count

println("training Mean Squared Error = " + MSE)

聚 類算法屬于非監督式學習，通常被用于探索性的分析，是根據“物以類聚”的原理，将本身沒有類别的樣本聚內建不同的組，這樣的一組資料對象的集合叫做簇，并 且對每一個這樣的簇進行描述的過程。它的目的是使得屬于同一簇的樣本之間應該彼此相似，而不同簇的樣本應該足夠不相似，常見的典型應用場景有客戶細分、客 戶研究、市場細分、價值評估。MLlib 目前支援廣泛使用的KMmeans聚類算法。

案例：導入訓練資料集，使用 KMeans 對象來将資料聚類到兩個類簇當中，所需的類簇個數會被傳遞到算法中，然後計算集内均方差總和 (WSSSE)，可以通過增加類簇的個數 k 來減小誤差。 實際上，最優的類簇數通常是 1，因為這一點通常是WSSSE圖中的 “低谷點”。

import org.apache.spark.mllib.clustering.KMeans

val data = sc.textFile("kmeans_data.txt")

val parsedData = data.map( _.split(' ').map(_.toDouble))

// 設定疊代次數、類簇的個數

val numClusters = 2

// 進行訓練

val clusters = KMeans.train(parsedData, numClusters, numIterations)

// 統計聚類錯誤的樣本比例

val WSSSE = clusters.computeCost(parsedData)

println("Within Set Sum of Squared Errors = " + WSSSE)

協同過濾常被應用于推薦系統，這些技術旨在補充使用者-商品關聯矩陣中所缺失的部分。MLlib目前支援基于模型的協同過濾，其中使用者和商品通過一小組隐語義因子進行表達，并且這些因子也用于預測缺失的元素。

案例：導入訓練資料集，資料每一行由一個使用者、一個商品和相應的評分組成。假設評分是顯性的，使用預設的ALS.train()方法，通過計算預測出的評分的均方差來評估這個推薦模型。

import org.apache.spark.mllib.recommendation.ALS

import org.apache.spark.mllib.recommendation.Rating

val data = sc.textFile("mllib/data/als/test.data")

val ratings = data.map(_.split(',') match {

case Array(user, item, rate) => Rating(user.toInt, item.toInt, rate.toDouble)

})

// 設定疊代次數

val model = ALS.train(ratings, 1, 20, 0.01)

// 對推薦模型進行評分

val usersProducts = ratings.map{ case Rating(user, product, rate) => (user, product)}

val predictions = model.predict(usersProducts).map{

case Rating(user, product, rate) => ((user, product), rate)

val ratesAndPreds = ratings.map{

}.join(predictions)

val MSE = ratesAndPreds.map{

case ((user, product), (r1, r2)) => math.pow((r1- r2), 2)

}.reduce(_ + _)/ratesAndPreds.count

println("Mean Squared Error = " + MSE)

本文轉自張昺華-sky部落格園部落格，原文連結：http://www.cnblogs.com/bonelee/p/7125615.html，如需轉載請自行聯系原作者