您的位置:首页 > 编程语言 > Java开发

scala IDE for Eclipse开发Spark程序

2017-07-19 00:00 483 查看

1、开发环境准备

scalaIDEforEclipse:版本(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、安装scalaIDEforEclipse

直接默认安装即可。

4、新建一个scalaproject



新建一个TestSpark工程,如图:



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







5、导入spark的所有jar包

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

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



如图:



6、WordCount简单示例

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



words.txt内容如下:



HelloHadoop

HelloBigData

HelloSpark

HelloFlume

HelloKafka

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

右键New------>选择ScalaObject:





本地模式,LocalWordCount.scala代码如下:



importorg.apache.spark.SparkConf

importorg.apache.spark.SparkContext

importorg.apache.spark.rdd.RDD

objectLocalWordCount{

defmain(args:Array[String]){

/**

*第一步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,

*例如说通过setMaster来设置程序要连接的Spark集群的Master的URL,

*如果设置为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差

*(例如只有1G的内存)的初学者

*/

valconf=newSparkConf()//创建SparkConf对象,由于全局只有一个SparkConf所以不需要工厂方法

conf.setAppName("wow,myfirstsparkapp")//设置应用程序的名称,在程序的监控界面可以看得到名称

conf.setMaster("local")//此时程序在本地运行,不需要安装Spark集群

/**

*第二步:创建SparkContext对象

*SparkContext是Spark程序所有功能的唯一入口,无论是采用Scala、Java、Python、R等都必须要有一个

*SparkContext

*SparkContext核心作用:初始化Spark应用程序运行所需要的核心组件,包括DAGScheduler,TaskScheduler,SchedulerBacked,

*同时还会负责Spark程序往Master注册程序等

*SparkContext是整个Spark应用程序中最为至关重要的一个对象

*/

valsc=newSparkContext(conf)//创建SpackContext对象,通过传入SparkConf实例来定制Spark运行的具体参数的配置信息

/**

*第三步:根据具体的数据来源(HDFS,HBase,Local,FileSystem,DB,S3)通过SparkContext来创建RDD

*RDD的创建基本有三种方式,(1)根据外部的数据来源(例如HDFS)(2)根据Scala集合(3)由其它的RDD操作

*数据会被RDD划分为成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴

*/

//读取本地文件并设置为一个Partition

vallines=sc.textFile("words.txt",1)//第一个参数为为本地文件路径,第二个参数minPartitions为最小并行度,这里设为1

//类型推断,也可以写下面方式

//vallines:RDD[String]=sc.textFile("words.txt",1)

/**

*第四步:对初始的RDD进行Transformation级别的处理,例如map,filter等高阶函数

*编程。来进行具体的数据计算

*第4.1步:将每一行的字符串拆分成单个的单词

*/

//对每一行的字符串进行单词拆分并把所有行的结果通过flat合并成一个大的集合

valwords=lines.flatMap{line=>line.split("")}

/**

*第4.2步在单词拆分的基础上,对每个单词实例计数为1,也就是word=>(word,1)tuple

*/

valpairs=words.map{word=>(word,1)}

/**

*第4.3步在每个单词实例计数为1的基础之上统计每个单词在文中出现的总次数

*/

//对相同的key进行value的累加(包括local和Reduce级别的同时Reduce)

valwordCounts=pairs.reduceByKey(_+_)

//打印结果

wordCounts.foreach(wordNumberPair=>println(wordNumberPair._1+":"+wordNumberPair._2))

//释放资源

sc.stop()

}

}

右键RunAs----->ScalaApplication



运行结果如图:



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

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



importorg.apache.spark.SparkConf

importorg.apache.spark.SparkContext

importorg.apache.spark.rdd.RDD

objectClusterWordCount{

defmain(args:Array[String]){

/**

*第一步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,

*例如说通过setMaster来设置程序要连接的Spark集群的Master的URL,

*如果设置为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差

*(例如只有1G的内存)的初学者

*/

valconf=newSparkConf()//创建SparkConf对象,由于全局只有一个SparkConf所以不需要工厂方法

conf.setAppName("wow,myfirstsparkapp")//设置应用程序的名称,在程序的监控界面可以看得到名称

//conf.setMaster("spark://192.168.168.200:7077")//此时程序在Spark集群

/**

*第二步:创建SparkContext对象

*SparkContext是Spark程序所有功能的唯一入口,无论是采用Scala、Java、Python、R等都必须要有一个

*SparkContext

*SparkContext核心作用:初始化Spark应用程序运行所需要的核心组件,包括DAGScheduler,TaskScheduler,SchedulerBacked,

*同时还会负责Spark程序往Master注册程序等

*SparkContext是整个Spark应用程序中最为至关重要的一个对象

*/

valsc=newSparkContext(conf)//创建SpackContext对象,通过传入SparkConf实例来定制Spark运行的具体参数的配置信息

/**

*第三步:根据具体的数据来源(HDFS,HBase,Local,FileSystem,DB,S3)通过SparkContext来创建RDD

*RDD的创建基本有三种方式,(1)根据外部的数据来源(例如HDFS)(2)根据Scala集合(3)由其它的RDD操作

*数据会被RDD划分为成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴

*/

//读取HDFS文件并切分成不同的Partition

vallines=sc.textFile("hdfs://192.168.168.200:9000/input/words.txt")

//vallines=sc.textFile("/index.html")

//类型推断,也可以写下面方式

//vallines:RDD[String]=sc.textFile("hdfs://192.168.168.200:9000/input/words.txt",1)

/**

*第四步:对初始的RDD进行Transformation级别的处理,例如map,filter等高阶函数

*编程。来进行具体的数据计算

*第4.1步:将每一行的字符串拆分成单个的单词

*/

//对每一行的字符串进行单词拆分并把所有行的结果通过flat合并成一个大的集合

valwords=lines.flatMap{line=>line.split("")}

/**

*第4.2步在单词拆分的基础上,对每个单词实例计数为1,也就是word=>(word,1)tuple

*/

valpairs=words.map{word=>(word,1)}

/**

*第4.3步在每个单词实例计数为1的基础之上统计每个单词在文中出现的总次数

*/

//对相同的key进行value的累加(包括local和Reduce级别的同时Reduce)

valwordCounts=pairs.reduceByKey(_+_)

//打印结果

wordCounts.collect.foreach(wordNumberPair=>println(wordNumberPair._1+":"+wordNumberPair._2))

//释放资源

sc.stop()

}

}

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

打包:右击、export、Java、jarFile





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



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

提交spark任务:



cd/usr/local/spark/spark-2.1.1-bin-hadoop2.7

bin/spark-submit--class"ClusterWordCount"--masterlocal[4]myApp/TestSpark.jar



集群模式运行成功!
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签:  scala Eclipse
相关文章推荐