scala IDE for Eclipse开发Spark程序

1、开发环境准备

scala IDE for Eclipse：版本（4.6.1）

官网下载：http://scala-ide.org/download/sdk.html

百度云盘下载：链接：http://pan.baidu.com/s/1c2NAZdA 密码：au8t

scala-2.11.8.msi ：

官网下载：http://www.scala-lang.org/download/2.11.8.html

百度云盘下载：链接：http://pan.baidu.com/s/1dEFlmcL 密码：u2fr

2、scala环境变量配置 

配置path，配置SCALA_HOME/bin目录，如图：

在命令行窗口输入scala，如图：

安装成功！

3、安装scala IDE for Eclipse

直接默认安装即可。

4、新建一个scala project

新建一个TestSpark工程，如图：

这里的scala版本默认使用的是2.12.2，我们要scala的版本设置为2.11.8，

 

5、导入spark的所有jar包

spark-2.1.1-bin-hadoop2.7jars ：此处我使用spark版本是spark-2.1.1-bin-hadoop2.7。

spark版本跟你装在linux上的spark环境是一致的。

如图：

6、WordCount简单示例

6.1在TestSpark工程下新建一个words.txt文件

words.txt内容如下：

1. HelloHadoop
2. HelloBigData
3. HelloSpark
4. HelloFlume
5. HelloKafka

6.2本地模式新建一个LocalWordCount.scala

右键New------>选择Scala Object：

本地模式，LocalWordCount.scala代码如下：

1. import org.apache.spark.SparkConf
2. import org.apache.spark.SparkContext
3. import org.apache.spark.rdd.RDD
4.  
5. objectLocalWordCount{
6. def main(args:Array[String]){
7. /**
8. * 第一步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
9. * 例如说通过setMaster来设置程序要连接的Spark集群的Master的URL，
10. * 如果设置为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差
11. * （例如只有1G的内存）的初学者
12. */
13. val conf =newSparkConf()//创建SparkConf对象，由于全局只有一个SparkConf所以不需要工厂方法
14. conf.setAppName("wow,my first spark app")//设置应用程序的名称，在程序的监控界面可以看得到名称
15. conf.setMaster("local")//此时程序在本地运行，不需要安装Spark集群
16. /**
17. * 第二步：创建SparkContext对象
18. * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala、Java、Python、R等都必须要有一个
19. * SparkContext
20. * SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler，TaskScheduler，SchedulerBacked，
21. * 同时还会负责Spark程序往Master注册程序等
22. * SparkContext是整个Spark应用程序中最为至关重要的一个对象
23. */
24. val sc=newSparkContext(conf)//创建SpackContext对象，通过传入SparkConf实例来定制Spark运行的具体参数的配置信息
25. /**
26. * 第三步：根据具体的数据来源（HDFS，HBase，Local，FileSystem，DB，S3）通过SparkContext来创建RDD
27. * RDD的创建基本有三种方式，（1）根据外部的数据来源（例如HDFS）（2）根据Scala集合（3）由其它的RDD操作
28. * 数据会被RDD划分为成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
29. */
30. //读取本地文件并设置为一个Partition
31. val lines=sc.textFile("words.txt",1)//第一个参数为为本地文件路径，第二个参数minPartitions为最小并行度，这里设为1
32. //类型推断 ,也可以写下面方式
33. // val lines : RDD[String] =sc.textFile("words.txt", 1)
34. /**
35. * 第四步：对初始的RDD进行Transformation级别的处理，例如map，filter等高阶函数
36. * 编程。来进行具体的数据计算
37. * 第4.1步：将每一行的字符串拆分成单个的单词
38. */
39. //对每一行的字符串进行单词拆分并把所有行的结果通过flat合并成一个大的集合
40. val words = lines.flatMap { line => line.split(" ")}
41. /**
42. * 第4.2步在单词拆分的基础上，对每个单词实例计数为1，也就是word=>（word,1）tuple
43. */
44. val pairs = words.map { word =>(word,1)}
45. /**
46. * 第4.3步在每个单词实例计数为1的基础之上统计每个单词在文中出现的总次数
47. */
48. //对相同的key进行value的累加（包括local和Reduce级别的同时Reduce）
49. val wordCounts = pairs.reduceByKey(_+_)
50. //打印结果
51. wordCounts.foreach(wordNumberPair => println(wordNumberPair._1 +":"+wordNumberPair._2))
52. //释放资源
53. sc.stop()
54. }
55. }

右键Run As----->Scala Application

运行结果如图：

6.3集群模式新建一个ClusterWordCount.scala

集群模式----ClusterWordCount.scala代码如下：

1. import org.apache.spark.SparkConf
2. import org.apache.spark.SparkContext
3. import org.apache.spark.rdd.RDD
4.  
5. objectClusterWordCount{
6. def main(args:Array[String]){
7. /**
8. * 第一步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
9. * 例如说通过setMaster来设置程序要连接的Spark集群的Master的URL，
10. * 如果设置为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差
11. * （例如只有1G的内存）的初学者
12. */
13. val conf =newSparkConf()//创建SparkConf对象，由于全局只有一个SparkConf所以不需要工厂方法
14. conf.setAppName("wow,my first spark app")//设置应用程序的名称，在程序的监控界面可以看得到名称
15. //conf.setMaster("spark://192.168.168.200:7077")//此时程序在Spark集群
16. /**
17. * 第二步：创建SparkContext对象
18. * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala、Java、Python、R等都必须要有一个
19. * SparkContext
20. * SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler，TaskScheduler，SchedulerBacked，
21. * 同时还会负责Spark程序往Master注册程序等
22. * SparkContext是整个Spark应用程序中最为至关重要的一个对象
23. */
24. val sc=newSparkContext(conf)//创建SpackContext对象，通过传入SparkConf实例来定制Spark运行的具体参数的配置信息
25. /**
26. * 第三步：根据具体的数据来源（HDFS，HBase，Local，FileSystem，DB，S3）通过SparkContext来创建RDD
27. * RDD的创建基本有三种方式，（1）根据外部的数据来源（例如HDFS）（2）根据Scala集合（3）由其它的RDD操作
28. * 数据会被RDD划分为成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
29. */
30. //读取HDFS文件并切分成不同的Partition
31.  
32. val lines=sc.textFile("hdfs://192.168.168.200:9000/input/words.txt")
33. //val lines=sc.textFile("/index.html")
34. //类型推断 ,也可以写下面方式
35. // val lines : RDD[String] =sc.textFile("hdfs://192.168.168.200:9000/input/words.txt", 1)
36. /**
37. * 第四步：对初始的RDD进行Transformation级别的处理，例如map，filter等高阶函数
38. * 编程。来进行具体的数据计算
39. * 第4.1步：将每一行的字符串拆分成单个的单词
40. */
41. //对每一行的字符串进行单词拆分并把所有行的结果通过flat合并成一个大的集合
42. val words = lines.flatMap { line => line.split(" ")}
43. /**
44. * 第4.2步在单词拆分的基础上，对每个单词实例计数为1，也就是word=>（word,1）tuple
45. */
46. val pairs = words.map { word =>(word,1)}
47. /**
48. * 第4.3步在每个单词实例计数为1的基础之上统计每个单词在文中出现的总次数
49. */
50. //对相同的key进行value的累加（包括local和Reduce级别的同时Reduce）
51. val wordCounts = pairs.reduceByKey(_+_)
52. //打印结果
53. wordCounts.collect.foreach(wordNumberPair => println(wordNumberPair._1 +":"+wordNumberPair._2))
54. //释放资源
55. sc.stop()
56. }
57. }

集群模式要运行在Spark集群环境下（Linux系统中）

打包 ：右击、export、Java 、jar File 

把TestSpark.jar包上传到spark集群服务器的 spark_home下的myApp下：

确保hdfs://192.168.168.200:9000/input/words.txt路径文件存在。

提交spark任务：

1. cd /usr/local/spark/spark-2.1.1-bin-hadoop2.7
2. bin/spark-submit --class"ClusterWordCount"--master local[4] myApp/TestSpark.jar

集群模式运行成功！