1、实验描述
- 利用Scala语言开发Spark WordCount程序
- 实验时长：
- 45分钟
- 主要步骤：
- 创建spark项目
- 编写wordcount 示例程序
- 运行Scala 程序
- 查看实验结果
2、实验环境
- 虚拟机数量：3(一主两从，主机名分别为：master、slave01、slave02)
- 系统版本：Centos 7.5
- Hadoop版本：Apache Hadoop 2.7.3
- Spark版本：Apache Spark 2.1.1
- JDK 版本：1.8.0_131
- Scala版本： scala2.11.11
- IDEA版本：ideaIC-2017.2.7
3、相关技能
- Spark集成开发环境IDEA开发程序
- Scala编写Spark程序
4、知识点
- 常见linux命令的使用
- Scala IDEA编程
- Spark RDD
- Scala原理
5、实验效果
spark RDD 单词计数结果查看操作最终效果如下图:
6、实验步骤
6.1打开虚拟机并启动Hadoop集群
6.1.1在master启动Hadoop集群
[zkpk@master ~]$ cd hadoop-2.7.3/
[zkpk@master hadoop-2.7.3]$ sbin/start-all.sh
[zkpk@master hadoop-2.7.3]$ hdfs dfsadmin -safemode leave
6.1.2在master上运行jps，确认NameNode, SecondaryNameNode, ResourceManager进程启动
6.1.3在slave01上运行jps，确认DataNode, NodeManager进程启动
6.1.4在slave02上运行jps，确认DataNode, NodeManager进程启动
6.2打开IDEA，配置软件包依赖，创建工程
[zkpk@master ~]$ cd idea-IC-172.4574.19/
[zkpk@master idea-IC-172.4574.19]$ nohup bin/idea.sh &
6.2.1进入如下界面，点击 Create New Project
6.2.2进入如下图界面，按照图标依次点击，最后点击next
6.2.3依次输入GroupId和ArtifactId和Version的值，随后点击next
6.2.4进入如下界面，设置本地Maven项目的setting.xml文件和warehouse仓库，点击next按钮
6.2.4.1本地setting.xml文件在/home/zkpk/apache-maven-3.5.0/conf目录下
6.2.4.2本地仓库文件夹warehouse在/home/zkpk/apache-maven-3.5.0/warehouse
6.2.5进入如下界面，输入工程名称spark_test，然后点击next
6.2.6进入如下界面，即表示工程spark_test创建成功
6.2.7如5.1.6步骤所示，工程创建完成后会自动打开一个名为zkpk的xml文件，删除如下图标红部分的依赖
6.2.8在xml文件中找到properties配置项，修改scala版本号（此处对应scala安装版本），并添加spark版本号（此处对应spark安装版本）
6.2.9找到dependency配置项，添加如下图标红部分的配置，分别是scala依赖和spark依赖，${scala.version}表示上述配置的scala.version变量
6.2.10一般修改pom.xml文件后，会提示enable auto-import，点击即可，如果没有提示，则可以点击工程名，依次选择Maven—->Reimport，即可根据pom.xml文件导入依赖包
6.2.11设置语言环境language level，点击菜单栏中的file，选择Project Structure
6.2.12弹出如下对话框，选择Modules，选择Language level为8，然后点击Apply，点击OK
6.2.13设置Java Compiler环境，点击菜单栏中的file，选择Setting
6.2.14弹出如下对话框，依次选择Build，Execution—->Compiler—->Java Compiler，设置图中的Project bytecode version为1.8，设置图中的Target bytecode version为1.8，然后依次点击Apply和OK
6.2.15如下图删除测试环境test中的测试类
6.2.16如下图删除，main文件夹中，包名下的App文件
6.2.17至此，Spark Maven工程创建完毕
6.3编写Scala程序完成Spark单词计数
6.3.1如下图依次打开src—>main—>scala，在org.zkpk.lab上点击右键，创建Scala Class
6.3.2弹出如下对话框，输入类名ScalaWordCount，点击ok
6.3.3在类ScalaWordCount中新建伴生对象object ScalaWordCount
6.3.4在伴生对象object ScalaWordCount中创建main方法
6.3.5在main方法中创建列表List对象并赋值给常量list，列表中包含4个元素，分别是：“hello
hi hi spark”，“hello spark hello hi sparksql”，“hello hi hi sparkstreaming”，“hello hi sparkgraphx”
6.3.6创建SparkConf对象，对Spark运行属性进行配置，调用该对象的setAppName方法设置Spark程序的名称为“word-count”，调用setMaster方法设置Spark程序运行模式，一般分为两种：本地模式和yarn模式，这里我们采用本地模式，参数为“local[*]”，属性设置完成后赋值给常量sparkConf
6.3.7创建SparkContext，参数为sparkconf，赋值给常量sc，该对象是Spark程序的入口
6.3.8调用SparkContext对象sc的方法parallelize，参数为列表对象list，该方法使用现成的scala集合生成RDD
lines，类型为String（RDD为Spark计算中的最小数据单元），该RDD存储元素是字符串语句
6.3.9调用RDD对象lines的flatMap方法按照空格切分RDD中的字符串元素，并存入新的RDD对象words中，参数类型为String，该RDD存储的元素是每一个单词
6.3.10调用RDD对象words的map方法，将RDD中的每一个单词转换为kv对，key是String类型的单词，value是Int类型的1，并赋值给新的RDD对象wordAndOne，参数为（String，Int）类型键值对
6.3.11调用RDD对象wordAndOne的reduceByKey方法，传入的参数为两个Int类型变量，该方法将RDD中的元素按照Key进行分组，将同一组中的value值进行聚合操作，得到valueRet，最终返回（key，valueRet）键值对，并赋值给新的RDD对象wordAndNum，参数为（String，Int）类型键值对
6.3.12调用RDD对象wordAndNum的sortBy方法，第一个参数为kv对中的value，即单词出现次数，第二个参数为boolean类型，true表示升序，false表示降序
6.3.13调用ret对象的collect方法，获取集合中的元素，再调用mkString方法，参数为“，”，将集合中的元素用逗号连接成字符串，调用println方法打印输出在控制台
6.3.14调用ret对象的saveAsTextFile，该方法的参数为运行时指定的参数，此方法的用处是将Spark程序运行的结果保存到指定路径，一般是把结果保存到HDFS中，所以这里的参数定义为：
hdfs://master:9000/sparktest
6.3.14.1HDFS根目录中不能存在sparktest此目录，spark程序会自动创建该目录
6.3.15调用SparkContext对象sc的stop方法，释放spark程序所占用的资源
6.4运行程序，查看结果
6.4.1程序编辑完成后，点击菜单栏上的Run按钮，选择Run…
6.4.2弹出对话框，选择第一个Edit Configurations
6.4.3进入运行参数配置页面，点击+号按钮，选择Application，进入Application配置界面
6.4.4指定Application的名称Name，主函数Main Class，参数arguments和运行模式VM
options（由于在编写程序是已经指定了本地运行模式，所以这里的运行模式不指定），最后点击运行
6.4.5查看控制台输出
6.4.6查看HDFS输出
7、参考答案
代码清单ScalaWordCount
package org.zkpk.lab
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf,SparkContext}
class ScalaWordCount {
}
object ScalaWordCount{
def main(args:Array[String]):Unit={
val list =List("hello hi hi spark",
"hello spark hello hi sparksql",
"hello hi hi sparkstreaming",
"hello hi sparkgraphx")
val sparkConf=new SparkConf().setAppName("word-count").setMaster("local[*]")
val sc=new SparkContext(sparkConf)
val lines:RDD[String]=sc.parallelize(list)
val words:RDD[String]=lines.flatMap((line:String) => {line.split(" ")})
val wordAndOne:RDD[(String,Int)]=words.map((word:String) => {(word,1)})
val wordAndNum:RDD[(String,Int)]=wordAndOne.reduceByKey((count1:Int,count2:Int) => {count1+count2})
val ret=wordAndNum.sortBy(kv=>kv._2,false)
println(ret.collect().mkString(","))
ret.saveAsTextFile(args(0))
sc.stop()
}
}
8、总结
本实验是利用scala语言完成spark单词计数，快速掌握Spark程序Scala编程
