Spark算子—实战应用详解大数据

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