林子雨spark scala版编程小结

Posted 2022-06-29 孤独之风。

dataframe有first()返回第一行数据，使用head(n)返回前n行数据，也可以使用take(n)返回前n行记录

在RDD的行动操作中，用take(n)返回前n个元素，用top(n)逆序返回前n个元素

在dataframe中使用count()输出dataframe对象的行数。

在dataframe中使用distinct()方法返回一个不包含重复记录的Dataframe

在RDD的转换操作中，也是使用distinct()方法进行去重操作

DataFrame的操作

val sqlContext = new org.apache.spark.sql.SQLContext(sc)//初始化SQLContext对象为sqlContext，sqlContext对象是Spark SQL的入口点。
var df = sqlContext.read.format("json").load("D:\\\\Long\\\\Spark\\\\employee.json")//使用json格式创建DataFrame
//1.查询所有的数据
df.show
+----+---+-----+
| age| id| name|
+----+---+-----+
|  36|  1| Ella|
|  29|  2|  Bob|
|  29|  3| Jack|
|  28|  4|  Jim|
|  28|  4|  Jim|
|null|  5|Damon|
|null|  5|Damon|
+----+---+-----+
//2.查询所有的数据，并去除重复的数据
df.distinct().show
+----+---+-----+
| age| id| name|
+----+---+-----+
|  36|  1| Ella|
|  29|  3| Jack|
|null|  5|Damon|
|  29|  2|  Bob|
|  28|  4|  Jim|
+----+---+-----+
//2.查询所有的数据，打印时去除id字段
df.select("age","name").show
+----+-----+
| age| name|
+----+-----+
|  36| Ella|
|  29|  Bob|
|  29| Jack|
|  28|  Jim|
|  28|  Jim|
|null|Damon|
|null|Damon|
+----+-----+
//或者使用 df.drop("id").show 使用drop一次只能删除一个字段
//3.筛选出age>30的记录
df.where("age>30").show   //注意where里的写法 "age > 30"
+---+---+-----+
|age| id| name|
+---+---+-----+
| 36|  1| Ella|
+---+---+-----+
//4.将数据按age分组
df.groupBy("age").count.show
+----+-----+
| age|count|
+----+-----+
|  29|    2|
|null|    2|
|  28|    2|
|  36|    1|
+----+-----+
//5.将数据按name升序排列
 df.sort("name").show
+----+---+-----+
| age| id| name|
+----+---+-----+
|  36|  1| Ella|
|  29|  2|  Bob|
|null|  5|Damon|
|null|  5|Damon|
|  29|  3| Jack|
|  28|  4|  Jim|
|  28|  4|  Jim|
+----+---+-----+
//6.取出前3行数据
df.take(3)//take返回的是前若干行数据的数组
Array[org.apache.spark.sql.Row] = Array([36,1, Ella], [29,2,Bob], [29,3,Jack])
df.limit(3).show//而使用limit是返回前若干行数组组成的Dateframe对象，可以用show方法进行查看
+---+---+-----+
|age| id| name|
+---+---+-----+
| 36|  1| Ella|
| 29|  2|  Bob|
| 29|  3| Jack|
+---+---+-----+
//7.查询所有记录的name列，并为其取别名为username
df.select(df("name").as("username")).show //这里必须df(...).as
+--------+
|username|
+--------+
|    Ella|
|     Bob|
|    Jack|
|     Jim|
|     Jim|
|   Damon|
|   Damon|
+--------+
//8.查询年龄age的平均值
 df.agg(avg("age")).show   //使用mean也是等价的df.agg(mean("age")).show
+--------+
|avg(age)|
+--------+
|    30.0|
+--------+
//9.查询年龄age的最小值
 df.agg(min("age")).show
+--------+
|min(age)|
+--------+
|      28|
+--------+

---

RDD操作
数据集

Aaron,OperatingSystem,100
Aaron,Python,50
Aaron,ComputerNetwork,30
....
共1000+数据

1. 该系总共有多少学生

   var rdd = sc.textFile("D:\\\\Long\\\\Spark\\\\chat4.txt")//使用文本文件创建RDD
   var per = rdd.map(x => (x.split(",")(0)))//数组0下标表示的是姓名
   per.distinct().count()//distinct是转换操作，目的是为了去重，count是行动操作，会直接统计元素的个数并输出
   ——————————————————————————————————————————
   var tem = rdd.maprow => var splits = row.split(",");(splits(0),splits(1),splits(2).toInt)
   tem.map(x => x._1).distinct().count
   

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-O4hvzEt8-1656335295914)(C:/Users/dell/AppData/Roaming/Typora/typora-user-images/image-20220625171550352.png)]

2. 该系共开设了多少门课程

   var per = rdd.map(x => (x.split(",")(1)))//下标1是课程名
   per.distinct().count()
   ——————————————————————————————————————————
   tem.map(x => x._1).distinct().count
   

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zOeKimyI-1656335295916)(C:/Users/dell/AppData/Roaming/Typora/typora-user-images/image-20220625171756016.png)]

3. Tom 同学的总成绩平均分是多少

   name.map(row => (row.split(",")(0),row.split(",")(2).toInt)).mapValues(x => (x,1)).reduceByKey((x,y) => (x._1+y._1,x._2+y._2)).mapValues(x => (x._1/x._2)).collect
   ————————————————————————————————————————————
   tem.filter(x => x._1 == "Tom").map(x => (x._1,x._3)).mapValues(x => (x,1)).reduceByKey((x,y) => (x._1+y._1,x._2+y._2)).mapValues(x => x._1/x._2).fore
   ach(println)
   

4. 求每名同学的选修的课程门数

   rdd.map(row => (row.split(",")(0),row.split(",")(1))).mapValues(x => (x,1)).reduceByKey((x,y) => ("",x._2+y._2)).mapValues(x => x._2).foreach(println)//("",x._2+y._2)前的这个双引号一定要加，不然会报类型匹配错误
   
   ——————————————————————————————————————————————————————————————————————————————————————————————————————————————
   tem.map(x => (x._1,x._2)).mapValues(x => (x,1)).reduceByKey((x,y) => ("",x._2+y._2)).mapValues(x => x._2).foreach(println)
   

5. 该系 DataBase 课程共有多少人选修

   val total = rdd.filter(row => row.split(",")(1)=="DataBase")
   total.count()
   //total.map(row => (row.split(",")(1),row.split(",")(0))).mapValues(x => (x,1)).reduceByKey((x,y) => ("",x._2+y._2)).mapVules(x =>x._2).foreach(println)
   ——————————————————————————————————————————————————————————————————————————————
   tem.filter(x => x._2 == "DataBase").map(x =>x._1).distinct().count
   

6. 各门课程的平均分是多少

   rdd.map(row => (row.split(",")(1),row.split(",")(2).toInt)).mapValues(x => (x,1)).reduceByKey((x1,x2) => (x1._1+x2._1,x1._2+x2._2)).mapValues(x => x._1/x._2).foreach(println)
   
   ——————————————————————————————————————————————————————————————————————————————————————————————————
   tem.map(x => (x._2,x._3)).mapValues(x => (x,1)).reduceByKey((x,y) => (x._1+y._1,x._2+y._2)).mapValues(x => x._1/x._2).foreach(println)
   

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ulFvphFY-1656335295916)(C:/Users/dell/AppData/Roaming/Typora/typora-user-images/image-20220625200010181.png)]

7. 词频统计

var rdd = sc.textFile("D:\\\\Long\\\\Spark\\\\word.txt")
var ree = rdd.flatMap(row => row.split(" "))//这里用flatMap而不是map的原因是因为每一行有很多单词，有很多行。用flatMap就可以将不同的集合合并为一个集合
var ree1 = ree.map(word => (word,1))//这里用map而不是mapValues是因为这里不是键值对形式，所以无法使用mapValues
ree1.reduceByKey((x,y) => x+y).mapValues(x => x).foreach(println)