本文内容和代码是接着上篇文章来写的，推荐先看一下哈~
我们上一篇文章是写了电影推荐的实现，但是推荐内容是否合理呢，这就需要我们对模型进行评估
针对推荐模型，这里根据 均方差 和 K值平均准确率 来对模型进行评估，MLlib也对这几种评估方法都有提供内置的函数
在真实情况下，是要不断地对推荐模型的三个关键参数 rank、iterations、lambda 分别选取不同的值，然后对不同参数生成的模型进行评估，从而选取出最好的模型。
下面就对两种推荐模型评估的方法进行说明~
1、均方差（MSE） 和 均方根误差（RMSE）
定义：各平方误差的和与总数目的商。其实可以理解为 预测到的评级 与 真实评级的差值 的平方。
均方根误差的使用也很普遍，其计算只需在MSE上取平方根即可~
评估代码为：
//格式：(userID，电影)
val userProducts: RDD[(Int, Int)] = ratings.map(rating => (rating.user, rating.product))
//模型推测出的评分信息，格式为：((userID，电影), 推测评分)
val predictions: RDD[((Int, Int), Double)] = model.predict(userProducts).map(rating => ((rating.user, rating.product),rating.rating))
//格式为：((userID，电影), (真实平评分，推测评分))
val ratingsAndPredictions: RDD[((Int, Int), (Double, Double))] = ratings.map(rating => ((rating.user, rating.product), rating.rating))
.join(predictions)
//均方差
val MSE = ratingsAndPredictions.map(rap => math.pow(rap._2._1 - rap._2._2, 2)).reduce(_+_) / ratingsAndPredictions.count()
println("MSE：" + MSE)
//均方根误差
val RMSE: Double = math.sqrt(MSE)
println("RMSE：" + RMSE)
上面是我们自己算出来的，也可以用MLlib内置的函数来算：
import org.apache.spark.mllib.evaluation.{RegressionMetrics, RankingMetrics}
val predictedAndTrue: RDD[(Double, Double)] = ratingsAndPredictions.map{ case((userID, product),(actual, predict)) => (actual, predict)}
val regressionMetrics: RegressionMetrics = new RegressionMetrics(predictedAndTrue)
println("MSE：" + regressionMetrics.meanSquaredError)
println("RMSE：" + regressionMetrics.rootMeanSquaredError)
输出为：
MSE：0.08231947642632852
RMSE：0.2869137090247319
2、K值平均准确率（MAPK）
K值平均准确率（MAPK）的意思是整个数据集上的K值平均准确率（APK）的均值。APK是信息检索中常用的一个指标。它用于衡量针对某个查询所返回的“前K个”文档的平均相关性。
如果结果中文档的实际相关性越高且排名也更靠前，那APK分值也就越高。如果在预测结果中得分更高（在推荐列表中排名也更靠前）的物品实际上也与用户更相关，那自然这个模型就更好。
ok，MAPK评估代码如下：
package ml
import org.apache.spark.mllib.evaluation.RankingMetrics
import org.apache.spark.mllib.recommendation.{Rating, ALS}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}
import org.jblas.DoubleMatrix
import sql.StreamingExamples
import scala.collection.Map
object MAPKTest{
def main(args: Array[String]) {
StreamingExamples.setStreamingLogLevels()
val conf = new SparkConf().setAppName("MAPKTest").setMaster("local[*]")
val sc = new SparkContext(conf)
/*用户 电影 评分*/
val rawData: RDD[String] = sc.textFile("file:///E:/spark/ml-100k/u.data")
//去掉时间的字段，rawRatings:Array
val rawRatings = rawData.map(_.split("\t").take(3))
//user moive rating
val ratings = rawRatings.map{case Array(user, movie, rating) =>{
Rating(user.toInt, movie.toInt, rating.toDouble)
}}
/**
* 得到训练的模型
* 注意：50代表我们得到的模型的因子的列的数量，名称叫 因子维数
*/
val model = ALS.train(ratings, 50, 10, 0.01)
/*获取模型中所有商品的 factor，并转换成矩阵*/
val itemFactors: Array[Array[Double]] = model.productFeatures.map{case (id, factor) => factor}.collect()
val itemMatrix: DoubleMatrix = new DoubleMatrix(itemFactors)
// println(itemMatrix.rows, itemMatrix.columns)
/*获得模型中每个用户对应的每个电影的评分*/
val allRecs = model.userFeatures.map{ case(userId, factor) => {
val userVector = new DoubleMatrix(factor)
/**
* socres是一个DoubleMatrix类型，值为1行N列的 Vector
* 为什么可以通过判断这两个矩阵的乘积的大小，从而来判断分数呢？
* 这归根于ALS算法，该算法是将一个 用户-商品 的矩阵 拆分成 用户、商品两个矩阵
* 因此这两个矩阵的乘积就是实际的 分数
*/
val scores = itemMatrix.mmul(userVector)//矩阵和向量的乘积，求出每个用户的分数
//根据评分倒数排序
val sortedWithId = scores.data.zipWithIndex.sortBy(-_._1)
//(score, itemId)
val recommendIds = sortedWithId.map(_._2 + 1).toSeq
//返回用户 和 各个商品评分的倒数的值 的 tuple: (userId,(sorce, itemId))
(userId, recommendIds)
}}
/*获取实际中的 每个用户对应的有评分过的电影的评分*/
val userMoives: RDD[(Int, Iterable[(Int, Int)])] = ratings.map{ case Rating(user, product, rating) => {
(user, product)
}}.groupBy(_._1)
val predictedAndTrueForRanking = allRecs.join(userMoives).map{ case( userId, (predicted, actualWithIds) ) => {
//实际的商品编号
val actual = actualWithIds.map(_._2)
(actual.toArray, predicted.toArray)
}}
val rankingMetrics: RankingMetrics[Int] = new RankingMetrics(predictedAndTrueForRanking)
println("使用内置的计算MAP：" + rankingMetrics.meanAveragePrecision)
}
输出结果为：
使用内置的计算MAP：0.0630466936422453
3、推荐模型完整代码
package ml
import org.apache.spark.mllib.evaluation.{RegressionMetrics, RankingMetrics}
import org.apache.spark.mllib.recommendation.{Rating, ALS}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}
import org.jblas.DoubleMatrix
import sql.StreamingExamples
import scala.collection.Map
/**
* 基于Spark MLlib 的推荐算法
* ALS：最小二乘法
*
* @author lwj
* @date 2018/05/04
*/
object Recommend{
/**
* 用于商品推荐
* 通过传入两个向量，返回这两个向量之间的余弦相似度
*
* @param vec1
* @param vec2
* @return
*/
def cosineSimilarity(vec1: DoubleMatrix, vec2: DoubleMatrix): Double = {
vec1.dot(vec2) / (vec1.norm2() * vec2.norm2())
}
/**
* 模型评估
* K值平均准确率(APK)
*
* @param actual
* @param predicted
* @param k
* @return
*/
def avgPrecisionK(actual: Seq[Int], predicted: Seq[Int], k: Int) : Double = {
val predK: Seq[Int] = 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) //TODO 为什么除以i.toDouble
}
}
if (actual.isEmpty){
1.0
}else{
score / math.min(actual.size, k).toDouble //TODO 为什么是min
}
}
def main(args: Array[String]) {
StreamingExamples.setStreamingLogLevels()
val conf = new SparkConf().setAppName("recommandTest").setMaster("local[*]")
val sc = new SparkContext(conf)
/*用户 电影 评分*/
val rawData: RDD[String] = sc.textFile("file:///E:/spark/ml-100k/u.data")
//去掉时间的字段，rawRatings:Array
val rawRatings = rawData.map(_.split("\t").take(3))
//user moive rating
val ratings = rawRatings.map{case Array(user, movie, rating) =>{
Rating(user.toInt, movie.toInt, rating.toDouble)
}}
//电影
val movies: RDD[String] = sc.textFile("file:///E:/spark/ml-100k/u.item")
//电影ID 电影名
val titles: Map[Int, String] = movies.map(_.split("\|").take(2)).map(array => (array(0).toInt, array(1))).collectAsMap()
/**
* 得到训练的模型
* 注意：50代表我们得到的模型的因子的列的数量，名称叫 因子维数
*/
val model = ALS.train(ratings, 50, 10, 0.01)
/**
* 基于用户进行推荐
*/
//用户因子的数量
// println(mode.userFeatures.count())
//商品因子的数量
// println(mode.productFeatures.count())
//查看某个用户对某个商品的预测评分，ALS模型的初始化是随机的，所以产生的结果可能会不同
// println(mode.predict(789, 123))
//为指定的用户推荐 N 个商品
val userID = 789
val K = 10
val topKRecs: Array[Rating] = model.recommendProducts(userID, 10)
// println(topKRecs.mkString("
"))
//获取指定用户所评价过的电影
val moviesForUser: Seq[Rating] = ratings.keyBy(_.user).lookup(789)
//打印出指定用户评价最高的10部电影的名称和评分
println("真实的：")
moviesForUser.sortBy(-_.rating).take(10).map(rating => {
(titles(rating.product),rating.rating)
}).foreach(println)
//打印出推荐给用户的10部电影的名称和评分，和上面的进行比较
println("推荐的：")
topKRecs.map(rating => {
(titles(rating.product),rating.rating)
}).foreach(println)
println("
-----------------------
")
/**
* 基于商品进行推荐
*/
/*通过商品ID获得与该商品相似的商品*/
val itemId = 567
val itemFactor: Array[Double] = model.productFeatures.lookup(itemId).head
val itemVector: DoubleMatrix = new DoubleMatrix(itemFactor)
//获得每个商品与给出的商品的余弦相似度
val sims = model.productFeatures.map{case (id, factor) => {
val factorVector = new DoubleMatrix(factor)
val sim = cosineSimilarity(factorVector, itemVector)
(id, sim)
}}
//打印出前N的商品
val topItem: Array[(Int, Double)] = sims.sortBy(-_._2).take(10 + 1)
println("与567商品相似的商品：
" + topItem.mkString("
") + "
")
/*校验商品*/
println("给定的商品名称为： " + titles(itemId))
println("相似的商品名称为：")
topItem.slice(1, 11).foreach(item => println(titles(item._1)))
println("
-----------------------
")
/*模型评估*/
/**
* 均方差评估
* 对model全量数据进行评估
*/
// val actualRating: Rating = moviesForUser.take(1)(0)
// val predictedRating: Double = model.predict(789, actualRating.product)
// println("
真实分：" + actualRating.rating + " 预测分：" + predictedRating)
//格式：(userID，电影)
val userProducts: RDD[(Int, Int)] = ratings.map(rating => (rating.user, rating.product))
//模型推测出的评分信息，格式为：((userID，电影), 推测评分)
val predictions: RDD[((Int, Int), Double)] = model.predict(userProducts).map(rating => ((rating.user, rating.product),rating.rating))
//格式为：((userID，电影), (真实平评分，推测评分))
val ratingsAndPredictions: RDD[((Int, Int), (Double, Double))] = ratings.map(rating => ((rating.user, rating.product), rating.rating))
.join(predictions)
//均方差
val MSE = ratingsAndPredictions.map(rap => math.pow(rap._2._1 - rap._2._2, 2)).reduce(_+_) / ratingsAndPredictions.count()
println("均方差MSE为： " + MSE)
//均方根误差
val RMSE: Double = math.sqrt(MSE)
println("均方根误差RMSE为： " + RMSE)
/**
* K值平均准确率评估
* 注意：该评估模型是针对对用户感兴趣和回去接触的物品的预测能力
* 也是就是说：这时针对基于用户推荐的 模型的评估
*/
/*计算 单个 指定用户推荐的APK指标*/
val actualMovies: Seq[Int] = moviesForUser.map(_.product)
val predictedMovies: Array[Int] = topKRecs.map(_.product)
val apk10: Double = avgPrecisionK(actualMovies, predictedMovies, 10)
println("789的APK值为：" + apk10)
/*获取模型中所有商品的 factor，并转换成矩阵*/
val itemFactors: Array[Array[Double]] = model.productFeatures.map{case (id, factor) => factor}.collect()
val itemMatrix: DoubleMatrix = new DoubleMatrix(itemFactors)
// println(itemMatrix.rows, itemMatrix.columns)
/*获得模型中每个用户对应的每个电影的评分*/
val allRecs = model.userFeatures.map{ case(userId, factor) => {
val userVector = new DoubleMatrix(factor)
/**
* socres是一个DoubleMatrix类型，值为1行N列的 Vector
* 为什么可以通过判断这两个矩阵的乘积的大小，从而来判断分数呢？
* 这归根于ALS算法，该算法是将一个 用户-商品 的矩阵 拆分成 用户、商品两个矩阵
* 因此这两个矩阵的乘积就是实际的 分数
*/
val scores = itemMatrix.mmul(userVector)//矩阵和向量的乘积，求出每个用户的分数
//根据评分倒数排序
val sortedWithId = scores.data.zipWithIndex.sortBy(-_._1)
//(score, itemId)
val recommendIds = sortedWithId.map(_._2 + 1).toSeq
//返回用户 和 各个商品评分的倒数的值 的 tuple: (userId,(sorce, itemId))
(userId, recommendIds)
}}
/*获取实际中的 每个用户对应的有评分过的电影的评分*/
val userMoives: RDD[(Int, Iterable[(Int, Int)])] = ratings.map{ case Rating(user, product, rating) => {
(user, product)
}}.groupBy(_._1)
val MAPK = allRecs.join(userMoives).map{ case( userId, (predicted, actualWithIds) ) => {
//实际的商品编号
val actual = actualWithIds.map(_._2).toSeq
avgPrecisionK(actual, predicted, 10)
}}.reduce(_ + _) / allRecs.count
println("MAPK：" + MAPK)
println("
-----------------------
")
/**
* 使用MLlib内置的评估器
*/
/*RMSE 和 MSE*/
val predictedAndTrue: RDD[(Double, Double)] = ratingsAndPredictions.map{ case((userID, product),(actual, predict)) => (actual, predict)}
val regressionMetrics: RegressionMetrics = new RegressionMetrics(predictedAndTrue)
println("使用内置的计算MSE：" + regressionMetrics.meanSquaredError)
println("使用内置的计算RMSE：" + regressionMetrics.rootMeanSquaredError)
/*MAPK*/
val predictedAndTrueForRanking = allRecs.join(userMoives).map{ case( userId, (predicted, actualWithIds) ) => {
//实际的商品编号
val actual = actualWithIds.map(_._2)
(actual.toArray, predicted.toArray)
}}
val rankingMetrics: RankingMetrics[Int] = new RankingMetrics(predictedAndTrueForRanking)
println("使用内置的计算MAP：" + rankingMetrics.meanAveragePrecision)
}
}