SparkML之推荐引擎（二)---推荐模型评估

本文内容和代码是接着上篇文章来写的，推荐先看一下哈~ 
我们上一篇文章是写了电影推荐的实现，但是推荐内容是否合理呢，这就需要我们对模型进行评估 
针对推荐模型，这里根据 均方差 和 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("\n")) //获取指定用户所评价过的电影
    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("\n-----------------------\n") /** * 基于商品进行推荐 */
    /*通过商品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商品相似的商品：\n" + topItem.mkString("\n") + "\n") /*校验商品*/ println("给定的商品名称为： " + titles(itemId)) println("相似的商品名称为：") topItem.slice(1, 11).foreach(item => println(titles(item._1))) println("\n-----------------------\n") /*模型评估*/
    /** * 均方差评估 * 对model全量数据进行评估 */
// val actualRating: Rating = moviesForUser.take(1)(0) // val predictedRating: Double = model.predict(789, actualRating.product) // println("\n真实分：" + 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("\n-----------------------\n") /** * 使用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) } }