基于Spark实现电影点评系统用户行为分析

文章目录

1、项目背景
2、数据描述
3、代码实现

1、项目背景

电影推荐系统（MovieLens）是美国明尼苏达大学（Minnesota）计算机科学与工程学院的GroupLens项目组创办的，是一个非商业性质的、以研究为目的的实验性站点。电影推荐系统注要使用协同过滤和关联规则相结合的技术，向用户推荐他们感兴趣的电影。本项目的数据集来源：https://grouplens.org/datasets/movielens/
需求：

统计电影中平均得分最高（口碑最好）的电影及观看人数最高的电影（流行度最高）TopN。
统计最受男性喜爱的电影TopN和最受女性喜爱的电影TopN。

2、数据描述

本项目总4个数据集，分别是评级文件ratings.dat、用户文件users.dat、电影文件movies.dat、职业文件occupations.dat。
1）所有评级都包含在“ratings.dat”文件中，格式如下：

UserID::MovieID::Rating::Timestamp
- UserID的范围在1到6040之间
- MovieID的范围在1到3952之间
- 评级为5星级（仅限全星评级）
- 时间戳以秒为单位表示
- 每个用户至少有20个评级

评级文件ratings.dat中摘取部分记录如下：

1::1193::5::978300760
1::661::3::978302109
1::914::3::978301968
1::3408::4::978300275
1::2355::5::978824291
1::1197::3::978302268

2）所有用户信息都包含在“users.dat”文件中，格式如下：

UserID::Gender::Age::Occupation::Zip-code所有人口统计信息均由用户自愿提供，不会检查其准确性。此数据集中仅包含已提供某些人口统计信息的用户。性别用男性表示“M”，女性表示“F”表示
年龄选自以下范围：1：“18岁以下”
18：“18-24”
25：“25-34”
35：“35-44”
45：“45-49”
50：“50-55”
56：“56+”

用户文件users.dat中摘取部分记录如下：

1::F::1::10::48067
2::M::56::16::70072
3::M::25::15::55117
4::M::45::7::02460
5::M::25::20::55455

3）所有电影信息都包含在“movies.dat”文件中，格式如下：

MovieID::Title::Genres标题与IMDB提供的标题相同（包括发布年份），并从以下类型中选择：* Action* Adventure* Animation* Children's* Comedy* Crime* Documentary* Drama* Fantasy* Film-Noir* Horror* Musical* Mystery* Romance* Sci-Fi* Thriller* War* Western由于重复条目或测试条目，某些MovieID与电影不对应,
电影大多是手动输入的，因此可能存在错误和不一致

电影文件movies.dat中摘取部分记录如下：

1::Toy Story (1995)::Animation|Children's|Comedy
2::Jumanji (1995)::Adventure|Children's|Fantasy
3::Grumpier Old Men (1995)::Comedy|Romance
4::Waiting to Exhale (1995)::Comedy|Drama
5::Father of the Bride Part II (1995)::Comedy

3）所有职业信息都包含在“occupations.dat”文件中，格式如下：

OccupationID::Occupation
职业ID、职业名
从以下选择中选择职业：0:  "other" or not specified
1:  "academic/educator"
2:  "artist"
3:  "clerical/admin"
4:  "college/grad student"
5:  "customer service"
6:  "doctor/health care"
7:  "executive/managerial"
8:  "farmer"
9:  "homemaker"
10:  "K-12 student"
11:  "lawyer"
12:  "programmer"
13:  "retired"
14:  "sales/marketing"
15:  "scientist"
16:  "self-employed"
17:  "technician/engineer"
18:  "tradesman/craftsman"
19:  "unemployed"
20:  "writer"

职业信息occupations.dat中摘取部分记录如下：

0::other or not specified
1::academic/educator
2::artist
3::clerical/admin
4::college/grad student
5::customer service

3、代码实现

在Spark2.x的时候，可以使用DataSet去实现业务功能，本篇使用RDD实现业务功能。
1）数据读取，使用RDD读取数据。

     /*** 创建Spark会话上下文SparkSession和集群上下文SparkContext，在SparkConf中可以进行各种依赖和参数的设置等，* 大家可以通过SparkSubmit脚本的help去看设置信息，其中SparkSession统一了Spark SQL运行的不同环境。*/val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("MovieUsersAnalyzerRDDCase")//从SparkSession获得的上下文，这是因为我们读原生文件的时候或者实现一些Spark SQL目前还不支持的功能的时候需要使用SparkContextval sc: SparkContext = new SparkContext(conf)val dataPath: String = "hdfs://hadoop3:8020/input/movieRecom/moviedata/medium/"val outputDir: String = "hdfs://hadoop3:8020/out/movieRecom_out2"//    val dataPath = "data/moviedata/medium/"    //数据存放的目录val startTime: Long = System.currentTimeMillis();/*** 读取数据，用什么方式读取数据呢？在这里是使用RDD!*/val usersRDD: RDD[String] = sc.textFile(dataPath + "users.dat")val moviesRDD: RDD[String] = sc.textFile(dataPath + "movies.dat")val occupationsRDD: RDD[String] = sc.textFile(dataPath + "occupations.dat")//    val ratingsRDD = sc.textFile(dataPath + "ratings.dat")val ratingsRDD: RDD[String] = sc.textFile(dataPath + "ratings.dat")

2）统计电影中平均得分最高（口碑最好）的电影及观看人数最高的电影（流行度最高）TopN。

得分最高的Top10电影实现思路：如果想算总的评分的话一般肯定需要reduceByKey操作或者aggregateByKey操作。
第一步：把数据变成Key-Value，大家想一下在这里什么是Key，什么是Value。把MovieID设置成为Key，把Rating设置为Value。
第二步：通过reduceByKey操作或者aggregateByKey实现聚合，然后呢？
第三步：排序，如何做？进行Key和Value的交换。

 /***  注意：*   1，转换数据格式的时候一般都会使用map操作，有时候转换可能特别复杂，需要在map方法中调用第三方jar或者so库；*   2，RDD从文件中提取的数据成员默认都是String方式，需要根据实际需要进行转换格式；*   3，RDD如果要重复使用，一般都会进行Cache*   4，重磅注意事项，RDD的cache操作之后不能直接在跟其他的算子操作，否则在一些版本中cache不生效*/println("所有电影中平均得分最高（口碑最好）的电影:")val ratings: RDD[(String, String, String)] = ratingsRDD.map(_.split("::")).map(x => (x(0), x(1), x(2))).cache()// (MovieID, 平均评分)ratings.map(x => (x._2, (x._3.toDouble, 1)))  // (MovieID, (Rating, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)) // (MovieID, (总评分, 总评分次数)).map(x => (x._1, x._2._1.toDouble / x._2._2)) // (MovieID, 平均评分).sortBy(_._2, false) // 对value降序排列.take(10).foreach(println)val ratingss: RDD[(String, (Double, Int))] = ratingsRDD.map(_.split("::")).map(x => (x(1), (x(2).toDouble, 1)))for (elem <- ratingss.collect().take(10)) {println(s"elem: ${elem}")}ratings.map(x => (x._2, (x._3.toDouble, 1)))/*** 上面的功能计算的是口碑最好的电影，接下来我们分析粉丝或者观看人数最多的电影*/println("所有电影中粉丝或者观看人数最多的电影:")ratings.map(x => (x._2, 1)).reduceByKey(_+_).map(x => (x._2, x._1)).sortByKey(false).map(x => (x._2, x._1)).take(10).foreach(println)
//     (MovieID, 总次数)ratings.map(x => (x._2, 1)).reduceByKey(_+_).sortBy(_._2, false).take(10).foreach(println)

3）统计最受男性喜爱的电影TopN和最受女性喜爱的电影TopN。

单从ratings中无法计算出最受男性或者女性喜爱的电影Top10，因为该RDD中没有Gender信息，如果我们需要使用Gender信息来进行Gender的分类，此时一定需要聚合。
当然我们力求聚合的使用是 mapjoin（分布式计算的Killer是数据倾斜，map端的join是一定不会数据倾斜），在这里可否使用mapjoin呢？不可以，因为用户的数据非常多！所以在这里要使用正常的Join，此处的场景不会数据倾斜，因为用户一般都很均匀的分布（但是系统信息搜集端要注意黑客攻击）。

 /*** Tips：*   1，因为要再次使用电影数据的RDD，所以复用了前面Cache的ratings数据*   2, 在根据性别过滤出数据后关于TopN部分的代码直接复用前面的代码就行了。*   3, 要进行join的话需要key-value；*   4, 在进行join的时候时刻通过take等方法注意join后的数据格式  (3319,((3319,50,4.5),F))*   5, 使用数据冗余来实现代码复用或者更高效的运行，这是企业级项目的一个非常重要的技巧！*/val male = "M"val female = "F"val ratings2: RDD[(String, (String, String, String))] = ratings.map(x => (x._1, (x._1, x._2, x._3)))val usersRDD2: RDD[(String, String)] = usersRDD.map(_.split("::")).map(x => (x(0), x(1)))val genderRatings: RDD[(String, ((String, String, String), String))] = ratings2.join(usersRDD2).cache()
//    genderRatings.take(500).foreach(println)val maleRatings: RDD[(String, String, String)] = genderRatings.filter(x => x._2._2.equals("M")).map(x => x._2._1)val femaleRatings: RDD[(String, String, String)] = genderRatings.filter(x => x._2._2.endsWith("F")).map(x => x._2._1)println("所有电影中最受男性喜爱的电影Top10: ")maleRatings.map(x => (x._2, (x._3.toDouble, 1))).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).map(x => (x._1, x._2._1.toDouble / x._2._2)).sortBy(_._2, false).take(10).foreach(println)println("所有电影中最受女性喜爱的电影Top10: ")femaleRatings.map(x => (x._2, (x._3.toDouble, 1))).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).map(x => (x._1, x._2._1.toDouble / x._2._2)).sortBy(_._2, false).take(10).foreach(println)

源码和数据
https://github.com/fengqijie001/movieRecommendation

希望可以帮到各位，不当之处，请多指教~?