Spark MLlib协同过滤之交替最小二乘法ALS实践

摘自Spark MLlib协同过滤之交替最小二乘法ALS原理与实践

为什么要使用最小二乘法？（稀疏）

在实际应用中，由于用户只会评价或交易少部分物品，评分矩阵一般都非常稀疏。这种情况下的挑战是用相对少的有效评分得到准确的预测。直接做法就是使用矩阵因子分解从评分模式中抽取出一组潜在的因子（latent factors）并通过这些因子向量描述用户和物品。

为什么使用Spark来实现？（基于内存）

Apache Mahout是使用MapReduce实现基于用户和物品的协同过滤推荐算法，我们知道，MapReduce在集群各计算节点的迭代计算中会产生很多的磁盘文件读写操作，严重影响了算法的执行效率，而Spark MLlib是基于内存的分布式计算框架。

最小二乘法（ALS）分类？

显式反馈交替最小二乘法（ALS）：用户对物品有明确的评分喜好。

隐式反馈交替最小二乘法（ALS）：用户对物品没有明确的评分喜好。们只能通过用户的某些行为来推断他对物品的偏好，例如用户浏览，收藏，或交易过某个物品，我们可以认为该用户对这个物品可能感兴趣。例如，在用户浏览某个物品中，对该物品的点击次数或者在物品所在页面上的停留时间越长，这时我们可以推用户对该物品偏好程度更高，但是对于没有浏览该物品，可能是由于用户不知道有该物品，我们不能确定的推测用户不喜欢该物品。ALS-WR通过置信度权重c来解决这些问题：对于更确信用户偏好的项赋以较大的权重，对于没有反馈的项，赋以较小的权重。

Spark MLlib显式ALS算法实现

1、数据准备

alice.txt文件：用户ID，物品ID，用户对物品的评分

1    101 5
1   102 3
1   103 2.5
2   101 2
2   102 2.5
2   103 5
2   104 2
3   101 2.5
3   104 4
3   105 4.5
3   107 5
4   101 5
4   103 3
4   104 4.5
4   106 4
5   101 4
5   102 3
5   103 2
5   104 4
5   105 3.5
5   106 4

2、基于RDD实现


import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.sql.SparkSession//显式ALS实现
object ALSDemo{//解析数据，将数据转换成Rating对象def parseRating(str: String): Rating = {val fields = str.split(",")assert(fields.size == 3)Rating(fields(0).toInt,fields(1).toInt,fields(2).toFloat)}def main(args: Array[String]): Unit = {//切入点val spark = SparkSession.builder().master("local").appName("ASL").getOrCreate()//读取数据，生成RDD并转换为Rating对象val ratingsRDD = spark.sparkContext.textFile("E:\\data\\alice.csv").map(parseRating)//隐藏因子数val rank = 50//最大迭代次数val maxIter = 10//正则化因子val labmda = 0.01//训练模型val model = ALS.train(ratingsRDD,rank,maxIter,labmda)//推荐数量val proNum = 2//推荐val r = model.recommendProductsForUsers(proNum)//打印推荐结果r.foreach(x=>{println("用户 " + x._1)x._2.foreach(x=>{println(" 推荐物品 " + x.product + ", 预测评分 " + x.rating)println()})println("==============================================")})}
}

结果如下：

用户 4
推荐物品 101, 预测评分 4.986308207602999

推荐物品 104, 预测评分 4.504789199111508

==============================================
用户 1
推荐物品 101, 预测评分 4.99307153986906

推荐物品 104, 预测评分 4.851903605652233

==============================================
用户 3
推荐物品 107, 预测评分 4.995334489951884

推荐物品 105, 预测评分 4.510401234084622

==============================================
用户 5
推荐物品 101, 预测评分 4.020799171005919

推荐物品 104, 预测评分 3.99846951877771

==============================================
用户 2
推荐物品 103, 预测评分 4.9851017495990115

推荐物品 102, 预测评分 2.499896082668671

==============================================

计算预测值和真实值的均方根误差RESM

import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSessionobject ALSDemoRESM{//解析数据，将数据转换成Rating对象def parseRating(str: String): Rating = {val fields = str.split(",")assert(fields.size == 3)Rating(fields(0).toInt,fields(1).toInt,fields(2).toFloat)}/*** @param model 训练好的模型* @param data 真实数据* @param n 数据个数* @return 误差*/def rems(model: MatrixFactorizationModel, data: RDD[Rating], n: Long): Double = {//预测值 Rating(userId,itermId,rating)val preRDD: RDD[Rating] = model.predict(data.map(d => (d.user, d.product)))//关联：组成（预测评分，真实评分）val doubleRating = preRDD.map(x => ((x.user, x.product), x.rating)).join(//这里出现错误了，有人知道怎么改请在评论里讲一下，谢谢data.map { x => ((x.user, x.product), x.rating )}).values//计算RMESmath.sqrt(doubleRating.map(x => math.pow(x._1 - x._2, 2)).reduce(_ + _) / n)}def main(args: Array[String]): Unit = {//切入点val spark = SparkSession.builder().master("local").appName("ASL").getOrCreate()//读取数据，生成RDD并转换为Rating对象val ratingsRDD = spark.sparkContext.textFile("E:\\biyesheji\\data\\alice.csv").map(parseRating)//隐藏因子数val rank = 50//最大迭代次数val maxIter = 10//正则化因子val labmda = 0.01//训练模型val model = ALS.train(ratingsRDD,rank,maxIter,labmda)//计算误差val remsValue = rems(model, ratingsRDD, ratingsRDD.count)println("误差：  " + remsValue)}
}

结果如下：

误差：  0.011343969370562474

3、基于DataFrame实现


import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.sql.SparkSession/*** ASL基于DataFrame的Demo*/object ALSDFDemo {case class Rating(userId: Int, itemId: Int, rating: Float)/*** 解析数据：将数据转换成Rating对象* @param str* @return*/def parseRating(str: String): Rating = {val fields = str.split(",")assert(fields.size == 3)Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat)}def main(args: Array[String]): Unit = {//定义切入点val spark = SparkSession.builder().master("local").appName("ASL-DF-Demo").getOrCreate()//读取数据，生成RDD并转换成Rating对象import spark.implicits._val ratingsDF = spark.sparkContext.textFile("E:\\biyesheji\\data\\alice.csv").map(parseRating).toDF()//将数据随机分成训练数据和测试数据（权重分别为0.8和0.2）val Array(training, test) = ratingsDF.randomSplit(Array(0.8, 0.2))//定义ALS，参数初始化val als = new ALS().setRank(50).setMaxIter(10).setRegParam(0.01).setUserCol("userId").setItemCol("itemId").setRatingCol("rating")//训练模型val model = als.fit(training)//推荐:每个用户推荐2个物品val r = model.recommendForAllUsers(2)//关闭冷启动（防止计算误差不产生NaN）model.setColdStartStrategy("drop")//预测测试数据val predictions = model.transform(test)//定义rmse误差计算器val evaluator = new RegressionEvaluator().setMetricName("rmse").setLabelCol("rating").setPredictionCol("prediction")//计算误差val rmse = evaluator.evaluate(predictions)//打印训练数据training.foreach(x=>println("训练数据： "+x))//打印测试数据test.foreach(x=>println("测试数据： "+x))//打印推荐结果r.foreach(x=>print("用户 "+x(0)+" ,推荐物品 "+x(1)))//打印预测结果predictions.foreach(x=>print("预测结果:  "+x))//输出误差println(s"Root-mean-square error = $rmse")}
}

结果如下：

训练数据： [1,101,5.0]
训练数据： [1,102,3.0]
训练数据： [1,103,2.5]
训练数据： [2,101,2.0]
训练数据： [2,102,2.5]
训练数据： [2,104,2.0]
训练数据： [3,101,2.5]
训练数据： [3,105,4.5]
训练数据： [3,107,5.0]
训练数据： [4,101,5.0]
训练数据： [4,103,3.0]
训练数据： [4,104,4.5]
训练数据： [4,106,4.0]
训练数据： [5,102,3.0]
训练数据： [5,103,2.0]
训练数据： [5,104,4.0]
训练数据： [5,105,3.5]测试数据： [2,103,5.0]
测试数据： [3,104,4.0]
测试数据： [5,101,4.0]
测试数据： [5,106,4.0]用户 1 ,推荐物品 WrappedArray([101,4.98618], [105,3.477826])
用户 3 ,推荐物品 WrappedArray([107,4.9931526], [105,4.499714])
用户 5 ,推荐物品 WrappedArray([104,3.9853115], [105,3.4996033])
用户 4 ,推荐物品 WrappedArray([101,5.000056], [104,4.5001974])
用户 2 ,推荐物品 WrappedArray([105,3.0707152], [102,2.4903712])预测结果:  [5,101,4.0,3.1271331]
预测结果:  [2,103,5.0,1.0486442]
预测结果:  [5,106,4.0,1.8420099]
预测结果:  [3,104,4.0,1.4847627]Root-mean-square error = 2.615265256309832

Spark MLlib协同过滤之交替最小二乘法ALS实践相关推荐

mllib调参 spark_Spark MLlib协同过滤算法
算法说明协同过滤(Collaborative Filtering,简称CF,WIKI上的定义是:简单来说是利用某个兴趣相投.拥有共同经验之群体的喜好来推荐感兴趣的资讯给使用者,个人透过合作的机制给予 ...
mllib协同过滤 java实现_协同过滤(ALS)算法介绍及Spark MLlib调用实例(Scala/Java/Python)...
协同过滤算法介绍: 协同过滤常被用于推荐系统.这类技术目标在于填充"用户-商品"联系矩阵中的缺失项.Spark.ml目前支持基于模型的协同过滤,其中用户和商品以少量的潜在因子来描 ...
Spark ML - 协同过滤
http://ihoge.cn/2018/ML1.html 协同过滤算法获取spark自带的MovieLens数据集,其中每行包含一个用户.一个电影.一个该用户对该电影的评分以及时间戳.我们使用 ...
Spark机器学习——协同过滤推荐算法
协同过滤介绍协同过滤简单来说是利用某兴趣相投.拥有共同经验之群体的喜好来推荐用户感兴趣的信息,个人通过合作的机制给予信息相当程度的回应(如评分)并记录下来以达到过滤的目的进而帮助别人筛选信息,回应不 ...
Spark的协同过滤.Vs.Hadoop MR
基于物品的协同过滤推荐算法案例在TDW Spark与MapReudce上的实现对比,相比于MapReduce,TDW Spark执行时间减少了66%,计算成本降低了40%. 原文链接:http://w ...
Spark实现协同过滤CF算法实践
Spark编写Scala实现CF算法 UI矩阵–>II矩阵–>排序数据文件:user_item_score.data 下载链接:https://pan.baidu.com/s/1JH57 ...
Spark MLlib中的协同过滤
本文转自http://www.tuicool.com/articles/fANvieZ,所有权力归原作者所有. 本文主要通过Spark官方的例子,理解ALS协同过滤算法的原理和编码过程. 协同过滤协 ...
基于MLlib的机器学习--协同过滤与推荐
<Spark快速大数据分析> 11.5.4 协同过滤与推荐协同过滤是一种根据用户对各种产品的交互与评分来推荐新产品的推荐系统技术. 协同过滤引入的地方就在于它只需要输入一系列用户/产品的 ...
人工智能算法之梯度下降法、协同过滤、相似度技术、ALS算法（附案例分析）、模型存储与加载、推荐系统的冷启动问题
梯度下降法求解机器学习算法的模型参数,即无约束优化问题时,梯度下降法是最常采用的方法之一,另一种常用的方法是最小二乘法.这里对梯度下降法做简要介绍. 最小二乘法法适用于模型方程存在解析解的情况.如果 ...

Spark MLlib协同过滤之交替最小二乘法ALS实践

Spark MLlib协同过滤之交替最小二乘法ALS实践相关推荐

最新文章

热门文章