1.实验目的
(1)定制软件开发掌握使用访问本地文件和HDFS定制软件开发文件的方法
(2)掌握Spark定制软件开发应用程序的编写、定制软件开发编译和运行方法
2.实验平台
(1)操作系统:Ubuntu18.04(或Ubuntu16.04);
(2)Spark版本:2.4.0;
(3)版本:3.1.3。
3.实验步骤
(1)Spark读取文件系统的数据
1.在spark-shell中读取Linux系统本地文件“/home/hadoop/test.txt”,然后统计出文件的行数;
2.在spark-shell中读取系统文件“/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")
- }
- }