Spark写MySQL经典五十题

Mysql的版本为8.0.29

---

---

创建表及信息录入

# 学生表
create table student
(sid   varchar(10),sname varchar(32),saged datetime,ssex  varchar(10)
) comment '学生表';
# 学生表插入数据
insert into student(sid,sname,saged,ssex)values('01' , '赵雷' , '1990-01-01' , '男');
insert into student(sid,sname,saged,ssex)values('02' , '钱电' , '1990-12-21' , '男');
insert into student(sid,sname,saged,ssex)values('03' , '孙风' , '1990-05-20' , '男');
insert into student(sid,sname,saged,ssex)values('04' , '李云' , '1990-08-06' , '男');
insert into student(sid,sname,saged,ssex)values('05' , '周梅' , '1991-12-01' , '女');
insert into student(sid,sname,saged,ssex)values('06' , '吴兰' , '1992-03-01' , '女');
insert into student(sid,sname,saged,ssex)values('07' , '郑竹' , '1989-07-01' , '女');
insert into student(sid,sname,saged,ssex)values('08' , '王菊' , '1990-01-20' , '女');-- -------------------------------------------------------------
# 课程表
create table course
(cid   varchar(10),cname varchar(32),tid   varchar(10)
) comment '课程表';
# 插入课程数据
insert into course(cid,cname,tid)values('01' , '语文' , '02');
insert into course(cid,cname,tid)values('02' , '数学' , '01');
insert into course(cid,cname,tid)values('03' , '英语' , '03');-- --------------------------------------------------------
# 教师表
create table teacher
(tid   varchar(10),tname varchar(32)
)comment '教师表';
# 插入教师表信息
insert into teacher(tid,tname)values('01' , '张三');
insert into teacher(tid,tname)values('02' , '李四');
insert into teacher(tid,tname)values('03' , '王五');-- --------------------------------------------------
# 成绩表
create table sc
(sid   varchar(10),cid   varchar(10),score int(30)
)comment '成绩表';
# 插入成绩信息insert into SC values('01','01',80);
insert into SC values('01','02',90);
insert into SC values('01','03',99);
insert into SC values('02','01',70);
insert into SC values('02','02',60);
insert into SC values('02','03',80);
insert into SC values('03','01',80);
insert into SC values('03','02',80);
insert into SC values('03','03',80);
insert into SC values('04','01',50);
insert into SC values('04','02',30);
insert into SC values('04','03',20);
insert into SC values('05','01',76);
insert into SC values('05','02',87);
insert into SC values('06','01',31);
insert into SC values('06','03',34);
insert into SC values('07','02',89);
insert into SC values('07','03',98);

连接数据库

def main(args: Array[String]): Unit = {val spark: SparkSession = SparkSession.builder().master("local[*]").appName("mysql").getOrCreate()import spark.implicits._val driver = "com.mysql.cj.jdbc.Driver"val url = "jdbc:mysql://192.168.95.130:3306/mysql50"val user = "root"val password = "root"val prop = new Properties()prop.setProperty("driver",driver)prop.setProperty("user",user)prop.setProperty("password",password)val courseTable = "course"val scoreTable = "sc"val studentTable = "student"val teacherTable = "teacher"val courseDf: DataFrame = spark.read.jdbc(url, courseTable, prop)val scDf: DataFrame = spark.read.jdbc(url, scoreTable, prop)val studentDf: DataFrame = spark.read.jdbc(url, studentTable, prop)val teacherDf: DataFrame = spark.read.jdbc(url, teacherTable, prop)}}  

1.查询"01"课程比"02"课程成绩高的学生的信息及课程分数

    scDf.as("s1").join(scDf.as("s2"), "sid").filter("s1.cid=01 and s2.cid=02 and s1.score>s2.score").join(studentDf, "sid").show

2.查询"01"课程比"02"课程成绩低的学生的信息及课程分数

    scDf.as("s1").join(scDf.as("s2"), "sid").filter("s1.cid=01 and s2.cid=02 and s1.score<s2.score").join(studentDf, "sid").show

3.查询平均成绩大于等于60分的同学的学生编号和学生姓名和平均成绩

    val frame: DataFrame = scDf.as("s1").groupBy("sid").avg("score").join(studentDf.as("s2"), "sid").filter($"avg(score)">=60)frame.show()

4.查询平均成绩小于60分的同学的学生编号和学生姓名和平均成绩

      studentDf.as("s1").join(scDf.as("s2").groupBy("sid").avg("score"),Seq("sid"),"left_outer").filter($"avg(score)"<60 ||$"avg(score)".isNull).show()

5.查询所有同学的学生编号、学生姓名、选课总数、所有课程的总成绩

    studentDf.join(scDf.groupBy("sid").count(), Seq("sid"), "left_outer").join(scDf.groupBy("sid").sum(), Seq("sid"), "left_outer").show()

6.查询"李"姓老师的数量

    val l: Long = teacherDf.filter($"tname".like("李%")).count()println(l)

7.查询学过"张三"老师授课的同学的信息

    val value: Dataset[Row] = scDf.join(courseDf, "cid").join(teacherDf, "tid").join(studentDf, "sid").filter($"tname".equalTo("张三"))value.show()

8.查询没学过"张三"老师授课的同学的信息

  scDf.join(teacherDf.filter("tname ='张三' ").join(courseDf,"tid"),"cid").join(studentDf,Seq("sid"),"right_outer").filter($"tname" isNull).show()

9.查询学过编号为"01"并且也学过编号为"02"的课程的同学的信息

    studentDf.join(scDf.filter($"cid".equalTo("01")),"sid").join(scDf.filter($"cid".equalTo("02")),"sid").show()

10.查询学过编号为"01"但是没有学过编号为"02"的课程的同学的信息

    studentDf.join(scDf.filter("cid=2"), Seq("sid"), "leftouter").where("cid is null").join(scDf.filter("cid=1"), Seq("sid")).show()

11.查询没有学全所有课程的同学的信息

    studentDf.join(scDf,Seq("sid"),"left_outer").groupBy("sid").count().where("count<3").join(studentDf,"sid").show()

12.查询至少有一门课与学号为"01"的同学所学相同的同学的信息

    studentDf.join(scDf, "sid").as("s1").join(scDf.where("sid=01"), "cid").select("s1.sid").distinct().where("sid!=01").join(studentDf, "sid").show

13.查询和"01"号的同学学习的课程完全相同的其他同学的信息

    scDf.where("sid=01").as("s1").join(scDf.as("s2"), "cid").groupBy("s2.sid").count().as("s3").where(s"count=${scDf.where("sid=01").count()} and sid!=01").join(studentDf, "sid")show()

14.查询没学过"张三"老师讲授的任一门课程的学生姓名

      studentDf.join(scDf,"sid").join(courseDf,"cid").join(teacherDf.where("tname='张三'"),"tid").as("a").select("sid").join(studentDf.as("b"),Seq("sid"),"right_outer").where("a.sid is null").select("sname").show()

15.查询两门及其以上不及格课程的同学的学号，姓名及其平均成绩

    scDf.where("score<60").groupBy("sid").count().where("count>=2").join(scDf,"sid").groupBy("sid").avg("score").join(studentDf,"sid").show()

16.检索"01"课程分数小于60，按分数降序排列的学生信息

    scDf.where("cid=01").join(studentDf,Seq("sid"),"right_outer").where("score<60 or score is null").orderBy($"score".desc).show()

17.按平均成绩从高到低显示所有学生的所有课程的成绩以及平均成绩

 scDf.join(scDf.groupBy("sid").avg("score"),Seq("sid"),"left_outer").join(studentDf,"sid").orderBy($"avg(score)".desc).show()

18.查询各科成绩最高分、最低分和平均分：以如下形式显示：课程ID，课程name，最高分，最低分，平均分，及格率，中等率，优良率，优秀率

    import org.apache.spark.sql.functions._val s1: DataFrame = scDf.groupBy("cid").agg(max("score").as("maxscore"), min("score").as("minscore"), avg("score").as("avgscore"), count("score").as("num"))//及格率val jige: DataFrame = scDf.rdd.map(x => {if (x.getAs("score").toString.toInt > 60) (x(1).toString, 1) else (x(1).toString, 0)}).reduceByKey(_ + _).toDF("cid", "jige")//中等率val zhongdeng: DataFrame = scDf.rdd.map(x => {if (x.getAs("score").toString.toInt > 70) (x(1).toString, 1) else (x(1).toString, 0)}).reduceByKey(_ + _).toDF("cid", "zhongdeng")//优良率val youliang: DataFrame = scDf.rdd.map(x => {if (x.getAs("score").toString.toInt > 80) (x(1).toString, 1) else (x(1).toString, 0)}).reduceByKey(_ + _).toDF("cid", "youliang")//优秀率val youxiu: DataFrame = scDf.rdd.map(x => {if (x.getAs("score").toString.toInt > 90) (x(1).toString, 1) else (x(1).toString, 0)}).reduceByKey(_ + _).toDF("cid", "youxiu")//联表s1.join(jige,"cid").join(zhongdeng,"cid").join(youliang,"cid").join(youxiu,"cid").withColumn("jigelv",col("jige")/col("num")).withColumn("zhongdenglv",col("zhongdeng")/col("num")).withColumn("youlianglv",col("youliang")/col("num")).withColumn("youxiulv",col("youxiu")/col("num")).drop("jige","zhongdeng","youxiu","youliang").show()

19.按各科成绩进行排序，并显示排名

 scDf.selectExpr("*","row_number() over(partition by cid order by score desc)").show()

20.查询学生的总成绩并进行排名

    scDf.selectExpr("*","sum(score) over(partition by sid) as sumscore").dropDuplicates("sid").selectExpr("*","row_number() over(order by sumscore desc)").show()

21.查询不同老师所教不同课程平均分从高到低显示

    scDf.join(courseDf,"cid").join(teacherDf,"tid").groupBy("tid","cid").avg("score").orderBy($"avg(score)".desc).show()

22.查询所有课程的成绩第2名到第3名的学生信息及该课程成绩

    scDf.selectExpr("*","row_number() over(partition by cid order by score desc) num").where("num between 2 and 3").join(studentDf,"sid").show()

23.统计各科成绩各分数段人数：课程编号,课程名称,[100-85],[85-70],[70-60],[0-60]及所占百分比

    //分段val fenduan = scDf.rdd.map(x=>{if(x.getAs("score").toString.toInt < 60) (x(1).toString,1)else if(x.getAs("score").toString.toInt < 70) (x(1).toString,2)else if(x.getAs("score").toString.toInt < 85) (x(1).toString,3)else (x(1).toString,4)}).toDF("cid","fenduan")//联表fenduan.groupBy("cid").count.as("f1").join(fenduan.groupBy("cid","fenduan").count.as("f2"),"cid").withColumn("percent",$"f2.count"/$"f1.count").drop($"f1.count").join(courseDf,"cid").show()

24.查询学生平均成绩及其名次

    scDf.groupBy("sid").avg("score").selectExpr("*",s"row_number() over(order by 'avg(score)')").show()

25.查询各科成绩前三名的记录

    scDf.selectExpr("*","row_number() over(partition by cid order by score desc) num").where("num<=3").show()

26.查询每门课程被选修的学生数

scDf.groupBy("cid").count().show()

27.查询出只有两门课程的全部学生的学号和姓名

    scDf.groupBy("sid").count().where("count=2").join(studentDf,"sid").show()

28.查询男生、女生人数

 studentDf.groupBy("ssex").count().show()

29.查询名字中含有"风"字的学生信息

studentDf.where("sname like '%风%'").show()

30.查询同名同姓学生名单，并统计同名人数

    studentDf.groupBy("sname").count().where("count>1").show()

31.查询1990年出生的学生名单

studentDf.where("year(saged)=1990").show()

32.查询每门课程的平均成绩，结果按平均成绩降序排列，平均成绩相同时，按课程编号升序排列

scDf.groupBy("cid").avg("score").orderBy($"avg(score)".desc,$"cid").show()

33.查询平均成绩大于等于85的所有学生的学号、姓名和平均成绩

    scDf.groupBy("sid").avg("score").where("avg(score)>=85").join(studentDf,"sid").show()

34.查询课程名称为"数学"，且分数低于60的学生姓名和分数

    scDf.where("score<60").join(courseDf,"cid").where("cname='数学'").join(studentDf,"sid").show()

35.查询所有学生的课程及分数情况

studentDf.join(scDf,Seq("sid"),"left_outer").show()

36.查询任何一门课程成绩在70分以上的学生姓名、课程名称和分数

    scDf.where("score>70").join(studentDf,"sid").join(courseDf,"cid").show()

37.查询课程不及格的学生

    scDf.where("score<60 or score is null").join(studentDf,"sid").show()

38.查询课程编号为01且课程成绩在80分以上的学生的学号和姓名

    scDf.where("cid=01 and score>80").join(studentDf,"sid").show()

39.求每门课程的学生人数

scDf.groupBy("cid").count().show()

40.查询选修"张三"老师所授课程的学生中，成绩最高的学生信息及其成绩

    scDf.join(studentDf,"sid").join(courseDf,"cid").join(teacherDf,"tid").where("tname='张三'").join(studentDf,"sid").selectExpr("*","max(score) over() max").where("max=score").show()

41.查询不同课程成绩相同的学生的学生编号、课程编号、学生成绩

    scDf.as("s1").join(scDf.as("s2"),"sid").where("s1.score=s2.score and s1.cid!=s2.cid").show()

42.查询每门课程成绩最好的前三名

    scDf.selectExpr("*","row_number() over(partition by cid order by score desc)rank").where("rank<=3").show()

43.统计每门课程的学生选修人数（超过5人的课程才统计）要求输出课程号和选修人数，查询结果按人数降序排列，若人数相同，按课程号升序排列

   scDf.groupBy("cid").count().where("count>=5").orderBy($"count".desc).orderBy("cid").show()

44.检索至少选修两门课程的学生学号

    scDf.groupBy("sid").count().where("count>=2").show()

45.查询选修了全部课程的学生信息

    studentDf.join(scDf,Seq("sid"),"left_outer").groupBy("sid").count().where(s"count=${courseDf.select("cid").count() }").join(studentDf,"sid").show()

46.查询各学生的年龄(周岁)

studentDf.selectExpr("*","year(current_date)-year(saged)").show()

47.查询本周过生日的学生

    studentDf.where("unix_timestamp(cast(concat_ws('-',date_format(current_date(),'yyyy')," +"date_format(saged,'MM'),date_format(saged,'dd'))as date),'yyyy-MM-dd')" +" between unix_timestamp(current_date()) " +"and unix_timestamp(date_sub(next_day(current_date(),'MON'),1),'yyyy-MM-dd')").show()

48.查询下周过生日的学生

    studentDf.where(" unix_timestamp( cast( concat_ws('-',date_format(current_date(),'yyyy')" +",date_format(saged,'MM'),date_format(saged,'dd') ) as date ),'yyyy-MM-dd') " +"between unix_timestamp(date_sub(next_day(current_date(),'MON'),1),'yyyy-MM-dd') " +"and unix_timestamp(date_add(next_day(current_date(),'MON'),6),'yyyy-MM-dd') ").show()

49.查询本月过生日的学生

studentDf.where("month(saged)=month(current_date)").show()

50.查询下月过生日的学生

studentDf.where("month(saged)=month(current_date)+1").show()