基于SparkMLLib实现的商品推荐功能，包括：基于用户的协同过滤，基于物品的协同过滤，基于ALS交替最小二乘的协同过滤

# recommendation-based-on-SparkMLLib ## Spark MLLib 简介 Spark MLlib(Machine Learnig lib) 是Spark对常用的机器学习算法的实现库，同时包括相关的测试和数据生成器。Spark的设计初衷就是为了支持一些迭代的Job, 这正好符合很多机器学习算法的特点。 Spark MLlib目前支持4种常见的机器学习问题: 分类、回归、聚类和协同过滤。Spark MLlib基于RDD，天生就可以与Spark SQL、GraphX、Spark Streaming无缝集成，以RDD为基石，4个子框架可联手构建大数据计算中心！ 下图是MLlib算法库的核心内容：  ## 协同过滤推荐算法 协同过滤算法（Collaborative Filtering：CF）是很常用的一种算法，在很多电商网站上都有用到。CF算法包括基于用户的CF（User-based CF）和基于物品的CF（Item-based CF）。 ### （一）、基于用户（User CF）的协同过滤算法 #### 原理： * 构建用户对物品的打分矩阵  * 根据余弦相似度公式计算**用户**相似度矩阵 <a href="https://github.com/MrQuJL/product-recommendation-system" target="_blank">余弦相似度计算公式</a>：  <a href="https://github.com/MrQuJL/product-recommendation-system" target="_blank">用户相似度矩阵：</a>  * 找出与指定用户相似度最高的前N个用户 * 找出这N个用户评价过的商品，去掉被推荐的用户评价过的商品，则是推荐结果 #### 代码实现： import org.apache.log4j.Logger import org.apache.log4j.Level import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.rdd.RDD import org.apache.spark.mllib.linalg.distributed.MatrixEntry import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix import org.apache.spark.mllib.linalg.distributed.RowMatrix object UserBasedCF { def main(args: Array[String]): Unit = { Logger.getLogger("org.apache.spark").setLevel(Level.ERROR) Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF) // 创建一个SparkContext val conf = new SparkConf().setAppName("test").setMaster("local") val sc = new SparkContext(conf) // 读入数据 val data = sc.textFile("hdfs://qujianlei:9000/data/ratingdata.txt") // 解析出评分矩阵的每一行 val parseData:RDD[MatrixEntry] = data.map(_.split(",") match {case Array(user,item,rate) => MatrixEntry(user.toLong,item.toLong,rate.toDouble)}) // 构建关联矩阵 val ratings = new CoordinateMatrix(parseData) // 转置矩阵以计算列(用户)的相似性 val matrix:RowMatrix = ratings.transpose().toRowMatrix() // 计算得到用户的相似度矩阵 val similarities = matrix.columnSimilarities() println("输出用户相似度矩阵") similarities.entries.collect().map(x=>{ println(x.i + "--->" + x.j + "--->" + x.value) }) println("-----------------------------------------") // 得到某个用户对所有物品的评分 val ratingOfUser1 = ratings.entries.filter(_.i == 1). map(x=>(x.j,x.value)). sortBy(_._1). map(_._1). collect(). toList. toArray println("用户1对所有物品的评分") for (s <- ratingOfUser1) println(s) println("-----------------------------------------") // 得到用户1相对于其他用户的相似性 val similarityOfUser1 = similarities.entries.filter(_.i == 1). sortBy(_.value, false). map(_.value). collect println("用户1相对于其他用户的相似性") for (s <- similarityOfUser1) println(s) // 需求：为用户1推荐2个商品 // 思路：找到与用户1相似性最高的两个用户，将这两个用户评过分的物品，用户1没有评过分的物品推荐给用户1 val similarityTopUser = similarities.entries.filter(_.i == 1). sortBy(_.value, false). map(x=>(x.j, x.value)). collect. take(2) println("与用户1最相似的两个用户如下：") for (s <- similarityTopUser) { // 找到这两个用户评过分的商品，与用户1没有评过分的物品 val userId = s._1 val ratingOfTemp = ratings.entries.filter(_.i == userId). map(x=>(x.j,x.value)). sortBy(_._1). map(_._1). collect(). toList. toArray println("用户" + userId + "对物品的评分:") for (s <- ratingOfTemp) println(s) // 用户1与当前用户求差集 val dis = ratingOfTemp diff ratingOfUser1 println("用户" + userId + "要推荐给用户1的商品id为") dis.foreach(println) } sc.stop() } } ### （二）、基于物品（Item CF）的协同过滤算法 #### 原理： * 构建用户对物品的打分矩阵 * 根据余弦相似度公式计算**物品**相似度矩阵 * 对于当前用户评价高的物品，找出与之相似度最高的N个物品 * 将这N个物品推荐给用户 #### 代码实现： import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.mllib.linalg.distributed.MatrixEntry import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix import org.apache.log4j.Logger import org.apache.log4j.Level import org.apache.spark.rdd.RDD import org.apache.spark.mllib.linalg.distributed.RowMatrix import org.apache.spark.mllib.linalg.distributed.IndexedRow import org.apache.spark.mllib.linalg.SparseVector /* * 建立物品的相似度，来进行推荐 */ object ItemBasedCF { def main(args: Array[String]): Unit = { Logger.getLogger("org.apache.spark").setLevel(Level.ERROR) Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF) //读入数据 val conf = new SparkConf().setAppName("UserBaseModel").setMaster("local") val sc = new SparkContext(conf) val data = sc.textFile("hdfs://qujianlei:9000/data/ratingdata.txt") /*MatrixEntry代表一个分布式矩阵中的每一行(Entry) * 这里的每一项都是一个(i: Long, j: Long, value: Double) 指示行列值的元组tuple。 * 其中i是行坐标，j是列坐标，value是值。*/ val parseData: RDD[MatrixEntry] = data.map(_.split(",") match { case Array(user, item, rate) => MatrixEntry(user.toLong, item.toLong, rate.toDouble) }) //CoordinateMatrix是Spark MLLib中专门保存user_item_rating这种数据样本的 val ratings = new CoordinateMatrix(parseData) /* 由于CoordinateMatrix没有columnSimilarities方法，所以我们需要将其转换成RowMatrix矩阵，调用他的columnSimilarities计算其相似性 * RowMatrix的方法columnSimilarities是计算，列与列的相似度，现在是user_item_rating，与基于用户的CF不同的是，这里不需要进行矩阵的转置，直接就是物品的相似*/ val matrix: RowMatrix = ratings.toRowMatrix() //需求：为某一个用户推荐商品。基本的逻辑是：首先得到某个用户评价过（买过）的商品，然后计算其他商品与该商品的相似度，并排序；从高到低，把不在用户评价过 //商品里的其他商品推荐给用户。 //例如：为用户2推荐商品 //第一步：得到用户2评价过（买过）的商品 take(5)表示取出所有的5个用户 2:表示第二个用户 //解释：SparseVector：稀疏矩阵 val user2pred = matrix.rows.take(5)(2) val prefs: SparseVector = user2pred.asInstanceOf[SparseVector] val uitems = prefs.indices //得到了用户2评价过（买过）的商品的ID val ipi = (uitems zip prefs.values) //得到了用户2评价过（买过）的商品的ID和评分，即：(物品ID,评分) // for (s <- ipi) println(s) // println("*******************") //计算物品的相�