Spark 机器学习之电影推荐系统

一、实验介绍

1.1 实验内容

说到推荐系统，大家可能立马会想到协同过滤算法。本节课主要讲解协同过滤推荐算法，之后会实现一个简易的电影推荐系统，让您体会 Spark MLlib 在机器学习领域的强悍，然后学以致用。

1.2 先学课程

Spark 系列课程

1.3 实验知识点

• 协同过滤算法

1.4 实验环境

• spark-2.1.0-bin-hadoop2.6

• Xfce 终端

1.5 适合人群

本课程属于中级难度级别，适合具有 Spark 基础的用户，如果对 Scala 熟悉 ，能够更好的上手本课程。

二、理论基础

2.1机器学习概述

机器学习是从已经存在的数据进行学习来对将来进行数据预测，它是基于输入数据集创建模型做数据驱动决策。

常见的几类机器学习模型：

监督学习模型 : 监督学习模型对已标记的训练数据集训练出结果，然后对未标记的数据集进行预测。

非监督学习模型 : 非监督学习模型是用来从原始数据（无训练数据）中找到隐藏的模式或者关系，因而非监督学习模型是基于未标记数据集的。

半监督学习模型 :半监督学习模型用在监督和非监督机器学习中做预测分析，其既有标记数据又有未标记数据。典型的场景是混合少量标记数据和大量未标记数据。半监督学习一般使用分类和回归的机器学习方法。

增强学习模型 : 增强学习模型通过不同的行为来寻找目标回报函数最大化。

机器学习应用

推荐引擎：
异常监测:

Spark MLlib 库提供给了几个实现的算法，比如，线性 SVM、逻辑回归、决策树和贝叶斯算法。另外，一些集成模型，比如随机森林和 gradient-boosting 树。

2.2 基于 Spark MLlib 平台的协同过滤算法

什么是协同过滤 (Collaborative Filtering, 简称 CF)？

首先想一个简单的问题，如果你现在想看个电影，但你不知道具体看哪部，你会怎么做？

大部分的人会问问周围的朋友，看看最近有什么好看的电影推荐，而我们一般更倾向于从口味比较类似的朋友那里得到推荐。这就是协同过滤的核心思想。协同过滤算法又分为基于用户的协同过滤算法和基于物品的协同过滤算法。

协同过滤算法按照数据使用可以分为：

• 基于用户（UserCF）：通过不同用户对物品的评分来评测用户之间的相似性，基于用户之间的相似性做出推荐。简单来讲，就是给用户推荐和他兴趣相似的其他用户喜欢的物品。

• 基于商品（ItemCF）：通过用户对不同item的评分来评测item之间的相似性，基于item之间的相似性做出推荐。简单来讲，就是给用户推荐和他之前喜欢的物品相似的物品。

• 基于模型（ModelCF）：基于模型的协同过滤推荐就是基于样本的用户喜好信息，训练一个推荐模型，然后根据实时的用户喜好的信息进行预测，计算推荐。

关于以上3种不同的协同算法的详细内容请移动阅读： http://blog.csdn.net/yimingsilence/article/details/54934302

三、实验步骤

3.1 实验数据下载

• 电影数据集下载地址：grouplens

• 本实验使用的数据集下载地址：ml-1m.zip

可以通过以下方式下载并解压文件：

wget  http://labfile.oss.aliyuncs.com/courses/831/ml-1m.zip  
unzip ml-1m.zip
ls ml-1m

3.2 数据集各文件说明

该数据集，拥有6000+个用户，3800+部电影，100多万的评分数据，本次测试数据主要包括四个数据文件（详细的数据描述参见README文件）。

• movies.dat (电影资源数据)，格式为：

  电影id::电影名称::电影类型

使用 tail -f 命令查看 movies.dat ，可以看到有3800+部电影。

• ratings.dat (评分数据)，格式为：

  用户Id::电影Id::评分::时间

注意：这个数据集中不包含我的评分数据(person.txt)，也就是不包含用户Id为0的数据）

使用 tail -f 命令查看 ratings.dat ，可以看到6000+用户。

• users.dat (评分数据)，格式为：

  用户Id::性别::年龄::职业编号:邮编

使用 tail -f 命令查看 users.dat ，可以看到6000+用户。

• person.txt (注意：这个数据集是我们自己创建在ml-1m下的，用户 Id 为0)，格式为：

  我的用户Id::我看过的电影Id::我对该电影的评分::评分的时间戳

使用 more 命令查看 person.txt 。

3.2 整体思路

• 获取数据，即我们解压 ml-1m.zip 之后的，包含 movies.dat ， ratings.dat ， users.dat ， README 。

• 我在一个电影网站上看了几部电影，并都为其做了评分操作（0-5分），这个数据集是我们自己创建的 person.txt

• 该电影网站的推荐系统根据 person.txt 数据集（即我对那几部电影的评分），预测出在该网站的电影资源库中，有哪些电影是适合我的，并推荐给我看。

• 根据我的观影习惯和用户的一个个人信息，预测该网站用户库中，哪些人和我的兴趣爱好是差不多的，并推荐给我认识，预测方法涉及训练数据集，均方根误差等。

3.3 代码实现

在开始代码前，不妨先来个热身准备，统计得分最高的10部电影。

双击桌面上的 Xfce 终端，进入 spark-shell 模式.

#加载数据集
val ratingsRdd = sc.textFile("/home/shiyanlou/ml-1m/ratings.dat") 
#切分，缓存
val ratings = ratingsRdd.map(_.split("::")).map { x =>
(x(0), x(1), x(2))
}.cache

#取前10
 val topK10ScoreMovie = ratings.map{x =>
(x._2, (x._3.toInt, 1))
}.reduceByKey { (v1, v2) =>
(v1._1 + v2._1, v1._2 + v2._2)
}.map { x =>
(x._2._1.toFloat / x._2._2.toFloat, x._1)
}.sortByKey(false).
take(10).
foreach(println)

接下来正式进入主题，这里还是用上面的 spark-shell ，前三个数据文件用于模型训练，第四个数据文件用于测试模型。

1.首先导入依赖

import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd._
import org.apache.spark.{SparkContext, SparkConf}
import scala.io.Source

2.定义addRatings有参方法，加载用户评分文件 person.txt。

def addRatings(path:String):Seq[Rating] = {
    val lines = Source.fromFile(path).getLines()
    val ratings = lines.map{
      line =>
        val fields = line.split("::")
        Rating(fields(0).toInt,fields(1).toInt,fields(2).toDouble)
    }.filter(_.rating > 0.0)
    if(ratings.isEmpty){
      sys.error("No ratings provided.")
    }else{
      ratings.toSeq
    }
  }
    val myRatings = addRatings("/home/shiyanlou/ml-1m/person.txt")
    val myRatingsRDD = sc.parallelize(myRatings, 1)

3.加载样本评分数据，最后一列 Timestamp 取除10的余数作为 key，Rating 为值，即(Int，Rating)。

val ratings = sc.textFile("/home/shiyanlou/ml-1m/ratings.dat").map {
      line =>
        val fields = line.split("::")

        (fields(3).toLong % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble))
    }

4.加载movies.dat，取字段(movieId, movieName)。

val movies = sc.textFile("/home/shiyanlou/ml-1m/movies.dat").map {
      line =>
        val fields = line.split("::")
        (fields(0).toInt, fields(1)) 
    }.collect().toMap

5.统计用户数量和电影数量以及用户对电影的评分数目。

val numRatings = ratings.count()
    val numUsers = ratings.map(_._2.user).distinct().count()
    val numMovies = ratings.map(_._2.product).distinct().count()
    println("The data contains " + numRatings + " ratings from  " + numUsers + " users " + numMovies + " movies")

6.将样本评分表以 key 值切分成3个部分，分别用于训练(55%，并加入用户评分), 校验 (15%), 测试(30%)，该数据在计算过程中要多次应用到，这里用cache缓存。

val numPartitions = 4
    val training = ratings.filter(x => x._1 < 6).values.union(myRatingsRDD).repartition(numPartitions).cache()
    val validation = ratings.filter(x => x._1 >= 6 && x._1 < 8).values.repartition(numPartitions).cache()
    val test = ratings.filter(x => x._1 >= 8).values.cache()
    val numTraining = training.count()
    val numValidation = validation.count()
    val numTest = test.count()
    println("Training: " + numTraining + " validation: " + numValidation + " test: " + numTest)

7.训练不同参数下的模型，迭代次数,根据机器情况设定,并在校验集中验证，获取最佳参数下的模型。

val ranks = List(8, 12)
    val lambdas = List(0.1, 10.0)
    val numIters = List(10, 20)
    var bestModel: Option[MatrixFactorizationModel] = None
    var bestValidationRmse = Double.MaxValue
    var bestRank = 0
    var bestLambda = -1.0
    var bestNumIter = -1

8.定义 compute 函数校验集预测数据和实际数据之间的均方根误差。

def compute(model:MatrixFactorizationModel,data:RDD[Rating],n:Long):Double = {

    val predictions:RDD[Rating] = model.predict((data.map(x => (x.user,x.product))))
    val predictionsAndRatings = predictions.map{ x =>((x.user,x.product),x.rating)}
      .join(data.map(x => ((x.user,x.product),x.rating))).values
    math.sqrt(predictionsAndRatings.map( x => (x._1 - x._2) * (x._1 - x._2)).reduce(_+_)/n)
  }

9.三层嵌套循环，会产生8个ranks ，lambdas ，iters 的组合，每个组合都会产生一个模型，计算8个模型的方差，最小的那个记为最佳模型。

for (rank <- ranks; lambda <- lambdas; numIter <- numIters) {
      val model = ALS.train(training, rank, numIter, lambda)
      val validRmse = compute(model, validation, numValidation)
      println("validation= " + validRmse + " for the model trained with rank = "
        + rank + ",lambda = " + lambda + ",and numIter = " + numIter + ".")
      if (validRmse < bestValidationRmse) {
        bestModel = Some(model)
        bestValidationRmse = validRmse
        bestRank = rank
        bestLambda = lambda
        bestNumIter = numIter
      }
    }

注意:迭代过程执行时间可能有点长，耐心等待，部分截图如下：

10.用最佳模型预测测试集的评分，并计算和实际评分之间的均方根误。

val tRm = compute(bestModel.get, test, numTest)
    println("The best model was trained with rank = " + bestRank + " and lambda = " + bestLambda
      + ", and numIter = " + bestNumIter + ", and its RMSE on the test set is " + tRm + ".")

11.计算最佳模型与原始基础的相比其提升度。

val inspopular = training.union(validation).map(_.rating).mean
    val bRm = math.sqrt(test.map(x => (inspopular - x.rating) * (inspopular - x.rating)).reduce(_ + _) / numTest)
    val improvement = (bRm - tRm) / bRm * 100
    println("The best model improves the baseline by " + "%1.2f".format(improvement) + "%.")

12.推荐前15部最感兴趣的电影给我，注意要剔除用户(我)已经评分的电影。

val myinterestedIds = myRatings.map(_.product).toSet
    val choice = sc.parallelize(movies.keys.filter(!myinterestedIds.contains(_)).toSeq)
      var i = 1

    println("Movies recommended for you:")
    bestModel.get.predict(choice.map((0, _))).collect.sortBy(-_.rating).take(15).foreach { r =>
      println("%2d".format(i) + ": " + movies(r.product))
      i += 1
    }

这样，一个简单的基于模型的电影推荐应用就算 OK 了。

关于本实验的一点深入思考：

推荐的最终结果不一定准确，可以调整参数使得预测结果偏优。

• 增加迭代次数，次数越多，lambda 较小，均方差会较小，推荐结果更好
• 置数据最好随机划分，结果更有说服力。
• 数据量增大时，通过提高并行度，可以减少运行时间。

四、实验总结

Spark MLlib 是 Spark 实现的机器学习库中的一种，经常用来做业务数据的预测分析，比如个性化推荐引擎和异常监测系统。本节课主要介绍了协同过滤推荐算法，并基于电影数据集进行一个简易的电影推荐系统，希望对你有些帮助，有些启发。

五、参考资料 - 列表项

• http://snglw.blog.51cto.com/5832405/1662153