电影受众分析系统

spark on yarn模式分为两种模式：

一、Yarn-cluster模式

1、通过spark-submit提交spark jar包（Application）,与RM进行通信请求启动AM

2、RM接收到请求之后，会在一个相对空闲的nodemanager中分配Container,然后再此Container中启动AM

3、AM启动之后，会向RM进行通信，请求资源用于启动nodemagaer(相当于Worker）节点中的Executor

4、RM分配一批Container容器到NM中，用于启动Executor

5、AM与其他NM（worker）进行通信，启动Executor

6、executor启动之后，就会向AM（Driver）进行反注册

7、接下来的执行流程，就与Standalone模式的执行流程是一样的

二、Yarn-Client模式

1、通过spark-submit提交spark jar包（Application）,与RM进行通信请求启动AM，并且在客户端本地启动Driver进程

2、RM接收到请求之后，会在一个相对空闲的nodemanager中分配Container,然后再此Container中启动AM

3、AM启动之后，会向RM进行通信，请求资源用于启动nodemagaer(相当于Worker）节点中的Executor

4、RM分配一批Container容器到NM中，用于启动Executor

5、AM与其他NM（worker）进行通信，启动Executor

6、Executor启动之后，就会向客户端本地的Driver进行反注册。

7、接下来的执行流程，就与Standalone模式的执行流程是一样的。

三、总结：

经过上面的描述，yarn-cluster与yarn-client模式有一个很大的区别，就是Driver进程所运行的位置。

在yarn-cluster模式中，Driver运行在AM所在节点上

在yarn-client模式中，Driver运行在客户端本地上

在yarn-client模式中，由于Driver运行在客户端本地，如果运行中出现Bug,那么可以在本地直接通过查Log日志来判定Bug的原因，从而对程序进行调整。

在yarn-client模式中，由于Driver运行在AM所在节点上，如果运行中出现BUG，只能通过集群中产生的Log日志进行判定，需要用到application_id来进行查找log,过程比较麻烦。

综上所述，yarn-client模式便于排错，一般用于测试环境，而yarn-cluster模式一般用于真实的生产环境中。

案例分析：Spark项目案例-电影受众分析系统

一、数据结构
users.dat：
UserID::Gender::Age::Occupation（职业） ::Zip-code（邮政编码）
movies.dat
MovieID::Title::Genres（题材）
ratings.dat
UserID::MovieID::Rating::Timestamp

二、功能需求
 1、看过“ Lord of the Rings,The(1978) ” 用户年龄和性别分布
 2、年龄段在“ 18-24” 的男性年轻人，最喜欢看哪10部电影
 3、得分最高的10部电影
 4、看过电影最多的前10个人
 5、女性看过最多的10部电影
 6、男性看过最多的10部电影
 7、年龄段在“ 45-49” 的男性观众中，得分最高的10部电影名称
 8、 1995年最受欢迎的前3部电影的用户年龄和性别数量分布

9、moiveid=2116这部电影各年龄段的平均影评（年龄段，平均影评分）

10、统计最喜欢看电影的那位女士最喜欢的10部电影，观众对他们的平均影评分（电影名，影评）

1、

object MovieDemo {
  var sc:SparkContext=null
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("MovieDemo")
              .setMaster("local")
    sc=new SparkContext(conf)
    ageAndSexCount
  }
 
  /**
    * @param sc SparkContext对象
    * @return 返回用户信息
    */
  def getUsers(sc:SparkContext):RDD[Array[String]]={
    val scobj=sc
    val users=scobj.textFile("file:///home/tg/datas/users.dat")
                  .map(_.split("::"))
    users
  }
 
  /**
    * @param sc
    * @return 返回电影信息
    */
  def getMovies(sc:SparkContext):RDD[Array[String]]={
    val scobj=sc
    val movies=scobj.textFile("file:///home/tg/datas/movies.dat")
                .map(_.split("::"))
    movies
  }
 
  /**
    *
    * @param sc
    * @return 电影评分信息
    */
  def getRatings(sc:SparkContext):RDD[Array[String]]={
    val scobj=sc
    val ratings=scobj.textFile("file:///home/tg/datas/ratings.dat")
              .map(_.split("::"))
    ratings
  }
 
  /**
    * 看过“Lord of the Rings,The (1978) ”用户年龄和性别(数量)分布
    */
  def ageAndSexCount: Unit ={
    //获取“Lord of the Rings,The (1978) ”这部电影的movieId
    val movies=getMovies(sc)
    val movieId=movies.filter(x=>x.length==3 && x(1).equals("Lord of the Rings, The (1978)"))
              .map(x=>x(0).trim).toArray()(0)
    //从users.dat中获取所有user的(userid,(age,gender))
    val users=getUsers(sc)
    val usersRDD=users.map(x=>{
      (x(0),(x(2),x(1))) //(userid,(age,gender))
    })
    //获取看过"Lord of the Rings,The (1978)"这部电影的(userid,movieid)
    val ratings=getRatings(sc)
    val ratingRDD=ratings.map(x=>{
      (x(0),x(1)) //ratings数据中所有的(userid,movieid)
    }).filter(x=>x._2.equals(movieId))
 
    //将ratingRDD和usersRDD进行join操作，结果<userid,(movieid,(age,gender))>
    val userratingRDD=ratingRDD.join(usersRDD)
    println("看过“Lord of the Rings,The (1978) ”用户年龄和性别(数量)分布")
    userratingRDD.map(x=>{
      (x._2._2,1)
    }).reduceByKey(_+_)
      .foreach(println(_))
  }
}

2、

/**
    * 年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影
    */
  def top10LookeMovie: Unit ={
    //获取年龄段在“18-24”的男性年轻人的userid
    val users=getUsers(sc)
    val userList=users.filter(x=>x(1).equals("M") && x(2).toInt==18)
      .map(x=>x(0)).collect()
    //注意：HashSet()后面要带小括号
    val userSet=HashSet() ++ userList
    //创建广播变量
    val broadcastUserSet=sc.broadcast(userSet)
    //统计出１８－２４岁男性喜欢看的前１０名电影的movieid和次数
    val ratings=getRatings(sc)
    val topNMovies=ratings.map(x=>(x(0),x(1))) //ratings中所有的(userid,movieid)
      //从rating数据过滤出“18-24”的男性年轻人的观影信息
      .filter(x=>broadcastUserSet.value.contains(x._1))
      .map(x=>(x._2,1))
      .reduceByKey(_+_) //(movieid,次数)
      .sortBy(_._2,false)
      .take(10) //(movieid,次数)
 
    val movies=getMovies(sc)
    //获取所有电影的(movieid,title)
    val movieTitle=movies.map(x=>(x(0),x(1))).collect().toMap
    topNMovies.map(x=>(movieTitle.getOrElse(x._1,null),x._2))
      .foreach(x=>println(x._1+"  "+x._2))
  }

3、

def getTenMovie: Unit ={
    val ratings=getRatings(sc)
    //获取得分最高的１０部电影的movieId
    val top10moiveid=ratings.map(x=>(x(0),x(1),x(2))) //(userid,movieid,rating)
      .map(x=>(x._2,(x._2.toInt,1))) //(movieid,(rating,1))
      .reduceByKey((x,y)=>{
      (x._1+y._1,x._2+y._2) //注意此行代码，求出了电评的总分、评价的次数
    }).map(x=>{
      (x._2._1.toDouble/x._2._2.toDouble,x._1) //(电影平均分,movieid)
    }).sortByKey(false).take(10).map(x=>x._2)
 
    //广播变量
    val broadcastMovieList=sc.broadcast(top10moiveid)
    val movies=getMovies(sc)
    val result=movies.map(x=>(x(0),x(1))) //取出全部电影的(movieid,title)
      .filter(x=>broadcastMovieList.value.contains(x._1))
      println("得分最高的10部电影是：")
    result.foreach(x=>println(x._1+"  "+x._2))
  }

4、

5、

/**
    * 女性看过最多的10部电影
    */
  def top10FaleLookMovie: Unit ={
    val users = getUsers(sc)
    //获取所有女性的userid
    val faleUserId = users.filter(x => x(1).equals("F"))
      .map(x => x(0)).collect()
    val faleUserSet = HashSet() ++ faleUserId
    //创建广播变量，里面存储所有女性的userid
    val broadcastFaleSet = sc.broadcast(faleUserSet)
 
    val ratings = getRatings(sc)
    //统计出女性看过最多的１０部电影的(movieid,观看次数)
    val top10moiveid = ratings.map(x => (x(0), x(1))) //(userid,movieid)
      //过滤出女性观影数据
      .filter(x => broadcastFaleSet.value.contains(x._1))
      .map(x => (x._2, 1)) //(movieid,1)
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .take(10)
    val top10movieRDD=sc.parallelize(top10moiveid) //(movieid,次数)
 
    val movies=getMovies(sc)
    val allmoviesRDD=movies.map(x=>(x(0),x(1))) //(movieid,title)
    //对两个ＲＤＤ进行join操作，取二者的共同匹配项
    allmoviesRDD.join(top10movieRDD) //(movieid,(title,次数))
      .map(x=>(x._1,x._2._1,x._2._2))
      .foreach(x=>println(x._1+"  "+x._2+"  "+x._3))
  }

6、解决方法：将女性改为男性就好了

7、

/**
    * 年龄段在“45-49”的男性观众中，得分最高的10部电影名称
    */
  def top10MaleMovie: Unit ={
    //获取所有年龄段在“45-49”的男性观众的userid
    val users=getUsers(sc)
    val usersList=users.filter(x=>x(1).equals("M") && x(2).toInt==45)
        .map(x=>x(0)).collect()
    val userSet=HashSet() ++ usersList
    //创建广播变量
    val broadcastUserSet=sc.broadcast(userSet)
 
    val ratings=getRatings(sc)
    //得分最高的10部电影的(movieid,avg)
    val topmovies=ratings.map(x=>(x(0),x(1),x(2)))  //(userid,movieid,rating)
      .filter(x=>broadcastUserSet.value.contains(x._1))
      .map(x=>(x._2,(x._3.toInt,1))) //(movieid,(rating,1))
      .reduceByKey((x,y)=>{
      (x._1+y._1,x._2+y._2)
    }).map(x=>{
      val sum=x._2._1.toDouble
      val rcount=x._2._2.toDouble
      val avg=sum/rcount  //求出平均分
      (avg,x._1)
    }).sortByKey(false).take(10)
      .map(x=>(x._2,x._1))  //(movieid,avg)
    val topmoviesRDD=sc.parallelize(topmovies)
 
    val movies=getMovies(sc)
    val allmovies=movies.map(x=>(x(0),x(1))).collect().toMap
    topmoviesRDD.map(x=>(allmovies.getOrElse(x._1,null),x._2))
      .foreach(println(_))
  }

8、第一步统计出1995年播放电影的moviesid(substring) title.substring(title.length-5,title.length-1).equals("1995")

第二步从ratings中统计出1995年最受欢迎的前三部电影的movieid

第三步计算出观看过1995年最受欢迎的前三部电影的（userid,movieid）

第四步统计出1995年最受欢迎的前3部电影的观众年龄和性别数量分布

/**
    * 1995年最受欢迎的前3部电影的观众年龄和性别数量分布
    */
  def top3AgeAndSexCount: Unit ={
    //第一步：统计出1995年播放电影的moviesid
    val movies=getMovies(sc)
    val moviesid=movies.filter(x=>{
      var title=x(1).trim
      title.substring(title.length-5,title.length-1).equals("1995")
    }).map(x=>x(0))collect()
    val moviesidSet=HashSet()++moviesid
    //创建广播变量
    val broadcastmoviesidSet=sc.broadcast(moviesidSet)
 
    //第二步：从ratings中统计出1995年最受欢迎的前3部电影的movieid
    val ratings=getRatings(sc)
    val top3moviesid=ratings.map(x=>(x(0),x(1))) //(userid,movieid)
      .distinct() //数据去重
      //过滤出1995年观影的(userid,movieid)
      .filter(x=>broadcastmoviesidSet.value.contains(x._2))
      .map(x=>(x._2,1)) //(movieid,1)
      .reduceByKey(_+_) //(movieid,次数)
      .map(x=>(x._2,x._1))
      .sortByKey(false).take(3)
      .map(x=>x._2) //movieid
 
    val top3moviesidSet=HashSet()++top3moviesid
    val broadcasttop3moviesidSet=sc.broadcast(top3moviesidSet)
    //第三步：统计出观看过1995年最受欢迎的前3部电影的(userid,movieid)
    val top3useridmovieid=ratings.map(x=>(x(0),x(1)))//所有的(userid,movieid)
      .filter(x=>broadcasttop3moviesidSet.value.contains(x._2))
    //第四步：统计出1995年最受欢迎的前3部电影的观众年龄和性别数量分布
    val users=getUsers(sc)
    val allusers=users.map(x=>(x(0),(x(2),x(1)))) //(userid,(age,gender))
 
    top3useridmovieid.join(allusers)//(userid,(movieid,(age,gender)))
      .map(x=>(x._2._2,1)) //((age,gender),1)
      .reduceByKey(_+_)
      .map(x=>(x._1,x._2)) //((age,gender),数量)
      .foreach(println(_))
  }

9、

/**
    * movieid = 2116 这部电影各年龄段的平均影评（年龄段，平均影评分）
    */
  def ageAndRating: Unit ={
    //第一步：从users中获取所有的(userid,age)
    val usersRDD=getUsers(sc)
      .map(x=>(x(0),x(2))) //(userid,age)
    //第二步：从ratings中获取观众对movieid=2116这部电影的影评(userid,rating)
    val useridratingRDD=getRatings(sc)
      .map(x=>(x(0),x(1),x(2))) //(userid,movieid,rating)
      .filter(x=>x._2.toInt==2116)
      .map(x=>(x._1,x._3)) //(userid,rating)
    //第三步：usersRDD join useridratingRDD,这两个RDD进行join操作
    usersRDD.join(useridratingRDD) //(userid,(age,rating))
      .map(x=>(x._2._1,(x._2._2.toInt,1))) //(age,(rating,1))
      .reduceByKey((x,y)=>{
      (x._1+y._1,x._2+y._2) //分别对rating和１进行累计求和
    }).map(x=>{
      val age=x._1
      val avg=x._2._1.toDouble/x._2._2.toDouble
      (age,avg)
    }).foreach(x=>println(x._1+"  "+x._2))
  }

10、第一步：统计最喜欢看电影的那位女士的userid 第二步：那位女士最喜欢的10部电影的movieid 第三步：观众对这10部电影的平均影评

//将小数保留两位，四舍五入
  def twoNum(num:Double):String={
    val df = new DecimalFormat("#.00");
    val res=df.format(num)
    res
  }

/**
    * 统计最喜欢看电影的那位女士最喜欢的10部电影，
    * 观众对它们的平均影评分(电影名，影评)
    */
  def titleAndRating: Unit ={
    //第一步：统计最喜欢看电影的那位女士的userid
    //从users中获取所有女性的userid
    val users=getUsers(sc)
    val faleusersid=users.filter(x=>x(1).equals("F"))
      .map(x=>x(0)).collect()
    val faleuseridSet=HashSet()++faleusersid
    val broadcastfaleidSet=sc.broadcast(faleuseridSet)
 
    val ratings=getRatings(sc)
    val fuserid=ratings.map(x=>(x(0),x(1))) //(userid,movieid)
      .distinct()
      //过滤出所有女性的观影数据
      .filter(x=>broadcastfaleidSet.value.contains(x._1))
      .map(x=>(x._1,1))  //(userid,1)
      .reduceByKey(_+_)
      .map(x=>(x._2,x._1))
      .sortByKey(false).take(1)
      .map(x=>x._2) //此userid是最喜欢看电影的那位女士的userid
 
    val broadfuesrid=sc.broadcast(fuserid)
 
    //第二步：那位女士最喜欢的10部电影的moviesid
    val top10moviesid=ratings.filter(x=>x(0).equals(broadfuesrid.value(0)))
      .map(x=>(x(1),(x(2).toInt,1))) //(movieid,(rating,1))
      .reduceByKey((x,y)=>{
      (x._1+y._1,x._2+y._2)
    }).map(x=>{
      val movieid=x._1
      val avg=x._2._1.toDouble/x._2._2.toDouble
      (avg,movieid)
    }).sortByKey(false).take(10)
      .map(x=>x._2)  //movieids
 
    val top10moviesidSet=HashSet()++top10moviesid
    val broadtop10moviesSet=sc.broadcast(top10moviesidSet)
 
    //第三步：观众对这10部电影的平均影评，影评保留两位小数，四舍五入
    val top10moviesavg=ratings.map(x=>(x(1),x(2))) //(movieid,rating)
      .filter(x=>broadtop10moviesSet.value.contains(x._1))
      .map(x=>(x._1,(x._2.toInt,1))) //(movieid,(rating,1))
      .reduceByKey((x,y)=>{
      (x._1+y._1,x._2+y._2)
    }).map(x=>{
      val movieid=x._1
      val avg=x._2._1.toDouble/x._2._2.toDouble
      val resavg=twoNum(avg)//对平均影评保留两位小数
      (movieid,resavg)  //(movieid,平均分)
    })
    //(电影名,影评)
    val movies=getMovies(sc)
    val allmovies=movies.map(x=>(x(0),x(1))) //(movieid,title)
 
    allmovies.join(top10moviesavg) //(movieid,(title,平均分))
      .map(x=>(x._2._1,x._2._2)) //(title,平均分)
      .foreach(println(_))
  }