【博客存檔】Machine Learning With Spark Note 2：構建一個簡單的推薦系統

01-28

Matrix Factorization

MF在Netflix Prize中得到最好的名詞，關於MF的一片overview：http://techblog.netflix.com/2012/04/netflix-recommendations-beyond-5-stars.html。

Explicit matrix factorization

user ratings 數據：

Tom, Star Wars, 5

Jane, Titanic, 4 nBill, Batman, 3 nJane, Star Wars, 2 nBill, Titanic, 3 n以user為行，movie為列構造對應rating matrix：n

MF就是一種直接建模user-item矩陣的方法，利用兩個低維度的小矩陣的乘積來表示，屬於一種降維的技術。

如果我們有U個用戶，I個items，若不經過MF處理，它看來會使這樣的：

是一個極其稀疏的矩陣，經過MF處理後，表示為兩個維度較小的矩陣相乘：

這類模型被稱為latent feature models，旨在尋找那些潛在的特徵，來間接表示user-item rating的矩陣。這類潛在的features並不直接建模user對item的rating關係，而是通過latent features更趨近於建模用戶對某類items的偏好，例如某類影片、風格等等，而這些事通過MF尋找其內在的信息，無需items的詳細描述（和基於content的方法不同）。

MF模型如何計算一個user對某個item的偏好，對應向量相乘即可：

如何計算兩個item的相似度：

MF模型的好處是一旦模型創建好後，predict變得十分容易，並且性能也很好，但是在海量的用戶和itemset時，存儲和生產MF中的如上圖的這兩個矩陣會變得具有挑戰性。

Implicit matrix factorization

前面我們都在討論顯式的一些偏好信息，比如rating，但是在大部分應用中，拿不到這類信息，我們更多滴搜集的是一些隱性的反饋信息，這類反饋信息沒有明確地告訴某個用戶對某個item的偏好信息，但是卻可以從用戶對某個item的交互信息中建模出來，例如一些二值特徵，包括是否瀏覽過、是否購買過產品、以及多少次看過某部電影等等。

MLlib中提供了一種處理這類隱性特徵的方法，將前面的輸入ratings矩陣其實可以看做是兩個矩陣：二值偏好矩陣P和信心權重矩陣C；

舉個例子：假定我們的網站上面沒有設計對movie的rating部分，只能通過log查看到用戶是否觀看過影片，然後通過後期處理，可以看出他觀看到過多少次某部影片，這裡P來表示影片是否被某用戶看過，C來描述這裡的confidence weighting也就是觀看的次數：

這裡我們把P和C的dot product來替代前面的rating矩陣，那麼我們最終建模來預估某用戶對item的偏好

Alternating least squares

ALS是解決MF問題的一個優化技術，被證明高效、高性能並且能有效地並行化，目前為止，是MLlib中推薦模塊的唯一一個演算法。Spark官網上有專門地描述。

特徵提取

特徵提取是從已有數據中找到有用的數據來對演算法進行建模，本文中使用顯式數據也就是用戶對movie的rating信息，這個數據來源於網路上的MovieLens標準數據集，以下代碼為《Machine Learning with Spark》這本書裡面的python的重寫版本，會有專門的ipython notebook放到github上。

rawData = sc.textFile("../data/ML_spark/MovieLens/u.data")nprint rawData.first()nrawRatings = rawData.map(lambda x: x.split(t))nrawRatings.take(5)n

數據分別是userId，itemId，rating和timestamp。

from pyspark.mllib.recommendation import Ratingnfrom pyspark.mllib.recommendation import ALSnratings = rawRatings.map(lambda x : Rating(int(x[0]),int(x[1]),float(x[2])))nprint ratings.first()n

格式化數據，用於後面建模數據，導入Rating，ALS模塊，下面是ALS類的使用說明：

其中rank就是上面latent feature model中矩陣的k，在下面的實驗中，我們設為50：

model = ALS.train(ratings,50)n# modelImplicit = ALS.(ratings,50,alpha=0.02)nuserFeatures = model.userFeatures()nprint userFeatures.take(2)n

這裡user1與user2，均用50維的向量來表示，也就是上面U*k那個矩陣的每個向量

predictRating = model.predict(789,123)nprint predictRatingn

預測用戶789對item 123的rating值，結果為3.76599662082。

topKRecs = model.recommendProducts(userId,K)nfor rec in topKRecs:n print recnmoviesForUser = ratings.groupBy(lambda x : x.user).mapValues(list).lookup(userId)n# print moviesForUsernfor i in sorted(moviesForUser[0],key=lambda x : x.rating,reverse=True):n print i.productn# forn# print moviesForUsern

使用recommendProducts來為用戶推薦top10的items，其items順序為降序。MoviesForUser是從ratings數據中找出的用戶789rating最高的數據，仔細看下發現數據和我們的ratings裡面找出的數據貌似一個都沒有相同的，那麼是不是說明我們的演算法不給力呢？！這個可不一定，想想看，如果推薦系統只是推薦給你看過的電影，那麼它一定是一個失敗的，並且完全對系統的kpi數據無提升作用，前面提到，MF的實質是通過latent feature去找到與用戶過去偏好高的有某些隱性相同特徵的電影（這些由整體用戶的集體智慧得到），比如可能是某一類型的電影、又或者相同的演員等等，所以這裡不能說明推薦系統不給力，但是確實也很難具有解釋性。

Item recommendations

基於MF的方法中，我們可以利用之前看到k*I的矩陣，計算兩個向量質檢的相似性，也就是item的相似性。這樣，可以很容易做相似商品推薦的場景。這裡我們定義相似函數為餘弦相似性：

import numpy as npndef cosineSImilarity(x,y):n return np.dot(x,y)/(np.linalg.norm(x)*np.linalg.norm(y))ntestx = np.array([1.0,2.0,3.0])nprint cosineSImilarity(testx,testx)n

然後，通過ALS建模的item的向量，拿到對應地item的向量表示：

itemId = 567nitemFactor = model.productFeatures().lookup(itemId)[0]n# itemFactor = itemFactor[1]nprint itemFactorn# model.productFeatures().collect()nsims = model.productFeatures().map(lambda (id,factor):(id,cosineSImilarity(np.array(factor),n np.array(itemFactor))))nsims.sortBy(lambda (x,y):y,ascending=False).take(10)n

利用ALS的item向量拿到itemId為567的向量表示，然後對model的item的特徵向量來計算與567的相似度，按降序排序並取top10

這樣，可以找到與567這個item相似性最大的itemlist。

如何衡量推薦系統的性能

怎麼判斷我們生成的模型性能呢？常用的有一些比如Mean Squared Error，Root Mean Squared Error，但是這類標準無法考量推薦最終的items的排序問題，在實際工作中用的比較多的是Mean Average Precision，考慮到了item的排序造成的影響。

MSE&RMSE：

userProducts = ratings.map(lambda rating:(rating.user,rating.product))nprint userProducts.take(1)[0]npredictions = model.predictAll(userProducts).map(lambda rating:((rating.user,rating.product)n ,rating.rating))nprint predictions.take(5)nratingsAndPredictions = ratings.map(lambda rating:((rating.user,rating.product),rating.rating))n .join(predictions)nMSE = ratingsAndPredictions.map(lambda ((x,y),(m,n)):math.pow(m-n,2)).reduce(lambda x,y:x+y)/ratingsAndPredictions.count() print MSE print math.sqrt(MSE)n

先map ratings數據得到用戶對item的組合，然後對這類數據predictAll計算該用戶對item的rating估計值。然後利用join函數將預測的數據與ratings中的數據」聯合」起來，塞入相似度函數進行計算,最終結果如下：

備註：看到這裡肯定有人會問題，你之前在前面recommendProducts的，沒有一個item是與ratings的數據相同，但是這裡為什麼又對比ratings中的評分信息來衡量推薦模型的好壞呢。猜想：recommendProduct是基於最終預測的ratings的高低來推薦的，但是，考慮到前面分析的原因，應該是不僅僅是按predict的rating的高低來給定推薦產品而是參入了其他的考量，所以這裡並不矛盾。

APK：

什麼是APK？可以看下這裡，裡面有R，Matlab，Python的各種Metrics的實現，還有kaggle里對APK的說明，邏輯很簡單，相對於MSE和RMSE，考慮了推薦的排序對最後metrics的影響，如果檢索出來的item排序越靠前，得分越高。

def avgPrecisionK(actual, predicted,k=10):n if len(predicted)>k:n predicted = predicted[:k]nn score = 0.0n num_hits = 0.0n for i,p in enumerate(predicted):n if p in actual and p not in predicted[:i]:n num_hits += 1.0n score += num_hits / (i+1.0)nn if not actual:n return 1.0nn return score / min(len(actual), k)nitemFactors = model.productFeatures().map(lambda (id,factor):factor).collect()nitemMatrix = np.array(itemFactors)nimBroadcast = sc.broadcast(itemMatrix)n

拿到product的所有向量表示，初始化矩陣，然後broadcast到各個節點。

userVector = model.userFeatures().map(lambda (userId,array):(userId,np.array(array)))n# print userVector[0]nuserVector = userVector.map(lambda (userId,x):n (userId,imBroadcast.value.dot((np.array(x).transpose()))))nuserVectorId = userVector.map(lambda (userId,x) : (userId,[(xx,i) for i,xx in enumerate(x.tolist())]))nsortUserVectorId = userVectorId.map(lambda (userId,x):(userId,sorted(x,key=lambda x:x[0],reverse=True)))nsortUserVectorRecId = sortUserVectorId.map(lambda (userId,x): (userId,[xx[1] for xx in x]))n

為每一個user推薦一個對應的item list，並按user向量與item向量相乘計算的該用戶對該item的rating值來進行排序，最終給定一個有序的item的list。

userMovies = ratings.map(lambda rating: (rating.user,rating.product)).groupBy(lambda (x,y):x)nuserMovies = userMovies.map(lambda (userId,x):(userId, [xx[1] for xx in x] ))nallAPK=sortUserVectorRecId.join(userMovies).map(lambda (userId,(predicted, actual))n :avgPrecisionK(actual,predicted,2000))nprint allAPK.reduce(lambda x,y:x+y)/allAPK.count()n

然後從rating中找到對應的的item 列表，然後塞入之前我們寫的apk函數，然後求平均，最終結果為0.115484271925。

當然我們可以直接使用MLlib內置的evaluation模塊來對我們的模型進行評價，如MSE，RMSE：

from pyspark.mllib.evaluation import RegressionMetricsnfrom pyspark.mllib.evaluation import RankingMetricsnpredictedAndTrue = ratingsAndPredictions.map(lambda ((userId,product),(predicted, actual))n :(predicted,actual))n# print predictedAndTrue.take(1)nregressionMetrics = RegressionMetrics(predictedAndTrue)nprint "Mean Squared Error = %f"%regressionMetrics.meanSquaredErrornprint "Root Mean Squared Error %f"% regressionMetrics.rootMeanSquaredErrorn

MAP：

#MAPn# The implementation of the average precision at the K function in RankingMetrics is slightly differentn# from ours,n# so we will get different results. However, the computation of the overall mean average precisionn#(MAP, which does not use a threshold at K) is the same as our function if we select K to be very highn# (say, at least as high as the number of items in our item set)nsortedLabels = sortUserVectorRecId.join(userMovies).map(lambda (userId,(predicted, actual))n :(predicted,actual))n# print sortedLabels.take(1)nrankMetrics = RankingMetrics(sortedLabels)nprint "Mean Average Precision = %f" % rankMetrics.meanAveragePrecisionnprint "Mean Average Precision(at K=10) = %f" % rankMetrics.precisionAt(5)n

這裡結果與我們前面取k=2000的結果相同，說明我們的計算和MLlib是一致的，但是K=10或者比較小的值時，不一樣，這是因為MLlib在precisionAt(k)這個函數與我們前面邏輯不同，這裡我們不做考慮。

本章的代碼放到了github上面，是ipython notebook的可以直接調用試用下，這版代碼是我學習spark寫的，水平很差，而且notebook中也沒有基本的代碼說明，算是對原書中這部分的scala的一次重寫，喜歡python和spark的可以研究下，一步一步看下還是會熟悉python操作spark的流程的