摘要：協同過濾提出了一種支持不完整評分矩陣的矩陣分解方法不用對評分矩陣進行估值填充。使用的是交叉最小二乘法來最優化損失函數。

推薦引擎背后的想法是預測人們可能喜好的物品并通過探尋物品之間的聯系來輔助這個過程

在學習Spark機器學習這本書時，書上用scala完成，自己不熟悉遂用pyshark完成，更深入的理解了spark對協同過濾的實現

在這里我們的推薦模型選用協同過濾這種類型，使用Spark的MLlib中推薦模型庫中基于矩陣分解（matrix factorization）的實現。

協同過濾簡單來說是利用某興趣相投、擁有共同經驗之群體的喜好來推薦用戶感興趣的信息，個人通過合作的機制給予信息相當程度的回應（如評分）并記錄下來以達到過濾的目的進而幫助別人篩選信息，回應不一定局限于特別感興趣的，特別不感興趣信息的紀錄也相當重要。

很簡單的例子來介紹就是日常我們生活中經常找電影會通過向和自己品味類似的朋友要求推薦，這就是協同過濾的思想

基于用戶或物品的方法的得分取決于若干用戶或是物品之間依據相似度所構成的集合（即鄰居），故它們也常被稱為最鄰近模型。

這里我們要處理的數據是用戶提供的自身偏好數據，即用戶對物品的打分數據。

這些數據可以被轉換成用戶為行，物品為列的二維矩陣，即評分矩陣A（m*n）表示m個用戶對n個物品的打分情況

這個矩陣A很多元素都是空的，我們稱其為缺失值（missing value）。

協同過濾提出了一種支持不完整評分矩陣的矩陣分解方法,不用對評分矩陣進行估值填充。

在推薦系統中，我們希望得到用戶對所有物品的打分情況，如果用戶沒有對一個物品打分，那么就需要預測用戶是否會對該物品打分，以及會打多少分。這就是所謂的矩陣補全（矩陣分解）

對于這個矩陣分解的方式就是找出兩個低維度的矩陣，使得他們的乘積是原始的矩陣。

因此這也是一種降維技術。要找到和‘用戶-物品’矩陣近似的k維矩陣，最終要求得出表示用戶的m x k維矩陣，和一個表示物品的k x n維矩陣。
這兩個矩陣也稱作因子矩陣。

對于k的理解為對于每個產品，這里已電影為例，可以從k個角度進行評價，即k個特征值

由于是對‘用戶-物品’矩陣直接建模，用這些模型進行預測也相對直接，要計算給定用戶對某個物品的預計評級，就從用戶因子矩陣和物品因子矩陣分別選取對應的行與列，然后計算兩者的點積。

假設對于用戶A，該用戶對一部電影的綜合評分和電影的特征值存在一定的線性關系，

即電影的綜合評分=(a1d1+a2d2+a3d3+a4d4)

其中a1-4為用戶A的特征值，d1-4為之前所說的電影的特征值

最小二乘法（Alternating Least Squares, ALS）是一種求解矩陣分解問題的最優化方法。它功能強大、效果理想而且被證明相對容易并行化。這使得它很適合如Spark這樣的平臺。

使用用戶特征矩陣$ U(m*k) $ 中的第i個用戶的特征向量$ u_i $ ,

和產品特征矩陣$ V(n*k) $第j個物品的特征向量$ v_i $來預測打分矩陣$ A(m*n) $中的$ a_{ij} $，

得出矩陣分解模型的損失函數如下

$$ large C = sumlimits_{(i,j)in R}[(a_{ij} - u_iv_j^T)^2+lambda(u_i^2+v_j^2)] $$

通常的優化方法分為兩種：交叉最小二乘法（alternative least squares）和隨機梯度下降法（stochastic gradient descent）。Spark使用的是交叉最小二乘法（ALS）來最優化損失函數。
算法的思想就是：我們先隨機生成然后固定它求解，再固定求解，這樣交替進行下去，直到取得最優解$ min(C) $

我們這里的數據集是Movielens 100k數據集，包含了多個用戶對多部電影的10萬次評級數據

下載地址

讀取評級數據集，該數據包括用戶ID，影片ID，星級和時間戳等字段，使用/t分隔

通過sc.textFile()讀取數據為RDD,通過分隔符取前3個屬性分別為用戶ID，影片ID，星級

rawData = sc.textFile("/home/null/hadoop/data/ml-100k/u.data")
rawData.first()
type(rawData)

pyspark.rdd.RDD

rawRatings = rawData.map(lambda line: line.split("	")[:3])
rawRatings.first()

["196", "242", "3"]

# 導入spark中的mllib的推薦庫
import pyspark.mllib.recommendation as rd

生成Rating類的RDD數據

# 由于ALS模型需要由Rating記錄構成的RDD作為參數，因此這里用rd.Rating方法封裝數據
ratings = rawRatings.map(lambda line: rd.Rating(int(line[0]), int(line[1]), float(line[2])))
ratings.first()

Rating(user=196, product=242, rating=3.0)

rank: 對應ALS模型中的因子個數，即矩陣分解出的兩個矩陣的新的行/列數，即$ A approx UV^T , k << m,n $m,n中的k

iterations: 對應運行時的最大迭代次數

lambda: 控制模型的正則化過程，從而控制模型的過擬合情況。

# 訓練ALS模型
model = rd.ALS.train(ratings, 50, 10, 0.01)
model

# 對用戶789預測其對電影123的評級
predictedRating = model.predict(789,123)
predictedRating

3.1740832151065774

# 獲取對用戶789的前10推薦
topKRecs = model.recommendProducts(789,10)
topKRecs

[Rating(user=789, product=429, rating=6.302989890089658),
 Rating(user=789, product=496, rating=5.782039583864358),
 Rating(user=789, product=651, rating=5.665266358968961),
 Rating(user=789, product=250, rating=5.551256887914674),
 Rating(user=789, product=64, rating=5.5336980239740186),
 Rating(user=789, product=603, rating=5.468600343790217),
 Rating(user=789, product=317, rating=5.442052952711695),
 Rating(user=789, product=480, rating=5.414042111530209),
 Rating(user=789, product=180, rating=5.413309515550101),
 Rating(user=789, product=443, rating=5.400024900653429)]

這里首先將電影的數據讀入，講數據處理為電影ID到標題的映射

然后獲取某個用戶評級前10的影片同推薦這個用戶的前10影片進行比較

#檢查推薦內容
movies = sc.textFile("/home/null/hadoop/data/ml-100k/u.item")
movies.first()

"1|Toy Story (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Toy%20Story%20(1995)|0|0|0|1|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0"

titles_data= movies.map(lambda line: line.split("|")[:2]).collect()
titles = dict(titles_data)
titles

moviesForUser = ratings.keyBy(lambda rating: rating.user).lookup(789)
type(moviesForUser)

list

moviesForUser = sorted(moviesForUser,key=lambda r: r.rating, reverse=True)[0:10]
moviesForUser

[Rating(user=789, product=127, rating=5.0),
 Rating(user=789, product=475, rating=5.0),
 Rating(user=789, product=9, rating=5.0),
 Rating(user=789, product=50, rating=5.0),
 Rating(user=789, product=150, rating=5.0),
 Rating(user=789, product=276, rating=5.0),
 Rating(user=789, product=129, rating=5.0),
 Rating(user=789, product=100, rating=5.0),
 Rating(user=789, product=741, rating=5.0),
 Rating(user=789, product=1012, rating=4.0)]

[(titles[str(r.product)], r.rating) for r in moviesForUser]

[("Godfather, The (1972)", 5.0),
 ("Trainspotting (1996)", 5.0),
 ("Dead Man Walking (1995)", 5.0),
 ("Star Wars (1977)", 5.0),
 ("Swingers (1996)", 5.0),
 ("Leaving Las Vegas (1995)", 5.0),
 ("Bound (1996)", 5.0),
 ("Fargo (1996)", 5.0),
 ("Last Supper, The (1995)", 5.0),
 ("Private Parts (1997)", 4.0)]

[(titles[str(r.product)], r.rating) for r in topKRecs]

[("Day the Earth Stood Still, The (1951)", 6.302989890089658),
 ("It"s a Wonderful Life (1946)", 5.782039583864358),
 ("Glory (1989)", 5.665266358968961),
 ("Fifth Element, The (1997)", 5.551256887914674),
 ("Shawshank Redemption, The (1994)", 5.5336980239740186),
 ("Rear Window (1954)", 5.468600343790217),
 ("In the Name of the Father (1993)", 5.442052952711695),
 ("North by Northwest (1959)", 5.414042111530209),
 ("Apocalypse Now (1979)", 5.413309515550101),
 ("Birds, The (1963)", 5.400024900653429)]

定義為各平方誤差的和與總數目的商，其中平方誤差是指預測到的評級與真實評級的差值平方

直接度量模型的預測目標變量的好壞

對MSE取其平方根，即預計評級和實際評級的差值的標準差

# evaluation metric
usersProducts = ratings.map(lambda r:(r.user, r.product))
predictions = model.predictAll(usersProducts).map(lambda r: ((r.user, r.product),r.rating))
predictions.first()

((316, 1084), 4.006135662882842)

ratingsAndPredictions = ratings.map(lambda r: ((r.user,r.product), r.rating)).join(predictions)
ratingsAndPredictions.first()

((186, 302), (3.0, 2.7544572973050236))

# 使用MLlib內置的評估函數計算MSE,RMSE
from pyspark.mllib.evaluation import RegressionMetrics
predictionsAndTrue = ratingsAndPredictions.map(lambda line: (line[1][0],line[1][3]))
predictionsAndTrue.first()

(3.0, 2.7544572973050236)

# MSE
regressionMetrics = RegressionMetrics(predictionsAndTrue)
regressionMetrics.meanSquaredError

0.08509832708963357

# RMSE
regressionMetrics.rootMeanSquaredError

0.2917161755707653

參考：

深入理解Spark ML：基于ALS矩陣分解的協同過濾算法與源碼分析

如何解釋spark mllib中ALS算法的原理？——知乎

Maching Learning With Spark——人民郵電出版社