实验4-RDD编程初级实践-题目-厦门大学-林子雨-Spark编程基础(Scala版)
1.spark-shell 交互式编程
数据集下载专区地址:http://dblab.xmu.edu.cn/post/spark/
将chapter5-data1.txt文件放置在usr/local/sparkdata/中,新建/usr/local/sparkdata文件夹
将数据集文件放置在sparkdata中,通过Filezilla上传到文件里。
(1)该系总共有多少学生;
1
2
3
|
val lines = sc.textFile(
"file:///usr/local/sparkdata/Data01.txt"
)
val par = lines.map(row=>row.split(
","
)(0))
val distinct_par = par.distinct()<br><br>distinct_par.count
|
(2)该系共开设来多少门课程;
1
2
3
4
|
val lines = sc.textFile(
"file:///usr/local/sparkdata/Data01.txt"
)
val par = lines.map(row=>row.split(
","
)(1))
val distinct_par = par.distinct()
distinct_par.count
|
(3)Tom 同学的总成绩平均分是多少;
1
2
3
4
5
6
7
|
val lines = sc.textFile(
"file:///usr/local/sparkdata/Data01.txt"
)
lines.filter(row=>row.split(
","
)(0)==
"Tom"
)
.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()
|
(4)求每名同学的选修的课程门数;
1
2
3
4
5
6
7
8
9
|
val line=sc.textFile(
"file:///usr/local/sparkdata/Data01.txt"
)
line.map(row=>(row.split(
","
)(0),row.split(
","
)(1))).
mapValues(x=>(1)).
reduceByKey((x,y)=>(x+y)).
collect()
|
(5)该系 DataBase 课程共有多少人选修;
1
2
3
4
5
|
val line=sc.textFile(
"file:///usr/local/sparkdata/Data01.txt"
)
line.filter(row=>row.split(
","
)(1)==
"DataBase"
).
count()
|
(6)各门课程的平均分是多少;
1
2
3
4
5
6
7
8
9
10
11
|
val line=sc.textFile(
"file:///usr/local/sparkdata/Data01.txt"
)
line.map(row=>(row.split(
","
)(1),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()
|
(7)使用累加器计算共有多少人选了 DataBase 这门课。
1
2
3
4
5
|
val lines = sc.textFile(
"file:///usr/local/sparkdata/chapter5-data1.txt"
)
val rdd = lines.filter(t=>t.split(
","
)(1)==
"DataBase"
).map(t=>(t.split(
","
)(1),1))
val accum = sc.longAccumulator(
"My Accumulator"
)
rdd.values.
foreach
(t=>accum.add(t))
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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
package sn
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:///usr/local/sparkdata/data42"
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"
)
}
}
|
实验三
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
24
25
26
27
|
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.HashPartitioner
object
AvgScore
{
def main(args:Array[String])
{
val conf =
new
SparkConf().setAppName(
"AvgScore"
)
val sc =
new
SparkContext(conf)
val dataFile =
"file:///usr/local/spark/mycode/avgscore/data"
val data = sc.textFile(dataFile,3)
val res=data.filter(_.trim().length>0).map(line=>(line.split(
" "
)(0).trim(),line.split(
" "
)(1).trim().toInt)).partitionBy(
new
HashPartitioner(1)).groupByKey().map(x=>{
var
n=0
var
sum=0.0
for
(i<-x._2){
sum=sum+i
n=n+1
}
val avg=sum/n
val format=f
"$avg%1.2f"
.toDouble
(x._1,format)
})
res.saveAsTextFile(
"result2"
)
}
}
|