spark练习——影评案例

第一次写博客，新人上路，欢迎大家多多指教！！！

---------------------------------------------------------------------分割线---------------------------------------------------------------------

现有如此三份数据：
1、users.dat 数据格式为： 2::M::56::16::70072
对应字段为：UserID BigInt, Gender String, Age Int, Occupation String, Zipcode String
对应字段中文解释：用户 id，性别，年龄，职业，邮政编码
2、movies.dat 数据格式为： 2::Jumanji (1995)::Adventure|Children's|Fantasy
对应字段为：MovieID BigInt, Title String, Genres String
对应字段中文解释：电影 ID，电影名字，电影类型
3、ratings.dat 数据格式为： 1::1193::5::978300760
对应字段为：UserID BigInt, MovieID BigInt, Rating Double, Timestamped String
对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳

需求：

1、求被评分次数最多的 10 部电影，并给出评分次数（电影名，评分次数）
2、分别求男性，女性当中评分最高的 10 部电影（性别，电影名，影评分）
3、分别求男性，女性看过最多的 10 部电影（性别，电影名）
4、年龄段在“18-24”的男人，最喜欢看 10 部电影
5、求 movieid = 2116 这部电影各年龄段（因为年龄就只有 7 个，就按这个 7 个分就好了）
的平均影评（年龄段，影评分）
6、求最喜欢看电影（影评次数最多）的那位女性评最高分的 10 部电影的平均影评分（观影
者，电影名，影评分）
7、求好片（评分>=4.0）最多的那个年份的最好看的 10 部电影
8、求 1997 年上映的电影中，评分最高的 10 部 Comedy 类电影
9、该影评库中各种类型电影中评价最高的 5 部电影（类型，电影名，平均影评分）
10、各年评分最高的电影类型（年份，类型，影评分）

先建立一个Utils类，主要用于初始化配置信息以及解析原始数据

package movie_ratingimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object Utils {//初始化SparkConf对象private[movie_rating] val conf = new SparkConf().setAppName("FileReview").setMaster("local")//初始化sc对象private[movie_rating]  val sc = new SparkContext(conf)sc.setLogLevel("ERROR")//读取hdfs上的数据private[movie_rating] val movie = sc.textFile("hdfs://myha01/mydata/film_review/movies.dat")private[movie_rating] val ratings = sc.textFile("hdfs://myha01/mydata/film_review/ratings.dat")private[movie_rating] val users = sc.textFile("hdfs://myha01/mydata/film_review/users.dat")//将原始数据转为RDD格式private[movie_rating] val movieRdd: RDD[(String, String, String)] = movie.map(_.split("::")).map(m => (m(0), m(1), m(2)))private[movie_rating] val ratingsRdd: RDD[(String, String, String, String)] = ratings.map(_.split("::")).map(r => (r(0), r(1), r(2), r(3)))private[movie_rating] val usersRdd: RDD[(String, String, String, String, String)] = users.map(_.split("::")).map(u => (u(0), u(1), u(2), u(3), u(4)))}

 第一问：

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand01 {/*** 1、求被评分次数最多的 10 部电影，并给出评分次数（电影名，评分次数）*/def main(args: Array[String]): Unit = {//获取电影id与对应的评分次数val movieID_rating: RDD[(String, Int)] = Utils.ratingsRdd.map(x => (x._2, 1))val movieID_times: RDD[(String, Int)] = movieID_rating.reduceByKey(_ + _).sortBy(_._2, false)//获得电影id和电影名val movieID_name: RDD[(String, String)] = Utils.movieRdd.map(x => (x._1, x._2))//关联movieID_times和movieID_name，获得电影id，电影名，评分次数val result: RDD[(String, Int)] = movieID_times.join(movieID_name).sortBy(_._2._1, false).map(x => (x._2._2, x._2._1))//输出结果result.take(10).foreach(println(_))}
}

第二问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand02 {/*** 2、分别求男性，女性当中评分最高的 10 部电影（性别，电影名，影评分）*/def main(args: Array[String]): Unit = {//(userID, sex)val userID_sex: RDD[(String, String)] = Utils.usersRdd.map(x => (x._1, x._2))//(userID, (movieID, rating))val userID_movieID_rating: RDD[(String, (String, String))] = Utils.ratingsRdd.map(x => (x._1, (x._2, x._3)))//(userID, (sex, (movieID, rating)))  ---> (sex, movieID, rating)val movieID_rating: RDD[(String, String, String)] = userID_sex.join(userID_movieID_rating).map(x => (x._2._1, x._2._2._1, x._2._2._2))//((sex, movieID), Iterable[(sex, movieID, rating)])  ---> (movieID, (sex, avg))val movieID_sex_avg: RDD[(String, (String, Double))] = movieID_rating.groupBy(x => (x._1, x._2)).map(x => {var sum, avg = 0dval list: List[(String, String, String)] = x._2.toListif (list.size > 50) {list.map(x => ( sum += x._3.toInt ))avg = sum * 1.0 / list.size}(x._1._2, (x._1._1, avg))})//(movieID, movieName)val movieID_movieName: RDD[(String, String)] = Utils.movieRdd.map(x => (x._1, x._2))//sex_movieID_avg与movie进行关联 (movieID, ((sex, avg), movieName)) ---> (sex, movieName, avg)val sex_movieName_avg: RDD[(String, String, Double)] = movieID_sex_avg.join(movieID_movieName).map(x => (x._2._1._1, x._2._2, x._2._1._2)).sortBy(x => (x._1, x._3), false)sex_movieName_avg.take(10).foreach(println(_))sex_movieName_avg.filter(_._1 == "F").take(10).foreach(println(_))}
}

第三问：

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand03 {/*** 3、分别求男性，女性看过最多的 10 部电影（性别，电影名）*/def main(args: Array[String]): Unit = {//(userID, sex)val userID_sex: RDD[(String, String)] = Utils.usersRdd.map(x => (x._1, x._2))//(userID, movieID)val userID_movieID: RDD[(String, String)] = Utils.ratingsRdd.map(x => (x._1, x._2))//(movieID, name)val movieID_name: RDD[(String, String)] = Utils.movieRdd.map(x => (x._1, x._2))//(userID, (sex, movieID))  ---> (movieID, sex)val movieID_sex: RDD[(String, String)] = userID_sex.join(userID_movieID).map(x => (x._2._2, x._2._1))//关联movieID_sex和movieID_name    (movieID, (sex, name))  ---> (movieID, sex, name)val movieID_sex_name: RDD[(String, String, String)] = movieID_sex.join(movieID_name).map(x => (x._1, x._2._1, x._2._2))//((sex, name), Iterable[(movieID, sex, name)])  ---> (sex, name, times)val sex_name_times: RDD[(String, String, Int)] = movieID_sex_name.groupBy(x => (x._2, x._3)).map(x => (x._1._1, x._1._2, x._2.toList.size)).sortBy(x => (x._1, x._3), false)//输出结果sex_name_times.take(10).foreach(println(_))sex_name_times.filter(_._1 == "F").take(10).foreach(println(_))}
}

第四问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand04 {/*** 4、年龄段在“18-24”的男人，最喜欢看 10 部电影（输出电影id和电影名字）*/def main(args: Array[String]): Unit = {// 年龄段在“18-24”的男人的userID     (userID, (sex, age))val userID_sex_age: RDD[(String, (String, Int))] = Utils.usersRdd.map(x => (x._1, (x._2, x._3.toInt))).filter(x =>{x._2._2 >= 18 && x._2._2 <= 24 && x._2._1 == "M"} )//(userID, (movieID, rating))val userID_movieID_rating: RDD[(String, (String, Int))] = Utils.ratingsRdd.map(x => (x._1, (x._2, x._3.toInt)))//关联userID与userID_movieID_rating    (userID, ((sex, age), (movieID, rating)))   ---> (movieID, rating)// --->(movieID, Iterable(movieID, rating))  ---> (movieID, avg)val movieID_avg : RDD[(String, Double)] = userID_sex_age.join(userID_movieID_rating).map(x => (x._2._2._1, x._2._2._2)).groupByKey().map(x => {var avg = 0dval len: Int = x._2.sizeif (len > 50){avg = 1.0 * x._2.sum / len}(x._1, avg)})//(movieID, name)val movieID_name: RDD[(String, String)] = Utils.movieRdd.map(x => (x._1, x._2))//关联movieID_avg与movieID_name    (movieID, (avg, name))val name_avg: RDD[(String, Double)] = movieID_avg.join(movieID_name).map(x => (x._2._2, x._2._1)).sortBy(_._2, false)//输出结果name_avg.take(10).foreach(println(_))}
}

第五问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand05 {/*** 5、求 movieid = 2116 这部电影各年龄段（因为年龄就只有 7 个，就按这个 7 个分就好了）* 的平均影评（年龄段，影评分）*/def main(args: Array[String]): Unit = {// 获得movieID = 2116  (userID, rating)val userID_rating: RDD[(String, Int)] = Utils.ratingsRdd.filter(_._2 == "2116").map(x => (x._1, x._3.toInt))//(userID, age)val userID_age: RDD[(String, String)] = Utils.usersRdd.map(x => (x._1, x._3))//关联userID_age和userID_rating   (userID, (age, rating)) --->(age, rating)  ---> (age, Iterable(rating))val age_avg: RDD[(String, Double)] = userID_age.join(userID_rating).map(x => (x._2._1, x._2._2)).groupByKey().map(x => (x._1, x._2.sum * 1.0 / x._2.size))//输出结果age_avg.sortByKey().foreach(println(_))}
}

第六问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand06 {/*** 6、求最喜欢看电影（影评次数最多）的那位女性评最高分的 10 部电影的平均影评分* （观影者userID，电影名，影评分)*/def main(args: Array[String]): Unit = {//(userID, Iterable(userID, movieID, rating, time_stamp))  ---> (userID, times)val userID_times: RDD[(String, Int)] = Utils.ratingsRdd.groupBy(_._1).map(x => (x._1, x._2.size))//(userID, (sex, times))找到最喜欢看电影（影评次数最多）的那位女性的userIDval userID: String = Utils.usersRdd.map(x => (x._1, x._2)).join(userID_times).filter(_._2._1 == "F").sortBy(_._2._2, false).map(_._1).first()//获得userID用户评分最高的10部电影的movieIDval movieID: Array[(String, Int)] = Utils.ratingsRdd.filter(_._1 == userID).map(x => (x._2, x._3.toInt)).sortBy(_._2, false).take(10)//获得该10部电影的平均影评分val movieID_rating: RDD[(String, String)] = Utils.ratingsRdd.map(x => (x._2, x._3))//关联movieID和movieID_rating   (movieID, (rat1, rating))  ---> (movieID, Iterable(rating))  --> (movieID, avg)val movieID_avg = Utils.sc.makeRDD(movieID).join(movieID_rating).map(x => (x._1, x._2._2.toInt)).groupByKey().map(x => {var avg = 0dif (x._2.size >= 50) {avg = x._2.sum * 1.0 / x._2.size}(x._1, avg)})//(movieID, (name, avg))   ---> (UserID, name, avg)val userID_name_avg: RDD[(String, String, Double)] = Utils.movieRdd.map(x => (x._1, x._2)).join(movieID_avg).map(x => (userID, x._2._1, x._2._2)).sortBy(_._3, false)userID_name_avg.foreach(println(_))}
}

第七问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand07 {/*** 7、求好片（评分>=4.0）最多的那个年份的最好看的 10 部电影（电影id， 电影名，平均评分）*/def main(args: Array[String]): Unit = {//1、找到所有的好片的movieID//(movieID, rating) ---> (movieID, Iterable(rating))  ---> (movieID, avg)（avg >= 4.0）val movieID_avg :RDD[(String, Double)]= Utils.ratingsRdd.map(x => (x._2, x._3.toInt)).groupByKey().map(x =>{var avg = 0dif(x._2.size >= 50)avg = x._2.sum * 1.0 / x._2.size(x._1, avg)}).filter(_._2 >= 4.0)//(movieID, (name, year))val movieID_name_year: RDD[(String, (String, String))] = Utils.movieRdd.map(x => (x._1, (x._2, x._2.substring(x._2.length - 5, x._2.length - 1))))//2、找到好片最多的年代//关联movieID_avg与movieID_name_year，(movieID, (avg, (name, year)))   --> (year, Iterable(movieID))val year_count: (String, Int) = movieID_avg.join(movieID_name_year).map(x => (x._2._2._2, x._1)).groupByKey().map(x => (x._1, x._2.size)).sortBy(_._2, false).first()//3、找到该年最好看的10部电影//(movieID, name) ---> (movieID, (name, avg))  ---> (movieID, name, avg)val movieID_name_avg = movieID_name_year.filter(_._2._2 == year_count._1).map( x => (x._1, x._2._1)).join(movieID_avg).map(x => (x._1, x._2._1, x._2._2)).sortBy(_._3, false).take(10)//输出结果movieID_name_avg.foreach(println(_))}
}

第八问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand08 {/*** 8、求 1997 年上映的电影中，评分最高的 10 部 Comedy 类电影（电影id，电影名字，类型，平均评分）*/def main(args: Array[String]): Unit = {//(movieID, (name, year, type))val movieID_name_year_type: RDD[(String, (String, String, String))] = Utils.movieRdd.map(x => (x._1, (x._2, x._2.substring(x._2.length - 5, x._2.length - 1), x._3)))//找到所有1997年的comedy类型的电影 (movieID, (name, 1997, comedy))val movieID_name_1997_comedy: RDD[(String, (String, String, String))] = movieID_name_year_type.filter(x => {x._2._2 == "1997" && x._2._3.toLowerCase.contains("comedy")} )//(movieID, (rating, (name, 1997, comedy)))  ---> (movieID, (name, comedy, rating))val movieID_name_comedy_rating: RDD[(String, (String, String, String))] = Utils.ratingsRdd.map(x => (x._2, x._3)).join(movieID_name_1997_comedy).map(x => (x._1, (x._2._2._1, x._2._2._3, x._2._1)))//(movieID, Iterable(rating))  ---> (movieID, avg)val movieID_avg: Array[(String, Double)] = movieID_name_comedy_rating.map(x => (x._1, x._2._3.toInt)).groupByKey().map(x => {var avg = 0dif (x._2.size >= 50)avg = x._2.sum * 1.0 / x._2.size(x._1, avg)}).distinct().sortBy(_._2, false).take(10)//(movieID, (avg, (name, comedy, rating)))  ---> (movieID, name, comedy, avg)val movieID_name_comedy_avg: RDD[(String, String, String, Double)] = Utils.sc.makeRDD(movieID_avg).join(movieID_name_comedy_rating).map(x => (x._1, x._2._2._1, x._2._2._2, x._2._1)).distinct().sortBy(_._4, false)//输出结果movieID_name_comedy_avg.foreach(println(_))}
}

第九问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand09 {/*** 9、该影评库中各种类型电影中评价最高的 5 部电影（类型，电影名，平均影评分）*/def main(args: Array[String]): Unit = {//获得所有电影的movieID，name，types      (movieID, (name, types))val movieID_name_types: RDD[(String, (String, String))] = Utils.movieRdd.map(x => (x._1, (x._2, x._3)))//获得所有的movieID，rating (movieID, rating)val movieID_rating: RDD[(String, String)] = Utils.ratingsRdd.map(x => (x._2, x._3))//关联movieID_name_types与movieID_rating   (movieID, ((name, types), rating))  ---> (types, name, rating)val types_name_rating: RDD[((String, String), Int)] = movieID_name_types.join(movieID_rating).map(x => ((x._2._1._2, x._2._1._1), x._2._2.toInt))//((types, name), Iterable(rating))  ---> (types, name, avg)val types_name_avg: RDD[(String, String, Double)] = types_name_rating.groupByKey().map(x => {var avg = 0dif (x._2.size >= 50)avg = x._2.sum * 1.0 / x._2.size(x._1._1, x._1._2, avg)})//(types, name, avg)     划分types：将Action|Adventure|Comedy|Sci-Fi拆开var tempArray: Array[(String, String, Double)] = Array(("", "", 0d))types_name_avg.collect().foreach(x => {//Action|Adventure|Comedy|Sci-Fi   ---> Arrays(Action, Adventure, Comedy, Sci-Fi)val types: Array[String] = x._1.split("\\|")//将所有的types_name_avg中的元素拆分后存于tempArray数组中tempArray = types.map((_, x._2, x._3)).union(tempArray)})//(type, name, avg)  包含所有类型电影的排序val type_name_avg = Utils.sc.makeRDD(tempArray).filter(_._3 > 0).sortBy(x => (x._1, x._3), false)//(type, Iterable(type, name, avg))  打印前五type_name_avg.groupBy(_._1).sortByKey().foreach(x => {var count = 0val list: List[(String, String, Double)] = x._2.toListwhile(count < list.size  && count < 5){println(list(count))count += 1}println()})}
}

第十问

package movie_ratingimport org.apache.spark.rdd.RDD/*** Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码* Utils.movieRdd：对应字段中文解释：电影 ID，电影名字，电影类型* Utils.ratingsRdd：对应字段中文解释：用户 ID，电影 ID，评分，评分时间戳*/
object Demand10 {/*** 10、各年评分最高的电影类型（年份，类型，影评分）*/def main(args: Array[String]): Unit = {//(movieID, year)val movieID_year: RDD[(String, String)] = Utils.movieRdd.map(x => (x._1, (x._2.substring(x._2.length - 5, x._2.length - 1))))//(movieID, rating)  ---> (movieID, Iterable(rating)) ---> (movieID, avg)val moviID_avg: RDD[(String, Double)] = Utils.ratingsRdd.map(x => (x._2, x._3.toDouble)).groupByKey().map(x => (x._1, x._2.sum / x._2.size))//关联movieID_year和moviID_avg   (movieID, (year, avg)) ---> (year, (movieID, avg))val year_mocvieID_avg: RDD[(String, (String, Double))] = movieID_year.join(moviID_avg).distinct().map(x => (x._2._1, (x._1, x._2._2)))//(year, (movieID, avg))  ---> (year, Iterable((movieID, avg)))  ---> (movieID, (year, topavg))val year_movieID_topavg: RDD[(String, (String, Double))] = year_mocvieID_avg.groupByKey().map(x => {val list: List[(String, Double)] = x._2.toList.sortBy(-_._2)(list(0)._1, (x._1, list(0)._2))})//(movieID, (type, (year, topavg)) ---> (year, type, topavg)val year_type_topavg: RDD[(String, String, Double)] = Utils.movieRdd.map(x => (x._1, x._3)).join(year_movieID_topavg).map(x => (x._2._2._1, x._2._1, x._2._2._2)).sortBy(_._1, false)//输出结果year_type_topavg.foreach(println(_))}
}

转载于:https://www.cnblogs.com/wang-bing/p/9129483.html

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！