spark-shell的Scala的一些方法详解

Tom,DataBase,80

Tom,Algorithm,50

Tom,DataStructure,60

Jim,DataBase,90

Jim,Algorithm,60

Jim,DataStructure,80

.......

根据给定的数据在spark-shell中通过编程来计算以下内容

（1） 该系总共有多少学生；

1 2 3 4

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val par = lines.map(row=>row.split(",")(0)) val distinct_par = par.distinct() //去重操作 distinct_par.count //取得总数

　　

答案为：265 人

（2） 该系共开设来多少门课程；

1 2 3 4

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val par = lines.map(row=>row.split(",")(1))//根据，切分的每行数据的第二列进行map val distinct_par = par.distinct()//去重 distinct_par.count//取总数

　　答案为 8 门

（3） Tom 同学的总成绩平均分是多少;

1 2 3 4 5 6 7 8 9 10 11 12 13 14

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val pare = lines.filter(row=>row.split(",")(0)=="Tom") pare.foreach(println) Tom,DataBase,26 Tom,Algorithm,12 Tom,OperatingSystem,16 Tom,Python,40 Tom,Software,60 pare.map(row=>(row.split(",")(0),row.split(",")(2).toInt)) .mapValues(x=>(x,1)).//mapValues是对值的操作,不操作key使数据变成(Tom,（26,1）) reduceByKey((x,y) => (x._1+y._1,x._2 + y._2))//接着需要按key进行reduce，让key合并当将Tom进行reduce后 这里的(x,y) 表示的是(26,1)(12,1) .mapValues(x => (x._1 / x._2))//接着要对value进行操作，用mapValues()就行啦 .collect() //res9: Array[(String, Int)] = Array((Tom,30))

　　Tom 同学的平均分为 30 分

（4） 求每名同学的选修的课程门数；

1 2 3

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val pare = lines.map(row=>(row.split(",")(0),row.split(",")(1))) pare.mapValues(x => (x,1))//数据变为(Tom,(DataBase,1)),(Tom,(Algorithm,1)),(Tom,(OperatingSystem,1)),<em id="__mceDel"><em id="__mceDel"><em id="__mceDel">(Tom,(Python,1)),</em></em></em><em id="__mceDel" style="line-height: 1.5;"><em id="__mceDel"><em id="__mceDel"><em id="__mceDel"><em id="__mceDel">(Tom,(Software,1))</em></em></em></em></em>

1	.reduceByKey((x,y) => (" ",x._2 + y._2))//<em id="__mceDel" style="line-height: 1.5;">数据变为(Tom,( ,5))</em>

1	.mapValues(x =>x._2)<em id="__mceDel" style="line-height: 1.5;">//<em id="__mceDel">数据变为(Tom, 5)</em></em>

1	<br>.foreach(println)

1

　　答案共 265 行

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

1 2 3 4

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val pare = lines.filter(row=>row.split(",")(1)=="DataBase")filter方法允许你提供一个判断条件（函数），来过滤集合元素 pare.count res1: Long = 126

　　答案为 126 人

（6） 各门课程的平均分是多少；

1 2 3 4 5

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val pare = lines.map(row=>(row.split(",")(1),row.split(",")(2).toInt)) pare.mapValues(x=>(x,1)).reduceByKey((x,y) => (x._1+y._1,x._2 + y._2)).mapValues(x => (x._1 / x._2)).collect() res0: Array[(String, Int)] = Array((Python,57), (OperatingSystem,54), (CLanguage,50), (Software,50), (Algorithm,48), (DataStructure,47), (DataBase,50), (ComputerNetwork,51))

　　答案为： (CLanguage,50) (Python,57) (Software,50) (OperatingSystem,54) (Algorithm,48) (DataStructure,47) (DataBase,50) (ComputerNetwork,51)

（7）使用累加器计算共有多少人选了 DataBase 这门课。

1 2 3 4 5 6

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt") val pare = lines.filter(row=>row.split(",")(1)=="DataBase").map(row=>(row.split(",")(1),1)) val accum = sc.longAccumulator("My Accumulator")//累加器函数Accumulator pare.values.foreach(x => accum.add(x)) accum.value res19: Long = 126

　　答案：共有 126 人

2.编写独立应用程序实现数据去重

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

对于两个输入文件 A 和 B，编写 Spark 独立应用程序，对两个文件进行合并，并剔除其 中重复的内容，得到一个新文件 C。下面是输入文件和输出文件的一个样例，供参考。 输入文件 A 的样例如下： 20170101 x 20170102 y 20170103 x 20170104 y 20170105 z 20170106 z 输入文件 B 的样例如下： 20170101 y 20170102 y 20170103 x 20170104 z 20170105 y 根据输入的文件 A 和 B 合并得到的输出文件 C 的样例如下： 20170101 x 20170101 y 20170102 y 20170103 x 20170104 y 20170104 z 20170105 y 20170105 z 20170106 z

　　eclipse代码

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

package my.scala import org.apache.spark.{SparkConf, SparkContext} object case2 {   def main(args: Array[String]): Unit = {     val conf = new SparkConf().setMaster("local").setAppName("reduce")     val sc = new SparkContext(conf)     sc.setLogLevel("ERROR")     //获取数据     val two = sc.textFile("hdfs://192.168.85.128:9000/quchong")     two.filter(_.trim().length>0) //需要有空格。         .map(line=>(line.trim,""))//全部值当key，(key value,"")           .groupByKey()//groupByKey,过滤重复的key value ，发送到总机器上汇总               .sortByKey() //按key value的自然顺序排序                   .keys.collect().foreach(println) //所有的keys变成数组再输出     //第二种有风险     two.filter(_.trim().length>0)           .map(line=>(line.trim,"1"))             .distinct()                 .reduceByKey(_+_)                     .sortByKey()                         .foreach(println)       //reduceByKey,在本机suffle后,再发送一个总map，发送到一个总机器上汇总，（汇总要压力小）     //groupByKey,发送本机所有的map,在一个机器上汇总（汇总压力大）     //如果数据在不同的机器上，则会出现先重复数据，distinct，reduceBykey，只是在本机上去重，谨慎一点的话，在reduceByKey后面需要加多一个distinct     } }

　　

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

3.编写独立应用程序实现求平均值问题 每个输入文件表示班级学生某个学科的成绩，每行内容由两个字段组成，第一个是学生 名字，第二个是学生的成绩；编写 Spark 独立应用程序求出所有学生的平均成绩，并输出到 一个新文件中。下面是输入文件和输出文件的一个样例，供参考。 Algorithm 成绩： 小明 92 小红 87 小新 82 小丽 90 Database 成绩： 小明 95 小红 81 小新 89 小丽 85 Python 成绩： 小明 82 小红 83 小新 94 小丽 91 平均成绩如下：  (小红,83.67)  (小新,88.33)  (小明,89.67) (小丽,88.67)

　　

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

package my.scala import org.apache.spark.{SparkConf, SparkContext} object pingjunzhi {   def main(args: Array[String]): Unit = {     val conf = new SparkConf().setMaster("local").setAppName("reduce")     val sc = new SparkContext(conf)     sc.setLogLevel("ERROR")     val fourth = sc.textFile("hdfs://192.168.85.128:9000/pingjunzhi")    val res = fourth.filter(_.trim().length>0).map(line=>(line.split(" ")(0).trim(),line.split(" ")(1).trim().toInt)).groupByKey().map(x => {    var num = 0.0    var sum = 0    for(i <- x._2){     sum = sum + i     num = num +1    }    val avg = sum/num    val format = f"$avg%1.2f".toDouble    (x._1,format)  }).collect.foreach(x => println(x._1+" "+x._2))   } }