package hw3

import org.apache.spark._

import scala.collection.immutable.HashSet

import org.apache.spark.rdd.RDD

/**

 * @author BIGDATA

 */

object spark_hw3{

  var sc:SparkContext=null

  def main(args: Array[String]): Unit = {

    val conf=new SparkConf().setAppName("MovieDemo")

              .setMaster("local")

    sc=new SparkContext(conf)
　　//准备数据

     val rating=sc.textFile("C:\\Users\\BIGDATA\\Desktop\\文件\\BigData\\Spark\\3.SparkCore_2\\data\\data\\ratings.dat")

              .map(_.split("::")).map {x => (x(0),x(1),x(2))}
　　//年龄段在“18-24”的男性年轻人，最喜欢看哪10部

    top10LookeMovie

　　//得分最高的10部电影

    val topKScoreMostMovie = rating.map{x =>

      (x._2, (x._3.toInt, 1))

    }.reduceByKey { (v1, v2) =>

      (v1._1 + v2._1, v1._2 + v2._2)

    }.map { x =>

      (x._2._1.toDouble / x._2._2.toDouble, x._1)

    }.sortByKey(false).

        take(10).

        foreach(println)

　　　//女性看最多的10部电影

    top10FaleLookMovie

　　//男性看最多的10部电影

    top10MaleLookMovie 

　　//看过电影最多的前10个人

    val topKmostPerson = rating.map{ x =>

      (x._1, 1)

    }.reduceByKey(_ + _).

        map(x =>(x._2, x._1)).

        sortByKey(false).

        take(10).

        foreach(println)

     val brower = sc.textFile("C:\\Users\\BIGDATA\\Desktop\\文件\\BigData\\Spark\\3.SparkCore_2\\SogouQ2012.mini\\SogouQ.mini")

     val brs=brower.map(_.split("\t")).map { x =>

          x(5)

    }.cache

    //访问量前10的网站

     val topKBrower = brs.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

    .sortBy(_._2, false)

    .take(10)

    .foreach(println)

  }

  /**

    * @param sc SparkContext对象

    * @return 返回用户信息

    */

  def getUsers(sc:SparkContext):RDD[Array[String]]={

    val scobj=sc

    val users=scobj.textFile("C:\\Users\\BIGDATA\\Desktop\\文件\\BigData\\Spark\\3.SparkCore_2\\data\\data\\users.dat")

                  .map(_.split("::"))

    users

  }

  /**

    * @param sc

    * @return 返回电影信息

    */

  def getMovies(sc:SparkContext):RDD[Array[String]]={

    val scobj=sc

    val movies=scobj.textFile("C:\\Users\\BIGDATA\\Desktop\\文件\\BigData\\Spark\\3.SparkCore_2\\data\\data\\movies.dat")

                .map(_.split("::"))

    movies

  }

  /**

    *

    * @param sc

    * @return 电影评分信息

    */

  def getRatings(sc:SparkContext):RDD[Array[String]]={

    val scobj=sc

    val ratings=scobj.textFile("C:\\Users\\BIGDATA\\Desktop\\文件\\BigData\\Spark\\3.SparkCore_2\\data\\data\\ratings.dat")

              .map(_.split("::"))

    ratings

  }

  def top10LookeMovie: Unit ={

    //获取年龄段在“18-24”的男性年轻人的userid

    val users=getUsers(sc)

    val userList=users.filter(x=>x(1).equals("M") && x(2).toInt>=18 && x(2).toInt<=24)

      .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))

  }

  /**

    * 女性看过最多的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))

  }

  /**

    * 男性看过最多的10部电影

    */

  def top10MaleLookMovie: Unit ={

    val users = getUsers(sc)

    //获取所有男性的userid

    val faleUserId = users.filter(x => x(1).equals("M"))

      .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))

  }

}