数据文件:
u.data(userid itemid rating timestamp)
u.item(主要使用 movieid movietitle)
数据操作
把u.data导入RDD， take() x.split(‘\t’)(1)
查看userid字段的统计信息
查看udata数据矩阵的 userid列上所有值的统计信息
使用ALS.train进行训练
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating
//读取userData前3个字段 用户，产品，评分来创建rawRatings
val userData=sc.textFile(”file:/home/hadoop/ALS/u.data”)
val rawRatings=userData.map(_.split(‘\t’).take(3))
//1.准备ALS训练数据
// import org.apache.spark.mllib.recommendation.Rating
//把3个split列case到3个名称变量中（user,movie,rating）后面便于操作
val ratingRDD= rawRatings.map{
case Array(user,movie,rating)=>Rating(user.toInt,movie.toInt,rating.toDouble)
}
//2.使用ALS.train进行训练
//使用显示Explicit rating评分训练.
// ALS.train(ratings:RDD[Rating],rank:Int,iterations:Int,lambda:Double):MatrixFactorizationModel
//隐式Implicit评分训练
// ALS.trainImplicit(ratings:RDD[Rating],rank:Int,iterations:Int,lambda:Double):MatrixFactorizationModel
//Rating(userID,productID,rating),rank 当矩阵分解matrix factorization时，将原本矩阵A(mXn)分解成X(m X rank) Y(rank X n)矩阵
//Iterator ALS算法重复计算次数，建议10~20
//lambda 建议0.01 会影响准确度|时间
//MatrixFactorizationMode训练完成后产生的模型，训练时会执行矩阵分解。模型对象成员 rank，
userFeatures:RDD[(Int,Array[Double])]分解后的用户矩阵X(m*rank)
productFeatures:RDD[(Int,Array[Double])]分解后的产品矩阵Y(m*rank)
val model=ALS.train(ratingrdd,10,10,0.01)
使用模型推荐
//1.针对用户 推荐电影
model.recommendProducts(196,5).mkString("\n")
//传入user=196被推荐用户ID，num=5 推荐数, 返回Array[Rating]
//2.针对用户 推荐产品的系统评分
model.predict(196,464) //系统对用户196推荐产品464的评分
//3.针对电影 推荐给用户
//当要促销某些电影时，可以找出可能对这些电影感兴趣的会员 使用model.recommendUsers(product:Int,num:Int):Array[Rating] 方法推荐
product要被推荐的电影ID，num推荐的记录数，返回系统针对产品推荐给用户的数组
model.recommendUser(464,5).mkString(“,”)
//把464推荐给用户471，系统评分 13.0232...
显示推荐电影名称
//1 创建电影ID-名称的对照表
val itemrdd=sc.textFile("file:/home/hadoop/ALS/u.item")
val moviet=itemrdd.map(_.split("\\|").take(2)).map(x=>(x(0),x(1))).collectAsMap
//map(x=>(x(0),x(1))).collect().toMap
//2显示对照表前5条数据
moviet.take(5).foreach(println)
//查询电影名称
scala> moviet(590)
res114: String = Hellraiser: Bloodline (1996)
//2 显示前5条推荐电影名称
model.recommendProducts(195,5).map(rating=>(rating.product,moviet(rating.product),rating.rating)).foreach(println)
Recommend项目代码
import java.io.File
import scala.io.Source
import org.apache.spark.{SparkContext,SparkConf}
import org.apache.spark.rdd._
import org.apache.spark.mllib.recommendation.{ALS,Rating,MatrixFactorizationModel}
main程序
val (ratingrdd,moviet)=PrepareData() //产生ratingrdd评分数据 与 moviet(id,title)
val model=ALS.train(ratingrdd,5,20,0.1) //训练模型
recommend(model,moviet) //推荐
//针对用户推荐电影
def RecommendMovies(model:MatrixfactorizationModel,movieTitle:Map[Int,String],inputUserID:Int)={
val RecomendMovie=model.recommendProducts(inputUserID,10)
RecomendMovie.foreach{r=>pritnln(i.toString+moviet(r.product)+r.rating); i+=1 }
}
//针对电影推荐用户
def RecommendUsers(model:MatrixfactorizationModel,movieTitle:Map[Int,String],inputMovieID:Int)={
val RecomendUser=model.recommendUsers(inputMovieID,10)
RecomendUser.foreach{r=>pritnln(i.toString+r.user+r.rating); i+=1 }
}
评估模型ALSEvaluation.main
调校model参数
model的rank，iterator，lambda这些参数设置会影响结果的准确度，运行时间。下面调校找出最佳参数组合
import java.io.File
import scala.io.Source
import org.apache.spark.{SparkContext,SparkConf}
import org.apache.spark.rdd._
import org.apache.spark.mllib.recommendation.{ALS,Rating,MatrixFactorizationModel}
import org.apache.spark.mllib.regression.LabeledPoint
def main(arg:Array[String]):Unit={
//1 准备数据阶段
val (trainData,validationData,testData)=prepareData()
trainData.persist();validationData.persist();testData.persist()
//2 trainData训练model，validationData评估模型的RMSE均方根误差
//参数设置会影响误差，需反复执行训练和评估找出最佳组合，最后返回
//bestModel模型进入下一阶段测试
val bestModel=trainValidation(trainData,validationData)
//3 使用testData在测试一次，避免出现overfitting（训练过度）.如果rmse在trainData训练//和testData测试差异不大代表无过度
val testRmse=computeRMSE(bestModel,testData)
println(“使用testData测试bestModel，结果rmse=”+testRmse)
trainData.unpersist();validationData.unpersist();testData.unpersist()
}
1.准备数据-prepareData
def prepareData():(RDD[Rating],RDD[Rating],RDD[Rating])={
val Array(trainData,validationData,testData)=ratingrdd.randomSplit(Array(0.8,0.1,0.1))
return (trainData,validationData,testData)
}
2.进行训练评估-trainvalidation
def trainvalidation(trainData:RDD[Rating],validationData:RDD[Rating]):MatrixFactorizationModel={
//评估rank
evaluateParameter(trainData,validationData,”rank”,Array(5,10,15,20,50,100),Array(10),Array(0.1)
//评估iter
evaluateParameter(trainData,validationData,”numIterators”,Array(10),Array(5,10,15,20,25),Array(0.1));
//评估lambda
evaluateParameter(trainData,validationData,”lambda”,Array(10),Array(10),Array(0.05,0.1,1,5,10.0));
//所有参数找出最好组合
val bestModel=evaluateAllParameter(trainData,validationData, Array(5,10,15,20,50,100), Array(5,10,15,20,25), Array(0.05,0.1,1,5,10.0));
return (bestModel)
}
//以上3个参数有5*5*5=125个组合，对每一种参数组合评估Rmse 最好找出具有最小误差Rmse的参数组合就是最佳组合。
1.评估单个参数-evaluateParameter
那个参数具有比较低的误差，绘制图形
def evaluateParameter(trainData,validationData,p:String,ranks:Array[Int],its:Array[String],lambdas:Array[Double]):(Double,String)={
for(rank<-ranks;it<-its;lambda<-lambdas){
val rmse=trainModel(trainData,validationData,rank,it,lambda)
val pData=p match{
case “rank”=>”rank:”+rank; //pData取到for中的值
case “it”=>”it:”+it;
case “lambda”=>”lambda:”+lambda;
}
}
(rmse,pData)
}
-----------包含画图
def evaluateParameter(trainData,validationData,p:String,ranks:Array[Int],its:Array[String],lambdas:Array[Double])={
var databarchart=new DefaultCategoryDataset();
var datalinechart=new DefaultCategoryDataset();
for(rank<-ranks;it<-its;lambda<-lambdas){
val rmse=trainModel(trainData,validationData,rank,it,lambda)
val pData=p match{
case “rank”=>rank; //pData取到for中的值
case “it”=>it;
case “lambda”=>lambda;
}
databarchart.addValue(rmse,p,pData.toString)
datslinechart.addValue(time,”Time”,pData.toString)
}
Chart.plotBarLineChart(“ALS evaluations ”+p,p,”RMSE”,0.58,5,”Time”,databarchart,datalinechart);
}
def evaluateAllParameter(trainData:RDD[Rating],validationData:RDD[Rating],ranks:Array[Int],its:Array[Int],lambdas:Array[Double]):(Double,String):MatrixFactorizationModel={
val evaluations=for(rank<-ranks;it<-its;lambda<-lambdas)
yield{
val rmse=trainModel(trainData,validationData,rank,it,lambda)
(rank,it,lambda,rmse)
}
val Eval=(evaluations.sortBy(_._4))
val BestEval=Eval(0)
println(“最佳model参数 rank=”+BestEval._1+”,it=”+BestEval._2+” lambda=”+BestEval._3)
val bestModel=ALS.train(trainData,BestEval._1,BestEval._2,BestEval._3)
(bestModel)
}
2.训练模型-trainModel
def trainModel(trainData:RDD[Rating], validationData:RDD[Rating],rank:Int,it:Int,lambda:Double):Double={
val model=ALS.train(trainData,rank,it,lambda)
val rmse=computeRmse(model,validationData)
(rmse)
}
3.计算RMSE- computeRmse
root mean square error用来计算推荐系统对用户喜好的预测与实际喜好的误差平均值
误差越小代表预测值与真实值越接近
公式
def computeRmse(model:MatrixFactorizationModel,validationData:RDD[Rating]):Double={
val num= validationrdd.count
val predictedRDD=model.predict(validationrdd.map(r=>(r.user,r.product)))
//取出user+product传入预测方法，预测将结果存入predictRDD.
val predictedAndRatings=predictedRDD.map(p=>((p.user,p.product),p.rating))
.join(validationrData.map(r=>((r.user,r.product),r.rating))).values
//predictRDD预测集join ratingrdd真实集
math.sqrt(predictedAndRatings.map(x=>(x._1-x._2)*(x._1-x._2)).reduce(_+_)/num)
//针对每条数据进行计算，传入x。x._1预测结果-x._2真实数据，相乘就是平方
//reduct将全部误差求和后，在除以个数num。最后计算sqrt平方根
}
3.运行ALSEvaluation
评估rank参数 看到对RMSE没有太大差别，但是越大运行时间会增加
评估it参数 对RMSE没有太大差别，时间会增加
评估lambda参数 0.1时 RMSE最小，所需时间差异不大
所有参数交叉评估找出最好参数组合
运行evaluationAllParameter后显示
最佳model的参数组合：rank20,it15,lamdba0.1 结果rmse=0.72312434463...
最好使用testData再次验证bestModel结果0.7191119923578929，确认无overfitting问题
相同的validationData，在model中
没优化得到RMSE值0.7745372...
优化后得到RMSE值0.7134127699...
最后 找出最佳参数组合，可以修改Recommend.scala为最佳组合
val model=ALS.train(ratings,20,15,0.1)
附录：Chart
import org.jfree.chart._
import org.jfree.data.xy._
import org.jfree.data.category.DefaultCategoryDataset
import java.awt.Color
import java.awt.BasicStroke
object Chart{
def plotBarLineChart(Title:String,xLabel:String,yBarLabel:String,yBarMin:Double,yBarMax:Double,yLineLabel:String,dataBarChart:DefaultCategoryDataset,dataLineChart:DefaultCategoryDataset):Unit={
//画出Bar chart
val chart=ChartFactory.createBarChart(“”,xLabel,yBarLabel,
dataBarChart, //bar chart数据
PlotOrientation.VERTICAL, //画图方向垂直
true, //包含legend
true, //显示tooltips
false //不要url generator
)
val plot=chart.getCategoryPlot();
plot.setBackgroundPaint(new Color(0xFF,0xFF,0xFF))
plot.setDomainAxisLocation(AxisLocation.BOTTOM_OR_RIGHT)
plot.setDataset(1,dataLineChart);
plot.mapDatasetToRangeAxis(1,1)
//画柱状图 Y轴
val vn=plot.getRangeAxis();vn.setRange(yBarMin,yBarMax);vn.setAutoTickUnitSelection(true)
//画折线图y轴
val axis2=new NumberAxis(yLineLabel);plot.setRangeAxis(1,axis2);
val renderer2=new LineAndShapeRender();
renderer2.setToolTipGenerator(new StandardCategoryToolTipGenerator());
//先画柱形图，再画折线图
plot.setRenderer(1,renderer2);plot.setDatasetRenderingOrder(DatasetRenderingOrder.FORWARD);
//创建画框
val frame=new ChartFrame(Title,chart); frame.setSize(500,500);
frame.pack();frame.setVisible(true);
}
}