Spark RDD算子/SparkSQL分别实现对电影数据集的简单数据分析

数据集

MovieLens 1M Dataset

users.dat

UserID::Gender::Age::Occupation::Zip-code



movies.dat

MovieID::Title::Genres



ratings.dat

UserID::MovieID::Rating::Timestamp

RDD算子实现

1.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影

val root = work.getClass.getResource("/")
val movieRdd = sc.textFile(root + "movies.dat", 3).map(_.split("::"))
val ratingRdd = sc.textFile(root + "ratings.dat", 3).map(_.split("::"))
val userRdd = sc.textFile(root + "users.dat", 3).map(_.split("::"))

val youngMale = userRdd.filter(a => a(1) == "M" && a(2).toInt >= 18 && a(2).toInt <= 25).map(a => (a(0), 1))
val rating = ratingRdd.map(a => (a(0), a(1)))
val movieID = rating.join(youngMale).map(_._2).reduceByKey(_ + _).sortBy(_._2, false).take(10)
//false降序，不加false，升序
val movieNmae = movieRdd.map(x => (x(0), x(1))).collect().toMap
val ans = movieID.map(x => movieNmae.get(x._1))
ans foreach println

2.得分最高的10部电影

val ratmovie = ratingRdd.map(x => (x(1), (x(2).toInt,1))).reduceByKey((x,y)=>(x._1+y._1,x._2+y._2)).map(x=>(x._1,x._2._1.toDouble/x._2._2.toDouble)).sortBy(_._2, false).take(10)
val ratMovName=ratmovie.map(x=>movieNmae.get(x._1))

输出电影ID



3.看过电影最多的前10个人

val userMovNum = ratingRdd.map(x => (x(0), x(1))).groupBy(_._1).map(x => (x._1, x._2.size)).sortBy(_._2, false).take(10)
userMovNum foreach println

输出用户ID和观看电影数量



4.5.男性看过最多的10部电影，女性看多最多的10部电影

val male = userRdd.filter(x => x(1) == "M").map(x => (x(0), 1))
val female = userRdd.filter(x => x(1) == "F").map(x => (x(0), 1))
val maleMovie = rating.join(male).map(_._2).reduceByKey(_ + _).sortBy(_._2, false).take(10)
val femaleMovie = rating.join(female).map(_._2).reduceByKey(_ + _).sortBy(_._2, false).take(10)
val maleMovNmae = maleMovie.map(x => movieNmae.get(x._1))
maleMovNmae foreach println
println
val femaleMovName = femaleMovie.map(x => movieNmae.get(x._1))
femaleMovName foreach println
println

male



female



数据集SogouQ2012.mini.tar.gz

访问时间\t用户ID\t[查询词]\t该URL在返回结果中的排名\t用户点击的顺序号\t用户点击的URL



1.将数据按照访问次数进行排序，求访问量前10的网站

val urlRdd=sc.textFile(root+"SogouQ.mini").map(_.split("\t")).map(x=>(x(5),1)).reduceByKey(_+_).sortBy(_._2,false).take(10)
urlRdd foreach println

val conf =new SparkConf().setMaster("local[*]").setAppName("sparksql")
val sc=new SparkContext(conf)
val sqlContext=new SQLContext(sc)
val root=work02.getClass.getResource("/")

SparkSQL实现

数据导入

import sqlContext.implicits._
val userRdd=sc.textFile(root+"users.dat").map(_.split("::"))
val userdf=userRdd.map(x=>user(x(0),x(1),x(2).toInt)).toDF

val movieRdd=sc.textFile(root+"movies.dat").map(_.split("::"))
val moviedf=movieRdd.map(x=>movie(x(0),x(1))).toDF

val ratingRdd=sc.textFile(root+"ratings.dat").map(_.split("::"))
val ratingdf=ratingRdd.map(x=>rating(x(0),x(1),x(2).toDouble)).toDF

年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影

val youngmale=userdf.filter("18<=age and gender='M'")
val youngratting=ratingdf.select("uid","mid")
val youngmovies=youngmale.join(youngratting, "uid").groupBy("mid").count.sort(-$"count").limit(10)
youngmovies.join(moviedf,"mid").select("title").show(false)

得分最高的10部电影

val top=ratingdf.groupBy("mid").agg("uid"->"count","rat"->"sum").withColumnRenamed("count(uid)","count").withColumnRenamed("sum(rat)","sum")//不改名则在SQL中不能用名来读到新增的这两列，不知为何
top.registerTempTable("top")
val top10=sqlContext.sql("select mid,sum/count from top").sort(-$"_c1").limit(10)
top10.join(moviedf,"mid").select("title").show(false)

看过电影最多的前10个人

ratingdf.groupBy("uid").count().sort(-$"count").show(10)

男性看过最多的10部电影

val male=userdf.filter("gender='M'").select("uid")
val allmovies=ratingdf.select("uid","mid")
val maleid=male.join(allmovies,"uid").groupBy("mid").count.sort(-$"count").limit(10)//.repartition($"count")没用
maleid.join(moviedf,"mid").select("title").show(false)

女性看多最多的10部电影

val female =userdf.filter("gender='F'").select("uid")
val femaleid=female.join(allmovies,"uid").groupBy("mid").count.sort(-$"count").limit(10)
femaleid.join(moviedf,"mid").select("title").show(false)
maleid.registerTempTable("maleid")
sqlContext.sql("select count(*) as count from maleid group by mid").join(moviedf).select("title").show(false)

访问量前10的网站

val urlRdd=sc.textFile(root+"SogouQ.mini").map(_.split("\t"))
val usrdf=urlRdd.map(w=>urlclass(w(5))).toDF
usrdf.groupBy("url").count.sort(-$"count").limit(10).show(false)

存在的问题：groupby和reuduceby相比很慢，尝试用SQL中的groupby效果一样，尝试用transform方法，调用RDD的reduceby报错，还在继续学习研究