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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

和產(chǎn)品特征矩陣$ V(n*k) $第j個(gè)物品的特征向量$ 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ù)集，包含了多個(gè)用戶對(duì)多部電影的10萬次評(píng)級(jí)數(shù)據(jù)

下載地址

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

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

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: 對(duì)應(yīng)ALS模型中的因子個(gè)數(shù)，即矩陣分解出的兩個(gè)矩陣的新的行/列數(shù)，即$ A approx UV^T , k << m,n $m,n中的k

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

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

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

# 對(duì)用戶789預(yù)測其對(duì)電影123的評(píng)級(jí)
predictedRating = model.predict(789,123)
predictedRating

3.1740832151065774

# 獲取對(duì)用戶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)題的映射

然后獲取某個(gè)用戶評(píng)級(jí)前10的影片同推薦這個(gè)用戶的前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ù)測到的評(píng)級(jí)與真實(shí)評(píng)級(jí)的差值平方

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

對(duì)MSE取其平方根，即預(yù)計(jì)評(píng)級(jí)和實(shí)際評(píng)級(jí)的差值的標(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)置的評(píng)估函數(shù)計(jì)算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——人民郵電出版社