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

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

在學(xué)習(xí)Spark機(jī)器學(xué)習(xí)這本書時，書上用scala完成，自己不熟悉遂用pyshark完成，更深入的理解了spark對協(xié)同過濾的實(shí)現(xiàn)

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

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

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

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

這里我們要處理的數(shù)據(jù)是用戶提供的自身偏好數(shù)據(jù)，即用戶對物品的打分?jǐn)?shù)據(jù)。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

得出矩陣分解模型的損失函數(shù)如下

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

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

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

下載地址

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

通過sc.textFile()讀取數(shù)據(jù)為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"]

# 導(dǎo)入spark中的mllib的推薦庫
import pyspark.mllib.recommendation as rd

生成Rating類的RDD數(shù)據(jù)

# 由于ALS模型需要由Rating記錄構(gòu)成的RDD作為參數(shù)，因此這里用rd.Rating方法封裝數(shù)據(jù)
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: 對應(yīng)ALS模型中的因子個數(shù)，即矩陣分解出的兩個矩陣的新的行/列數(shù)，即$ A approx UV^T , k << m,n $m,n中的k

iterations: 對應(yīng)運(yùn)行時的最大迭代次數(shù)

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

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

# 對用戶789預(yù)測其對電影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)]

這里首先將電影的數(shù)據(jù)讀入，講數(shù)據(jù)處理為電影ID到標(biāo)題的映射

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

#檢查推薦內(nèi)容
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)]

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

直接度量模型的預(yù)測目標(biāo)變量的好壞

對MSE取其平方根，即預(yù)計評級和實(shí)際評級的差值的標(biāo)準(zhǔn)差

# 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內(nèi)置的評估函數(shù)計算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矩陣分解的協(xié)同過濾算法與源碼分析

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

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