本文共 7313 字,大约阅读时间需要 24 分钟。
用户电影评分数据集下载
1) Item-Based,非个性化的,每个人看到的都一样
2) User-Based,个性化的,每个人看到的不一样对用户的行为分析得到用户的喜好后,可以根据用户的喜好计算相似用户和物品,然后可以基于相似用户或物品进行推荐。这就是协同过滤中的两个分支了,基于用户的和基于物品的协同过滤。在计算用户之间的相似度时,是将一个用户对所有物品的偏好作为一个向量,而在计算物品之间的相似度时,是将所有用户对某个物品的偏好作为一个向量。求出相似度后,接下来可以求相似邻居了。3)基于模型(ModelCF)
按照模型,可以分为:
1)最近邻模型:基于距离的协同过滤算法
2)Latent Factor Mode(SVD):基于矩阵分解的模型
3)Graph:图模型,社会网络图模型
适用场景
对于一个在线网站,用户的数量往往超过物品的数量,同时物品数据相对稳定,因此计算物品的相似度不但 计算量小,同时不必频繁更新。但是这种情况只适用于电子商务类型的网站,像新闻类,博客等这类网站的 系统推荐,情况往往是相反的,物品数量是海量的,而且频繁更新。r语言实现基于物品的协同过滤算法#引用plyr包 library(plyr) #读取数据集 train<-read.table(file="C:/users/Administrator/Desktop/u.data",sep=" ") train<-train[1:3] names(train)<-c("user","item","pref") #计算用户列表方法 usersUnique<-function(){ users<-unique(train$user) users[order(users)] } #计算商品列表方法 itemsUnique<-function(){ items<-unique(train$item) items[order(items)] } # 用户列表 users<-usersUnique() # 商品列表 items<-itemsUnique() #建立商品列表索引 index<-function(x) which(items %in% x) data<-ddply(train,.(user,item,pref),summarize,idx=index(item)) #同现矩阵 cooccurrence<-function(data){ n<-length(items) co<-matrix(rep(0,n*n),nrow=n) for(u in users){ idx<-index(data$item[which(data$user==u)]) m<-merge(idx,idx) for(i in 1:nrow(m)){ co[m$x[i],m$y[i]]=co[m$x[i],m$y[i]]+1 } } return(co) } #推荐算法 recommend<-function(udata=udata,co=coMatrix,num=0){ n<-length(items) # all of pref pref<-rep(0,n) pref[udata$idx]<-udata$pref # 用户评分矩阵 userx<-matrix(pref,nrow=n) # 同现矩阵*评分矩阵 r<-co %*% userx # 推荐结果排序 # 把该用户评分过的商品的推荐值设为0 r[udata$idx]<-0 idx<-order(r,decreasing=TRUE) topn<-data.frame(user=rep(udata$user[1],length(idx)),item=items[idx],val=r[idx]) topn<-topn[which(topn$val>0),] # 推荐结果取前num个 if(num>0){ topn<-head(topn,num) } #返回结果 return(topn) } #生成同现矩阵 co<-cooccurrence(data) #计算推荐结果 recommendation<-data.frame() for(i in 1:length(users)){ udata<-data[which(data$user==users[i]),] recommendation<-rbind(recommendation,recommend(udata,co,0)) }
mareduce 实现
参考文章:
代码下载
spark ALS实现
Spark mllib里用的是矩阵分解的协同过滤,不是UserBase也不是ItemBase。
参考文章:
import org.apache.spark.SparkConfimport org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}import org.apache.spark.rdd._import org.apache.spark.SparkContextimport scala.io.Sourceobject MovieLensALS { def main(args:Array[String]) { //设置运行环境 val sparkConf = new SparkConf().setAppName("MovieLensALS").setMaster("local[5]") val sc = new SparkContext(sparkConf) //装载用户评分,该评分由评分器生成(即生成文件personalRatings.txt) val myRatings = loadRatings(args(1)) val myRatingsRDD = sc.parallelize(myRatings, 1) //样本数据目录 val movielensHomeDir = args(0) //装载样本评分数据,其中最后一列Timestamp取除10的余数作为key,Rating为值,即(Int,Rating) val ratings = sc.textFile(movielensHomeDir + "/ratings.dat").map { line => val fields = line.split("::") // format: (timestamp % 10, Rating(userId, movieId, rating)) (fields(3).toLong % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble)) } //装载电影目录对照表(电影ID->电影标题) val movies = sc.textFile(movielensHomeDir + "/movies.dat").map { line => val fields = line.split("::") // format: (movieId, movieName) (fields(0).toInt, fields(1)) }.collect().toMap //统计有用户数量和电影数量以及用户对电影的评分数目 val numRatings = ratings.count() val numUsers = ratings.map(_._2.user).distinct().count() val numMovies = ratings.map(_._2.product).distinct().count() println("Got " + numRatings + " ratings from " + numUsers + " users " + numMovies + " movies") //将样本评分表以key值切分成3个部分,分别用于训练 (60%,并加入用户评分), 校验 (20%), and 测试 (20%) //该数据在计算过程中要多次应用到,所以cache到内存 val numPartitions = 4 val training = ratings.filter(x => x._1 < 6).values.union(myRatingsRDD).repartition(numPartitions).persist() val validation = ratings.filter(x => x._1 >= 6 && x._1 < 8).values.repartition(numPartitions).persist() val test = ratings.filter(x => x._1 >= 8).values.persist() val numTraining = training.count() val numValidation = validation.count() val numTest = test.count() println("Training: " + numTraining + " validation: " + numValidation + " test: " + numTest) //训练不同参数下的模型,并在校验集中验证,获取最佳参数下的模型 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 for (rank <- ranks; lambda <- lambdas; numIter <- numIters) { val model = ALS.train(training, rank, numIter, lambda) val validationRmse = computeRmse(model, validation, numValidation) println("RMSE(validation) = " + validationRmse + " for the model trained with rank = " + rank + ",lambda = " + lambda + ",and numIter = " + numIter + ".") if (validationRmse < bestValidationRmse) { bestModel = Some(model) bestValidationRmse = validationRmse bestRank = rank bestLambda = lambda bestNumIter = numIter } } //用最佳模型预测测试集的评分,并计算和实际评分之间的均方根误差(RMSE) val testRmse = computeRmse(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 " + testRmse + ".") //create a naive baseline and compare it with the best model val meanRating = training.union(validation).map(_.rating).mean() val baselineRmse = math.sqrt(test.map(x => (meanRating - x.rating) * (meanRating - x.rating)).reduce(_ + _) / numTest) val improvement = (baselineRmse - testRmse) / baselineRmse * 100 println("The best model improves the baseline by " + "%1.2f".format(improvement) + "%.") //推荐前十部最感兴趣的电影,注意要剔除用户已经评分的电影 val myRatedMovieIds = myRatings.map(_.product).toSet val candidates = sc.parallelize(movies.keys.filter(!myRatedMovieIds.contains(_)).toSeq) val recommendations = bestModel.get .predict(candidates.map((0, _))) .collect() .sortBy(-_.rating) .take(10) var i = 1 println("Movies recommended for you:") recommendations.foreach { r => println("%2d".format(i) + ": " + movies(r.product)) i += 1 } sc.stop() } /** 校验集预测数据和实际数据之间的均方根误差 **/ def computeRmse(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) } /** 装载用户评分文件 personalRatings.txt **/ def loadRatings(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 } }}
参考文章:
本文出自 “” 博客,请务必保留此出处
转载地址:http://ujhax.baihongyu.com/