一、實驗目的
（1）熟悉 Spark 的 RDD 基本操作及鍵值對操作；
（2）熟悉使用 RDD 編程解決實際具體問題的方法。
二、實驗平台
操作系統：centos6.4
Spark 版本：1.5.0
三、實驗內容
實驗一：
1．spark-shell 交互式編程
請到本教程官網的“下載專區”的“數據集”中下載 chapter5-data1.txt，該數據集包含 了某大學計算機系的成績，數據格式如下所示：
首先開始我們的第一步，打開linux系統中的終端。
請根據給定的實驗數據，在 spark-shell 中通過編程來計算以下內容：
將Data01.txt文件放置在usr/local/sparkdata/中
新建/usr/local/sparkdata文件夾
mkdir /usr/local/sparkdata
將Data01.txt文件放置在sparkdata中
發現權限不夠，給/usr/local/sparkdata賦予操作權限
chmod 777 /usr/local/spakrdata
之后將Data01.txt文件移動到sparkdata中
（1）該系總共有多少學生；
val lines = sc.textFile("file:///usr/local/sparkdata/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/sparkdata/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/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）求每名同學的選修的課程門數；
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 課程共有多少人選修；
val line=sc.textFile("file:///usr/local/sparkdata/Data01.txt")
line.filter(row=>row.split(",")(1)=="DataBase").
count()
（6）各門課程的平均分是多少；
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 這門課。
val lines = sc.textFile("file:///usr/local/sparkdata/Data01.txt")
val pare = lines.filter(row=>row.split(",")(1)=="DataBase").
map(row=>(row.split(",")(1),1))
val accum =sc.accumulator(0)
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
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)
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") } }