在完成基於大資料平臺的圖書館推薦系統後，最近把學習的中心放在機器學習上面。在接下來的幾個月中，希望自己能弄明白常見機器學習演算法的原理，並且能在spark平臺上進行實踐。

在我的機器學習學習和實踐之路的一個本書是《Spark機器學習》，這本書雖然比較舊，但是寫的還是比較好。書裡講了各種常見的機器學習演算法，並且在spark平臺上進行了實戰。在學習到此書第四章--構建基於spark的推薦系統引擎時覺得ALS演算法在待分析資料具備使用者對物品的打分時是一個很好用的演算法，所以做記錄如下。

推薦系統的產生主要是因為在公司的一些日常業務需求中需要通過使用者的一些購買、借閱、點評等記錄來準確預測使用者下一階段的類消費傾向以產生更大的經濟效益等。在進入大資料時代的今天，資料量巨大，對於企業而言如何構建一個高效的推薦系統來對使用者進行個性化一直是一個在研究和改進的問題。推薦系統的系統的核心在於推薦演算法，常見的推薦演算法有

ALS中文名作交替最小二乘法，在機器學習中，ALS特指使用最小二乘法求解的一個協同過濾演算法，是協同過濾中的一種。ALS演算法是2008年以來，用的比較多的協同過濾演算法。它已經整合到Spark的Mllib庫中，使用起來比較方便。從協同過濾的分類來說，ALS演算法屬於User-Item CF，也叫做混合CF，因為它同時考慮了User和Item兩個方面，即即可基於使用者進行推薦又可基於物品進行推薦。

一般而言使用者只會購買物品集中的極少數部分產品，並對其進行打分。考慮下面這樣一個包含使用者的打分矩陣（列為使用者u1-u6，行為物品I1-I8），我們可以看到這個使用者的評分矩陣是十分稀疏的，有很多使用者的購買的記錄是空的，而且在現實業務中，使用者的評分矩陣會更加的稀疏。如何通過這樣一個稀疏矩陣，對使用者進行協同推薦使用者可能很喜歡的物品對於推薦系統而言是一種很大的考驗。

在spark MLlib 機器學習庫中目前推薦模型只包含基於矩陣分解(matrix factorization)的實現。具體的分解思路，找出兩個低維的矩陣，使得它們的乘積是原始矩陣。因此這也是一種降維技術。假設我們的使用者和物品分別是U和I，那對應的“使用者-物品”矩陣的維度為U×I，類似圖一所示：

而言找到和“使用者-物品“矩陣近似的k維（低階）矩陣，最終還是要求出如下兩個矩陣：一個用於表示使用者U×k維矩陣，以及一個表徵物品的I×k維矩陣。這兩個矩陣也稱為因子矩陣，他們的矩陣乘積便是原始評級資料的一個近似值。值得注意的是，原始評級矩陣通常很稀疏，但因子矩陣卻是稠密的，如圖二所示：

ALS是求解矩陣分解問題的一種最優化方法，它功能強大，效果理想而且被證明相對容易實現。這使得它很適合如Spark這樣的平臺。

ALS的實現原理是迭代式求解一系列最小二乘迴歸問題。在每次迭代時，固定使用者因子矩陣或者是物品因子矩陣中的一個，然後用固定的這個矩陣以及評級資料來更新另一個矩陣。之後，被更新的矩陣被固定住，再更新另外一個矩陣。如此迭代，知道模型收斂（或者是迭代了預設好的次數）。

書中的利用ALS模式實現推薦的程式碼例項及註釋

// 載入觀眾影評資料集（觀眾ID，影片ID，評分）
val rawData = sc.textFile("dataSet/MLDataSet/u.data")
rawData.first()
val rawRating = rawData.map(_.split("\t").take(3))

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

import  org.apache.spark.mllib.recommendation.Rating
// 將rawRating由陣列型別轉換為rating（user,movie,rating）型別
//Rating(user,product,rating)
val rating = rawRating.map{case Array(user,movie,rating)=>Rating(user.toInt,movie.toInt,rating.toDouble)}
//訓練模型，rank，iterations，lambda引數值分分別為50,10,0.1.
val model = ALS.train(rating,50,10,0.01)
// 基於使用者的推薦
//預測出使用者789對123電影的評分
val productRating = model.predict(789,123)
// 返回使用者789的前10推薦電影
val userId = 789
val K = 10
val topKRecs = model.recommendProducts(userId,K)
print(topKRecs.mkString("\n"))
// 載入電影資料集（編號，電影名（上映年）....）
val movies = sc.textFile("dataSet/MLDataSet/u.item")
//只取（編號，電影名（上映年）），生成的是一個key->value
val titles = movies.map(line=>line.split("\\|").take(2)).map(array=>(array(0).toInt,array(1))).collectAsMap()
titles(123)
//檢視使用者789點評過的所有電影
val moviesForUser = rating.keyBy(_.user).lookup(789)
println(moviesForUser.size)
//檢視觀眾點評資料集中評分最高的前10影片並電影編號相應轉換成電影名
moviesForUser.sortBy(-_.rating).take(10).map(rating=>(titles(rating.product),rating.rating)).foreach(println)
//返回使用者789的前10推薦電影並電影編號相應轉換成電影名
topKRecs.map(rating=>(titles(rating.product),rating.rating)).foreach(println)

moviesForUser.sortBy(-_.rating).take(10).map(rating=>(titles(rating.product),rating.rating)).foreach(println)

//物品推薦
//匯入jblas包建立向量
import org.jblas.DoubleMatrix
val aMatrix = new DoubleMatrix(Array(1.0,2.0,3.0))
//定義計算輸入量為向量的餘弦形式度公式
def consineSimilarity(vec1:DoubleMatrix,vec2:DoubleMatrix):Double={
    vec1.dot(vec2)/(vec1.norm2()*vec2.norm2())
}

val itemId = 567
val itemFactor=model.productFeatures.lookup(itemId).head
val itemVector = new DoubleMatrix(itemFactor)
consineSimilarity(itemVector,itemVector)
//計算各個物品的相似度
val sims = model.productFeatures.map{case(id,factor)=>
    val factorVector=new DoubleMatrix(factor)
    val sim = consineSimilarity(factorVector,itemVector)
    (id,sim)
     }
val K = 10
//找到相似度排名前10的
val sortedSims = sims.top(K)(Ordering.by[(Int,Double),Double]{case(id,similarity)=>similarity})
println(sortedSims.take(10).mkString("\n"))

println(titles(itemId))

val sortedSims2 = sims.top(K+1)(Ordering.by[(Int,Double),Double]{case(id,similarary)=>similarary})
sortedSims2.slice(1,11).map{case (id,sim)=>(titles(id),sim)}.mkString("\n")
//取出user和product
val temp = rating.map{case Rating(user,product,rate)=>(user,product)}


//推薦結果效果的評定
//MSE均方差
//對於某一特定使用者
val actualRating = moviesForUser.take(1)(0)
val predictedRating = model.predict(789,actualRating.product)
val squaredError = math.pow(actualRating.rating-predictedRating,2.0)
//對於全部使用者
val usersProducts = rating.map{case Rating(user,product,rating)=>(user,product)}
val predictions = model.predict(usersProducts).map{case Rating(user,product,rating)=>((user,product),rating)}
val ratingsAndPredictions = rating.map{case Rating(user,product,rating)=>((user,product),rating)}.join(predictions)
val MSE =ratingsAndPredictions.map{case((user,product),(actual,predicted))=>math.pow((actual-predicted),2)}.reduce(_+_)/ratingsAndPredictions.count
//均方根差
val RMSE = math.sqrt(MSE)

//直接呼叫Mllib內建函式計算RMSE和MSE
import org.apache.spark.mllib.evaluation.RegressionMetrics
val  predictedAndTrue = ratingsAndPredictions.map{case((user,product),(predicted,actual))=>(predicted,actual)}
val regressionMetrics = new RegressionMetrics(predictedAndTrue)
println("MSE:"+regressionMetrics.meanSquaredError)
println("RMSE:"+regressionMetrics.rootMeanSquaredError)

其他程式碼連線 構建評分矩陣