基于 MovieLens 数据的 Spark ALS 实例

Question

MovieLens 100k 数据集包含表示多个用户对多部电影的 10 万次评级数据，也包含电影元数据和用户属性信息。该数据集不大，方便下载和用 spark 程序快速处理，故适合做讲解实例。可从 http://files.grouplens.org/datasets/movielens/ml-100k.zip 下载这个数据集。

本章节分为以下几个部分：

1. 一、提取有效特征
2. 二、训练推荐模型
3. 三、使用推荐模型
4. 四、推荐模型效果的评估​

一 提取有效特征

这里，我们将采用显示评级数据，这样所需的输入数据就只需包含每个评级对应的用户 ID、影片 ID 和具体的星级。

从 MovieLens 100k 数据集提取特征

该数据由用户 ID，影片 ID，星级和时间戳依次组成

val PATH = "file:///Users/lzz/work/SparkML/"
val rawData = sc.textFile( PATH + "data/ml-100k/u.data")
rawData.first()

196    242 3   881250949

各个纪录用 \t 分割，这会返回一个 Array[String] 数组。我们需要前面的三个字段（用户 ID，影片 ID，星级）时间戳不需要，所以提取前三个字段即可

val rawRatings = rawData.map(_.split('\t').take(3))
rawRatings.first()

Array(196, 242, 3)

下面使用 spark MLlib 来训练模型。先看一下有哪些可用模型及它们的输入如何。首先，从 MLlib 导入 ALS 模型：

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

Name: Compile Error
Message: <console>:18: error: ambiguous reference to overloaded definition,
both method train in object ALS of type (ratings: org.apache.spark.rdd.RDD[org.apache.spark.mllib.recommendation.Rating], rank: Int, iterations: Int)org.apache.spark.mllib.recommendation.MatrixFactorizationModel
and  method train in object ALS of type (ratings: org.apache.spark.rdd.RDD[org.apache.spark.mllib.recommendation.Rating], rank: Int, iterations: Int, lambda: Double)org.apache.spark.mllib.recommendation.MatrixFactorizationModel
match expected type ?
              ALS.train
                  ^
StackTrace: 

说明需要 ratings,rank(是在模型中的潜在因素的数量),iterations(迭代是运行的迭代次数),lambda(λ指定 ALS 正则化参数)

import org.apache.spark.mllib.recommendation.Rating
Rating()

Name: Compile Error
Message: <console>:15: error: not enough arguments for method apply: (user: Int, product: Int, rating: Double)org.apache.spark.mllib.recommendation.Rating in object Rating.
Unspecified value parameters user, product, rating.
              Rating()
                    ^
StackTrace: 

说明这个函数需要 user: Int, product: Int, rating: Double 这三个参数

val ratings = rawRatings.map{ case Array(user, movie, rating) => Rating(user.toInt,movie.toInt,rating.toDouble)}
ratings.first()

Rating(196,242,3.0)

二 训练推荐模型

从原始数据提取这些简单特征后，便可训练模型。MLlib 已实现模型训练的细节，这不需要我们担心。我们只需提供上述的指定类型的新 RDD 以及其他所需参数来作为训练的输入即可。

使用 MovieLens 100k 数据集训练模型

现在可用开始训练模型来，所需要的其他参数有以下几个。

• rank:对应 ALS 模型中的因子个数，也就是中低阶近似矩阵的隐含特征个数。因子个数一般越多越好。但它也会直接影响训练模型和保存时所需要的开销，尤其是在用户和物品很多的时候。因此实践中该参数常作为训练效果与系统开销之间的调节参数。通常，其合理取值为 10 到 200.
• iterations：对应运行时到迭代次数。ALS 能确保每次迭代都能降低评级矩阵的重建误差，但一把经少数次迭代后 ALS 模型便已能收敛为一个比较合理的好模型。这样，大部分情况下都没必要迭代太多次（10 次左右一般就挺好）。
• lambda：该参数控制模型的正则化过程，从而控制模型的过拟合情况。其值越高，正则化越严厉。该参数的赋值以实际数据的大小，特征和稀疏程度有关。和其他的机器学习模型一样，正则化参数应该通过用非样本的测试数据进行交叉验证来调整。 作为示列，这里将使用的 rank.iterations,lambda 参数的值分别为 50，10 和 0.01

val model = ALS.train(ratings, 50, 10, 0.01) // 返回 MatrixFactorizationModel 类型
println(model.userFeatures.count)
println(model.productFeatures.count)

943
1682

三 使用推荐模型

有了训练好的模型后便可用它来做预测。预测通常有两种：为某个用户推荐物品，或找出与某个物品相关或相似的物品。

3.1、用户推荐

用户推荐时指向给定用户推荐物品。它通常以 前 K 个 形式展现，即通过模型求出用户可能喜好程度最高的前 K 个商品。这个过程通过计算每个商品的预计得分并按照得分进行排序实现。 具体实现方法取决于所采用的模型。

比如若采用基于用户的模型，则会利用相似用户的评级来计算对某个用户的推荐。

利用矩阵分解方法时，是直接对评级数据进行建模，所以预计得分可视作相应用户因子向量和物品因子向量的点积。

从 MovieLens 100k 数据集生成电影推荐

MLlib 的推荐模型基于矩阵分解，因此可用模型所求得的因子矩阵来计算用户对物品的预计评级。下面只针对利用 MovieLens 中显式数据做推荐的情形，使用隐式模型时的方法与之相同。
MatrixFactorizationModel 类提供了一个 predict 函数，以方便地计算给定用户对给定物品的预期得分：

val predictedRating = model.predict(789,123)
predictedRating

4.386848715581662

该模型预测用户 789 对电影 123 的评分为 3.6882841660770698
用户 789 推荐的 10 个物品及对应的评分

val userId = 789
val K = 10
val topKPecs = model.recommendProducts(userId,K)
println(topKPecs.mkString("\n"))

Rating(789,48,5.949919087906642)
Rating(789,496,5.77835924855985)
Rating(789,32,5.707556027654178)
Rating(789,192,5.4583952405415115)
Rating(789,180,5.438986145011019)
Rating(789,136,5.405035211990188)
Rating(789,47,5.375404388444865)
Rating(789,64,5.224700595759688)
Rating(789,686,5.161532434245572)
Rating(789,302,5.118418747903516)

检验推荐的内容

为列直观的检验推荐的效果，可以简单对比下用户所评级过的电影的标题和被推荐的那些电影的电影名。首先，我们需要读入电影数据（这是在上一章探索过的数据集）。这些数据会导入为 Map[Int, String]类型，即从电影 ID 到标题到映射：

val movies = sc.textFile("../data/ml-100k/u.item")
val titles = movies.map(line => line.split("\\|").take(2)).map( array => ( array(0).toInt, array(1) ) ).collectAsMap()
titles(123)

Frighteners, The (1996)

对用户 789，我们可以找出他所接触过的电影，给出最高评级的前 10 部电影及名称。具体实现时，可先用 spark 的 keyBY 函数从 ratings RDD 来创建一个键值对 RDD。其主键为用户 ID。然后利用 lookup 函数只返回给定的键值（即特定的用户 ID）对应的那些评级数据到驱动程序。

val moviesForUser = ratings.keyBy(_.user).lookup(789)
println(moviesForUser.size)

33

可以看到 789 这个用户，对 33 部电影做过评级
接下来，我们要获取评级最高的前 10 部电影，具体做法是利用 Rating 对象的 rating 属性来对 moviesForUser 集合进行排序并选出排名前 10 对评级（含相应电影 ID）。之后以其为输入，借助 titles 映射为“（电影名称， 具体评级）”形式。再将名称与具体评级打印出来

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

(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)

现在看下对该用户对前 10 个推荐，并利用上述相同的方式来查看他们的电影名（注意这些推荐已排序）

topKPecs.map( rating => ( titles(rating.product), rating.rating ) ).foreach(println)

(Hoop Dreams (1994),5.949919087906642)
(It's a Wonderful Life (1946),5.77835924855985)
(Crumb (1994),5.707556027654178)
(Raging Bull (1980),5.4583952405415115)
(Apocalypse Now (1979),5.438986145011019)
(Mr. Smith Goes to Washington (1939),5.405035211990188)
(Ed Wood (1994),5.375404388444865)
(Shawshank Redemption, The (1994),5.224700595759688)
(Perfect World, A (1993),5.161532434245572)
(L.A. Confidential (1997),5.118418747903516)

3.2、物品推荐

物品推荐是为了回答如下问题：给定一个物品，有哪些物品与它最相似？这里，相似的明确定于取决于所使用的模型。大多数情况下，相似度是通过某种方式比较表示两个物品的向量而得到的。常见的相似度衡量方法包括皮尔森相关系数（Pearson correlation），针对实现向量的余弦相似度（cosine similarity） 和针对二元向量的杰卡德相似系数（Jaccard similarity）。

从 MovieLens 100k 数据集生成相似电影

MatrixFactorizationModel 当前的 API 不能直接支持物品之间相似度的计算。所以我们要自己实现。
这里会使用余弦相似度来衡量相似度。另外采用 jblas 线性代数库（MLlib 依赖库之一）来求向量点积。这些和现有的 predict 和 recommendProducts 函数的实现方式类似，但我们会用到余弦相似度而不仅仅是求点积。
我们想利用余弦相似度来对指定物品的因子向量与其他物品做比较。进行线性计算时，除了因子向量外，还需要创建一个 Array[Double]类型的向量对象。以该类型对象为构造函数的输入来创建一个 jblas.DoubleMatrix 类型对象的方法如下：

import org.jblas.DoubleMatrix
val aMatrix = new DoubleMatrix( Array(1.0,2.0,3.0) )
aMatrix

[1.000000; 2.000000; 3.000000]

我们需要定义一个函数来计算两个向量之间的余弦相似度。余弦相似度时两个向量做 n 维空间里两者夹角的度数。它是两个向量的点积与各向量范式（或长度）的乘积的商。（余弦相似度的范数为 L2-范数，L2-norm）这样，余弦相似度时一个正则化了的点积。

该相似度的取值在－1 到 1 之间。1 表示完全相似，0 表示两者互不相关（即无相似性）。这种衡量方法很有帮助，因为它还能捕捉负相关性。也就是说，当为－1 时则不仅表示两者不相关，还表示它们完全不同。

下面来创建这个 cosinesSimilarity 函数：

def cosinesSimilarity(vec1: DoubleMatrix, vec2: DoubleMatrix): Double = {vec1.dot(vec2)/(vec1.norm2()*vec2.norm2())}

下面以物品 567 为例从模型中取回其对应的因子。这可以通过调用 lookup 函数来实现。之前曾用过该函数来取回特定用户的评级信息。下面的代码中还使用了 head 函数。lookup 函数返回了一个数组而我们只需要第一个值（实际上，数组里也只会有一个值，也就是改物品的因子向量）。

这个因子的类型为 Array[Double],所以后面会用它来创建一个 Double[Matrix]对象，然后再用该对象来计算它与自己的相似度：

val itemId = 587
val itemFactor = model.productFeatures.lookup(itemId).head
val itemVector = new DoubleMatrix(itemFactor)
cosinesSimilarity(itemVector,itemVector)

1.0000000000000002

现在求各个物品的余弦相似度：

val sims = model.productFeatures.map{
    case(id, factor) => {
        val factorVector = new DoubleMatrix(factor)
        val sim = cosinesSimilarity(factorVector,itemVector)
        (id,sim)
    }
}

接下来，对物品按照相似度排序，然后取出与物品 567 相似的前 10 个物品：

val sortedSims = sims.top(K)(Ordering.by[(Int,Double),Double]{case(id,similarity)=>similarity})

上述代码里使用了 Spark 的 top 函数。相比使用 collect 函数将结果返回驱动程序然后再本地排序，它能分布式计算出“前 K 个”结果，因而更高效。（注意，推荐系统要处理的用户和物品数目可能数以百万计）
spark 需要知道如何对 sims RDD 里的（item　id,similarity socre） 对排序。为此，我们另外传人里一个参数给 top 函数。这个参数是一个 Scala Ordering 对象，它会告诉 spark 根据键值对里的值排序（也就是用 similarity 排序）。

println(sortedSims.take(10).mkString("\n"))

(587,1.0000000000000002)
(203,0.8384428799486454)
(498,0.8359235275715651)
(318,0.8292616548066134)
(357,0.8253684931815446)
(98,0.8210704299306487)
(216,0.8190782220833737)
(603,0.8168705070642182)
(659,0.8166916141179084)
(501,0.8165426793671007)

排名第一的最相似的物品就是我们给定的物品。之后便是以相似度排序的其他类似物品。

检查推荐的相似物品

来看一下我们所给定的电影的名称是什么：

println(titles(itemId))

Hour of the Pig, The (1993)

如在用户推荐中所做过的，我们可以看看推荐的那些电影名称是什么，从而直观上检查一下基于物品推荐的结果。这一次我们取前 11 部最相似电影，以排除给定的那部。所以，可以选取列表中的第 1 列到 11 项：

val sortedSim2 = sims.top(K+1)(Ordering.by[(Int,Double), Double]{case(id, similarity) => similarity})
sortedSim2.slice(1,11).map{ case(id, sim)=>(titles(id),sim)}.mkString("\n")

(Unforgiven (1992),0.8384428799486454)
(African Queen, The (1951),0.8359235275715651)
(Schindler's List (1993),0.8292616548066134)
(One Flew Over the Cuckoo's Nest (1975),0.8253684931815446)
(Silence of the Lambs, The (1991),0.8210704299306487)
(When Harry Met Sally... (1989),0.8190782220833737)
(Rear Window (1954),0.8168705070642182)
(Arsenic and Old Lace (1944),0.8166916141179084)
(Dumbo (1941),0.8165426793671007)
(Giant (1956),0.8154788688566539)

四、模型效果的评估

如何知道训练出来的模型是一个好模型？这就需要某种方式来评估它的预测效果。评估指标（evaluationg metric）指那些衡量模型预测能力或准确度的方法。它们有些直接度量模型的预测目标变量的好坏（比如均方差），有些则关注模型对那些其并未针对性的优化过但又十分接近真实应用场景数据的预测能力（比如平均准确率）。
评估指标提供了同一模型中不同参数下，又或是不同模型之间进行比较的标准方法。

通过这些指标，人们可以从待选的模型中找出表现最好的那个模型。

这里将会演示如何计算推荐系统和协同过滤模型里常用的两个指标：均方差以及 K 值平均准确率。

均方差

均方差（Mean　Squared Error,MSE）直接衡量“用户－物品” 评级矩阵的重建误差。它也是一些模型里所采用的最小化目标函数，特别是许多矩阵分解类方法，比如 ALS，因此，它常用于显示评级的情形。

它的定义为各平方误差的和与总数目的商。其中平房误差是指预测到的评级与真实评级的差值的平方。

下面以用户 789 为列做讲解。现在从之前计算的 moviesForUser 这个 Ratings 集合里找出该用户的第一个评级：

val actualRating = moviesForUser.take(1)(0)
actualRating

Rating(789,1012,4.0)

可以看到该用户对该电影的评级为 4，然后，求模型预计评级

val predictedRating = model.predict(789, actualRating.product)
predictedRating

4.066744111154669

可以看出来预测的评级差不多也是 4，十分接近用户真实的实际评级。最后，我们计算实际评级和预计评级的平方误差：

val squaredError = math.pow(predictedRating - actualRating.rating, 2.0)
squaredError

0.004454776373826862

要计算整个数据集赏的 MSE，需要对每一条（user,movie actual rating,predicted rating）记录都计算该平方误差，然后求和，再除以总的评级次数。具体实现如下：
首先从 ratings RDD 里提取用户和物品的 ID，并使用 model.predict 来对各个“用户－物品”对做预测。所得的 RDD 以“用户和物品 ID”对作为主键，对应的预计评级为值：

val userProducts = ratings.map{case Rating(user,product,rating)=>(user,product)}
val predictions = model.predict(userProducts).map{case Rating(user,product,rating)=>((user,product),rating)}

接着提取出真实的评级。同时，对 ratings RDD 做映射以让“用户－物品”对为主键，实际评级为对应的值。这样，就得到了两个主键组成相同的 RDD。将两者连接起来，以创建一个新的 RDD。这个 RDD 的主键“用户－物品”对，键值为相应的实际评级和预计评级。

val ratingsAndPredictions = ratings.map{case Rating(user,product,rating)=>((user,product),rating)}.join(predictions)
ratingsAndPredictions

MapPartitionsRDD[237] at join at <console>:29

最后，求上述 MSE。具体先用 reduce 来对平方误差求和，然后再除以 count 函数所求得对总记录数：

val MSE = ratingsAndPredictions.map{
    case((user,product),(actual,predicted)) => math.pow((actual - predicted),2)
}.reduce(_+_) / ratingsAndPredictions.count
println("Mean Squared Error = " + MSE)

Mean Squared Error = 0.08505159175058329

均方跟误差（Root Mean Squared Error,RMSE）对使用也很普遍，其计算只需在 MSE 上取平方根即可。这不难理解，因为两者背后使用的数据（即评级数据）相同。它等同于求预计评级和实际评级等差值的标准差。如下代码便可求出：

val RMSE = math.sqrt(MSE)
println("Root Mean Squared Error=" + RMSE)

Root Mean Squared Error=0.29163606044277735

K 值平均准确率

K 值平均准确率（MAKP）的意思是整个数据集上的 K 值平均准确率（Average Precision at K metric, APK）的均值。APK 是信息检索中常用的一个指标。它用于衡量针对某个查询所返回的“前 K 个”文档的平均相关性。对于每次查询，我们会将结果中前 K 个与实际相关的文档进行比较。

用 APK 指标计算时，结果中文档的排名十分重要。如果结果中文档的实际相关性越高且排名也更靠前，那 APK 分值也就越高。由此，它也很适合评估推荐的好坏。因为推荐系统也会计算“前 K 个”推荐物，然后呈现给用户。如果中预测结果中得分更高（值推荐列表中排名也更靠前）的物品实际上也与用户更相关，那那然这个模型就更好。APK 和其他基于排名的指标同样也更适合评估隐式数据集上的推荐。这里用 MSE 相对就不那么合适。

当用 APK 来做评估推荐模型时，每一个用户相当于一个查询，而每一个“前 K 个”推荐物组成的集合相当于一个查到的文档结果集。用户对电影的实际评级便对应着文档的实际相关性。这样，APK 所试图衡量的事模型对用户感兴趣和回去接触的物品的预测能力。

def avgPrecisionK(actual:Seq[Int],predicted:Seq[Int],k: Int): Double = {
    val predK = predicted.take(K)
    var score = 0.0
    var numHits = 0.0
    for((p,i) <- predK.zipWithIndex){
        if(actual.contains(p)){
            numHits += 1.0
            score += numHits / (i.toDouble + 1.0)
        }
    }
    if(actual.isEmpty){
        1.0
    }else{
        score / scala.math.min(actual.size, k).toDouble
    }
}

可以看到，该函数包括两个数组。一个以各个物品及其评级为内容，另一个以模型所预测的物品及评级为内容。
下面来计算对用户 789 推荐的 APK 指标怎么样。首先提取用户实际评级过的电影的 ID：

val actualMovies = moviesForUser.map(_.product)
actualMovies

ArrayBuffer(1012, 127, 475, 93, 1161, 286, 293, 9, 50, 294, 181, 1, 1008, 508, 284, 1017, 137, 111, 742, 248, 249, 1007, 591, 150, 276, 151, 129, 100, 741, 288, 762, 628, 124)

然后提取出推荐的物品列表，K 设定为 10

val predictedMovies = topKPecs.map(_.product)
predictedMovies

Array(693, 56, 320, 412, 527, 182, 108, 475, 76, 129)

计算平均准确率

val apk10 = avgPrecisionK(actualMovies,predictedMovies,10)
apk10

0.0325

这里，APK 的得分为 0，这表明该模型在为该用户做相关电影预测上的表现并不理想。

全局 MAPK 的求解要计算对每一个用户对 APK 得分，再求其平均。这就要为每一个用户都生成相应的推荐列表。针对大规模数据处理时，这并不容易，但我们可以通过 spark 将该计算分布式进行。不过，这就会有一个限制，即每个工作节点都要有完整的物品因子矩阵。

这样它们才能读物地计算某个物品向量与其他所有物品向量之间的相关性。然而当物品数量众多时，单个节点的内存可能保存不下这个矩阵。此时，这个限制也就成了问题。

下面看一看如何求解。首先取回物品因子向量并用它来构建一个 DoubleMatrix 对象：

val itemFactors = model.productFeatures.map{case(id, factor)=>factor}.collect()
val itemMatrix = new DoubleMatrix(itemFactors)
println(itemMatrix.rows, itemMatrix.columns)

(1682,50)

这说明 itemMatrix 的行列数分别为 1682 和 50.这正常，因为电影数目和因子维数分别就是这么多。接下来，我们将该矩阵以广播变量多方式分发出去，以便每个工作节点都能访问到：

val imBroadcast = sc.broadcast(itemMatrix)
imBroadcast

Broadcast(103)

现在可以计算每一个用户的推荐。这会对每一个用户因子进行一次 map 操作。在这个操作里，会对用户因子矩阵和电影因子矩阵做乘积，其结果为一个表示各个电源预计评级的向量（长度为 1682，即电影的总数目）。之后，用预计评级对它们排序：

val allRecs = model.userFeatures.map{
    case(userId,array)=>{
        val userVector = new DoubleMatrix(array)
        val scores = imBroadcast.value.mmul(userVector)
        val sortedWithId = scores.data.zipWithIndex.sortBy(-_._1)
        val recommendedIds = sortedWithId.map(_._2 + 1).toSeq
        (userId, recommendedIds)
    }
}
allRecs

MapPartitionsRDD[462] at map at <console>:43

这样就有了一个由每个用户 ID 及各自对应的电影 ID 列表构成的 RDD。这些电影 ID 按照预计评级的高低排序。

还需要每个用户对应的一个电影 ID 列表作为传人到 APK 函数的 actual 参数。我们已经由 ratings RDD，所以只需从中提取用户和电影的 ID 即可。
使用 spark 的 groupBY 操作便可得到一个新 RDD。该 RDD 包含每个用户 ID 所对应的（userid，movieid）对（因为 groupBY 操作所用的主键就是用户 ID）：

val userMovies = ratings.map{
    case Rating(user,product,rating) => (user,product)
}.groupBy(_._1)
userMovies

ShuffledRDD[465] at groupBy at <console>:35

最后，可以通过 spark 的 join 操作将这两个 RDD 以用户 ID 相连接。这样。对于每一个用户我们都有一个实际和预测的那些电影 ID。这写 ID 可以作为 APK 函数的输入。以计算 MSE 时类似，我们调用 reduce 操作来对这些 APK 得分求和，然后再除以总的用户数目（即 allRecs RDD 的大小）

val K = 10
val MAPK = allRecs.join(userMovies).map{
    case (userId,(predicted,actualWithIds)) => {
        val actual = actualWithIds.map(_._2).toSeq
        avgPrecisionK(actual,predicted,K)
    }
}.reduce(_+_) / allRecs.count
println("Mean Average Precision at K = " + MAPK)

Mean Average Precision at K = 0.03551574677237455

上述代码会输出指定 K 值时的平均准确度：
我们模型的 MAPK 得分相当低。但注意，推荐类任务的这个得分通常都较低，特别是当物品的数量极大时。
试着给 lambada 和 rank 设置其他的值，看一下你能否找到一个 RMSE 和 MAPK 得分更好的模型。

使用 MLlib 内置的评估函数

前面我们从零开始对模型进行了 MSE，RMSE 和 MAPK 三方面的评估。这是一段很有用的练习，同样，MLib 下的 RegressionMetrics 和 RankingMetrics 类也提供了相应的函数以方便模型的评估

RMSE 和 MSE

首先，我们使用 RegressionMetrics 来求解 MSE 和 RMSE 得分。实例化一个 RegressionMetrics 对象需要一个键值对类型的 RDD。其每一条记录对应每个数据点上相应的预测值与实际值。代码实现如下。这里仍然会用到之前已经算出的 ratingsAndPredictions RDD:

import org.apache.spark.mllib.evaluation.RegressionMetrics
val predictedAndTrue = ratingsAndPredictions.map{ case ((user,product),(predicted,actual)) => (predicted, actual)}
val regressionMetrics = new RegressionMetrics(predictedAndTrue)

println("Mean Squared Error = " + regressionMetrics.meanSquaredError)
println("Root Mean Squared Error = " + regressionMetrics.rootMeanSquaredError)

Mean Squared Error = 0.08425286297664057
Root Mean Squared Error = 0.29026343720255326

MAP

以计算 MSE 和 RMSE 一样，可以使用 MLlib 的 RankingMetrics 类来计算基于排名的评估指标。类似地，需要向我们之前的平均准确率函数传入一个键值对类型的 RDD。其健为给定用户预测的推荐物品的 ID 数组，而值则是实际的物品 ID 数组。
RankingMetrics 中的 K 值平均准确率函数实现与我们的有所不同，因而结果会不同。但全局平均准确率（Mean Average Precision, MAP, 并不设定阀值 K）会和当 K 值较大（比如设为总的物品数目）时我们模型的计算结果相同。
首先，使用 RankingMetrics 来计算 MAP

import org.apache.spark.mllib.evaluation.RankingMetrics
val predictedAndTrueForRanking = allRecs.join(userMovies).map{
    case(userId,(predicted,actualWithIds)) => {
        val actual = actualWithIds.map(_._2)
        (predicted.toArray, actual.toArray)
    }
}
val rankingMetrics = new RankingMetrics(predictedAndTrueForRanking)
println("Mean Average Precision = " + rankingMetrics.meanAveragePrecision)

Mean Average Precision = 0.0834671273746858

将 K 值设置为 2000

val MAPK2000 = allRecs.join(userMovies).map{
    case(userId,(predicted,actualWithIds)) => {
        val actual = actualWithIds.map(_._2).toSeq
        avgPrecisionK(actual,predicted,2000)
    }
}.reduce(_+_) / allRecs.count
println("Mean Average Precision =" + MAPK2000)

Mean Average Precision =0.0030767884677027776