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

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

现有如此三份数据：
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_rating

import 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_rating

import 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_rating

import org.apache.spark.rdd.RDD

/**
  * Utils.usersRdd：对应字段中文解释：用户 id，性别，年龄，职业，邮政编码
  * Utils.movie