1.实验目的

（1）掌握使用Spark访问本地文件和HDFS文件的方法

（2）掌握Spark应用程序的编写、编译和运行方法

2.实验平台

（1）操作系统：Ubuntu18.04（或Ubuntu16.04）；

（2）Spark版本：2.4.0；

（3）Hadoop版本：3.1.3。

3.实验步骤

（1）Spark读取文件系统的数据

1.在spark-shell中读取Linux系统本地文件“/home/hadoop/test.txt”，然后统计出文件的行数；

2.在spark-shell中读取HDFS系统文件“/user/hadoop/test.txt”（如果该文件不存在，请先创建），然后，统计出文件的行数；

 

 

3.编写独立应用程序（推荐使用Scala语言），读取HDFS系统文件“/user/hadoop/test.txt”（如果该文件不存在，请先创建），然后，统计出文件的行数；通过sbt工具将整个应用程序编译打包成 JAR包，并将生成的JAR包通过 spark-submit 提交到 Spark 中运行命令。

 

 SimpleApp.scala 和 simple.sbt内容：

/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
 
object SimpleApp {
    def main(args: Array[String]) {
        val logFile = " hdfs://localhost:9000/user/hadoop/test.txt"
        val conf = new SparkConf().setAppName("Simple Application")
        val sc = new SparkContext(conf)
        val logData = sc.textFile(logFile, 2)
        val num = logData.count()
        printf("The num of this file is %d rows", num)
    }
}

name := "Simple Project"
version := "1.0"
scalaVersion := "2.11.12"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.4.0"

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

对于两个输入文件A和B，编写Spark独立应用程序（推荐使用Scala语言），对两个文件进行合并，并剔除其中重复的内容，得到一个新文件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

 

remdup.scala

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")
    }
}

（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)

avgscore.scala

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:///home/hadoop/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("resultAvgscore")
    }
}