大数据技术Spark MLlib机器学习协同过滤电影推荐附源码和数据集

需要源码和数据集请点赞关注收藏后评论区留言私信~~~

协同过滤————电影推荐

协同过滤是利用大量已有的用户偏好来估计用户对其未接触过的物品的喜好程度。在协同过滤算法中有着两个分支，分别是基于群体用户的协同过滤（UserCF）和基于物品的协同过滤（ItemCF）。

在电影推荐系统中，通常分为针对用户推荐电影和针对电影推荐用户两种方式。若采用基于用户的推荐模型，则会利用相似用户的评级来计算对某个用户的推荐。若采用基于电影的推荐模型，则会依靠用户接触过的电影与候选电影之间的相似度来获得推荐。

在Spark MLlib实现了交替最小二乘（ALS）算法，它是机器学习的协同过滤式推荐算法，机器学习的协同过滤式推荐算法是通过观察所有用户给产品的评分来推断每个用户的喜好，并向用户推荐合适的产品

1:下载数据集

数据集网址

2：建立Scala类

推荐结果如下

部分代码如下 全部代码请点赞关注收藏后评论区留言私信

1.  
   import java.util.Random
2.  
    
3.  
   import org.apache.spark.SparkConf
4.  
   import org.apache.spark.SparkContext
5.  
   import org.apache.spark.rdd._
6.  
   import org.apache.spark.mllib.recommendation.{ALS, Rating, MatrixFactorizationModel}
7.  
    
8.  
   object MovieRecomment {
9.  
   def main(args: Array[String]): Unit = {
10.  
     
11.  
     
12.  
    //建立spark环境
13.  
    // val conf = new SparkConf().setAppName("movieRecomment")
14.  
    val conf = new SparkConf().setAppName("MovieLensALS").setMaster("local[4]")
15.  
    val sc = new SparkContext(conf)
16.  
    sc.setLogLevel("error")
17.  
    //装载样本评分数据，其中最后一列Timestamp取除10的余数作为key，Rating为值,即(Int,Rating)
18.  
    val ratings = sc.textFile("ratings.dat").map {
19.  
    line =>
20.  
    val fields = line.split("::")
21.  
    //时间戳、用户编号、电影编号、评分
22.  
    (fields(3).toLong % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble))
23.  
    }
24.  
    ratings.take(10).foreach { println }
25.  
    // 装载电影目录对照表（电影ID->电影标题）
26.  
    val movies = sc.textFile("movies.dat").map { line =>
27.  
    val fields = line.split("::")
28.  
    (fields(0).toInt, fields(1))
29.  
    }.collect.toMap
30.  
     
31.  
    //记录样本评分数、用户数、电影数
32.  
    val numRatings = ratings.count
33.  
    val numUsers = ratings.map(_._2.user).distinct.count
34.  
    val numMovies = ratings.map(_._2.product).distinct.count
35.  
    println("从" numRatings "记录中" "分析了" numUsers "的人观看了" numMovies "部电影")
36.  
    //提取一个得到最多评分的电影子集(电影编号），以便进行评分启发
37.  
    //矩阵最为密集的部分
38.  
    val mostRatedMovieIds = ratings.map(_._2.product)
39.  
    .countByValue()
40.  
    .toSeq
41.  
    .sortBy(-_._2)
42.  
    .take(50) //取前50个
43.  
    .map(_._1) //获取他们的id
44.  
    val random = new Random(0)
45.  
    //从目前最火的电影中随机获取十部电影
46.  
    val selectedMovies = mostRatedMovieIds.filter(
47.  
    x => random.nextDouble() < 0.2).map(x => (x, movies(x))).toSeq
48.  
    //调用函数 elicitateRatings,引导或者启发评论,让用户打分
49.  
    val myRatings = elicitateRatings(selectedMovies)
50.  
    val myRatingsRDD = sc.parallelize(myRatings)
51.  
    //将评分数据分成训练集60%，验证集20%，测试集20%
52.  
    val numPartitions = 20
53.  
    //训练集
54.  
    val training = ratings.filter(x => x._1 < 6).values
55.  
    .union(myRatingsRDD).repartition(numPartitions)
56.  
    .persist
57.  
    //验证集
58.  
    val validation = ratings.filter(x => x._1 >= 6 && x._1 < 8).values
59.  
    .repartition(numPartitions).persist
60.  
    //测试集
61.  
    val test = ratings.filter(x => x._1 >= 8).values.persist
62.  
    val numTraining = training.count
63.  
    val numValidation = validation.count
64.  
    val numTest = test.count
65.  
    println("训练集数量:" numTraining ",验证集数量: " numValidation ", 测试集数量:" numTest)
66.  
    //训练模型，并且在验证集上评估模型
67.  
    //模型中隐藏因子数
68.  
    val ranks = List(8, 12)
69.  
    //正则项的惩罚系数
70.  
    val lambdas = List(0.1, 10.0)
71.  
    //循环次数
72.  
    val numIters = List(10, 20)
73.  
    var bestModel: Option[MatrixFactorizationModel] = None
74.  
    var bestValidationRmse = Double.MaxValue
75.  
    var bestRank = 0
76.  
    var bestLambda = -1.0
77.  
    var bestNumIter = -1
78.  
    for (rank <- ranks; lambda <- lambdas; numIter <- numIters) {
79.  
    val model = ALS.train(training, rank, numIter, lambda)
80.  
    val validationRmse = computeRmse(model, validation, numValidation)
81.  
    println("RMSE (validation)=" validationRmse "for the model trained with rand =" rank ", lambda=" lambda ", and numIter= " numIter ".")
82.  
    if (validationRmse < bestValidationRmse) {
83.  
    bestModel = Some(model)
84.  
    bestValidationRmse = validationRmse
85.  
    bestRank = rank
86.  
    bestLambda = lambda
87.  
    bestNumIter = numIter
88.  
    }
89.  
    }
90.  
    //在测试集中 获得最佳模型
91.  
    val testRmse = computeRmse(bestModel.get, test, numTest)
92.  
    println("The best model was trained with rank=" bestRank " and lambda =" bestLambda ", and numIter =" bestNumIter ", and itsRMSE on the test set is" testRmse ".")
93.  
    //产生个性化推荐
94.  
    val myRateMoviesIds = myRatings.map(_.product).toSet
95.  
    val candidates = sc.parallelize(movies.keys.filter(!myRateMoviesIds.contains(_)).toSeq)
96.  
    val recommendations = bestModel.get.predict(candidates.map((0, _)))
97.  
    .collect()
98.  
    .sortBy((-_.rating))
99.  
    .take(10)
100.  
     var i = 1
101.  
     println("以下电影推荐给你")
102.  
     recommendations.foreach { r =>
103.  
     println("-".format(i) ":" movies(r.product))
104.  
     i = 1
105.  
      
106.  
     }
107.  
     }
108.  
     /**计算 均方根误差RMSE */
109.  
     def computeRmse(model: MatrixFactorizationModel, data: RDD[Rating], n: Long) = {
110.  
     val predictions: RDD[Rating] = model.predict(data.map(x => (x.user, x.product)))
111.  
     val predictionsAndRatings = predictions.map(x => ((x.user, x.product), x.rating))
112.  
     .join(data.map(x => ((x.user, x.product), x.rating)))
113.  
     .values
114.  
     math.sqrt(predictionsAndRatings.map(x => (x._1 - x._2) * (x._1 - x._2)).reduce(_ _) / n)
115.  
     }
116.  
     /** 从命令行获取用户电影评级. */
117.  
     def elicitateRatings(movies: Seq[(Int, String)]) = {
118.  
     val prompt = "给以下电影评分（1——5分）"
119.  
     println(prompt)
120.  
     val ratings = movies.flatMap { x =>
121.  
     var rating: Option[Rating] = None
122.  
     var valid = false
123.  
     while (!valid) {
124.  
     print(x._2 ": ")
125.  
     try {
126.  
     val r = Console.readInt
127.  
     if (r < 0 || r > 5) {
128.  
     println(prompt)
129.  
     } else {
130.  
     valid = true
131.  
     if (r > 0) {
132.  
     rating = Some(Rating(0, x._1, r))
133.  
     }
134.  
     }
135.  
     } catch {
136.  
     case e: Exception => println(prompt)
137.  
     }
138.  
     }
139.  
     rating match {
140.  
     case Some(r) => Iterator(r)
141.  
     case None => Iterator.empty
142.  
     }
143.  
     }
144.  
     if (ratings.isEmpty) {
145.  
     error("No rating provided!")
146.  
     } else {
147.  
     ratings
148.  
     }
149.  
      
150.  
     }
151.  
     }

 创作不易 觉得有帮助请点赞关注收藏~~~

这篇好文章是转载于：学新通技术网