大三寒假生活7

今天你完成实验四前两个实验。但是!!!在跟eclipse的scala插件做斗争的时候,把hadoop的插件给搞没了,按着林子雨老师的教程走了一遍还是不行,不知不觉又搞了一下午,还是没解决。明天把实验四完成,做一些实验五,希望明天可以把eclipse的问题解决。

1.spark-shell 交互式编程
请到本教程官网的“下载专区”的“数据集”中下载 chapter5-data1.txt,该数据集包含
了某大学计算机系的成绩,数据格式如下所示:
Tom,DataBase,80
Tom,Algorithm,50
Tom,DataStructure,60
Jim,DataBase,90
Jim,Algorithm,60
Jim,DataStructure,80
……
请根据给定的实验数据,在 spark-shell 中通过编程来计算以下内容:
 
(1)该系总共有多少学生; 
 
 
val lines = sc.textFile("file:///usr/local/hadoop/Data01.txt")
val par = lines.map(row=>row.split(",")(0)) 
val distinct_par = par.distinct() //去重操作
distinct_par.count //取得总数
(2)该系共开设来多少门课程;
 

val lines = sc.textFile("file:///usr/local/hadoop/Data01.txt")
val par = lines.map(row=>row.split(",")(1)) 
val distinct_par = par.distinct() 
distinct_par.count
(3)Tom 同学的总成绩平均分是多少;
 
val lines = sc.textFile("file:///usr/local/hadoop/Data01.txt")
val pare = lines.filter(row=>row.split(",")(0)=="Tom")
pare.foreach(println)
pare.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()
//res9: Array[(String, Int)] = Array((Tom,30))
(4)求每名同学的选修的课程门数;
 
val lines = sc.textFile("file:///usr/local/hadoop/Data01.txt")
val pare = lines.map(row=>(row.split(",")(0),row.split(",")(1)))
pare.mapValues(x => (x,1)).reduceByKey((x,y) => (" ",x._2 + y._2)).mapValues(x => 
x._2).foreach(println)
(5)该系 DataBase 课程共有多少人选修;
 
val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val pare = lines.filter(row=>row.split(",")(1)=="DataBase")
pare.count
res1: Long = 126
(6)各门课程的平均分是多少;
 
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))
(7)使用累加器计算共有多少人选了 DataBase 这门课。
 
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")
pare.values.foreach(x => accum.add(x))
accum.value

2.编写独立应用程序实现数据去重 
对于两个输入文件 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

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.HashPartitioner
object RemDup {
 def main(args: Array[String]) {
 val conf = new SparkConf().setAppName("RemDup")
 val sc = new SparkContext(conf)
 val dataFile = "file:///home/charles/data"
 val data = sc.textFile(dataFile,2)
 val res = data.filter(_.trim().length>0).map(line=>(line.trim,"")).partitionBy(new 
HashPartitioner(1)).groupByKey().sortByKey().keys
 res.saveAsTextFile("result")
 } }

相关推荐