Eclipse下开发Scala(DT大数据梦工厂)

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Eclipse下开发Scala(DT大数据梦工厂)相关的知识,希望对你有一定的参考价值。

本讲主要内容:环境安装、配置、本地模式、集群模式、自动化脚本、web状态监控

==========单机============

开发工具开发

下载最新版Scala For Eclipse

1、建立工程,修改scala编译版本

技术分享

2、加入Spark1.6.0的jar文件依赖

下载 http://apache.opencas.org/spark/spark-1.6.0/spark-1.6.0-bin-hadoop2.6.tgz

spark-assembly-1.6.0-hadoop2.6.0.jar


3、找到依赖的spark jar文件并导入到eclipse中的jar依赖中

4、src下建立spark工程包com.dt.spark

5、创建scala入口类

技术分享

6、把class编程object,并编写main入口方法

找不到类crtl+shift+o

例子代码:

package com.dt.spark

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

/**

 * 使用Scala开发本地测试的spark wordcount程序

 * @author DT_大数据梦工厂

 * 新浪微博:http://weibo.com/ilovepains

 * */

object WordCount {

  def main(args:Array[String]){

    /**

     * 1、创建Spark配置对象SparkConf,设置Spark程序的运行时的程序配置信息

     * 例如:通过setMaster来设置程序要连接的Spark集群的url,

     * 如果设置为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如只有1G的内存)的初学者

     */

    val conf = new SparkConf()//创建SparkConf对象

    conf.setAppName("My First Spark App!")//设置应用程序的名称,在程序运行的监控界面可以看到名称

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

    /**

     * 2、创建SparkContext对象

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

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

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

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

     */

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

    /**

     * 3、根据具体的数据来源(HDFS、HBase、Local FS、DB、S3等)通过SparkContext来创建RDD

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

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

     */

    //sc.textFile(文件路径,最小并行度)

    //val lines: RDD[String],通过类型推断得到lines是String类型的RDD

    //val lines: RDD[String] = sc.textFile("F:/安装文件/操作系统/spark-1.6.0-bin-hadoop2.6/README.md", 1)

    val lines = sc.textFile("F:/安装文件/操作系统/spark-1.6.0-bin-hadoop2.6/README.md", 1)

    /**

     * 4、对初始的RDD进行transformation级别的处理,例如map、filter等高阶函数的编程,来进行具体的数据计算

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

     */

    val words = lines.flatMap { line => line.split(" ") }//对每一行的字符串进行单次拆分并把所有行的拆分结果通过flat合并成一个大的单次集合

    /**

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

     */

    val pairs = words.map { word => (word,1) }//其实编程了Tuple  (word,1)

    /**

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

     */

    val wordCounts = pairs.reduceByKey(_+_)//对相同的key,进行value的累加(包括local和Reducer级别同时reduce)

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

    /**

     * 5、释放相关资源

     */

    sc.stop()

  }

}

我的结果不是老师的结果,暂时不管

另外里面开始报错了,一定会报错,这个错误是正确的,因为运行的时候要找Hadoop,没有找到,但是这不是程序错误,也不影响我们的任何功能

==========集群============

package com.dt.spark

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

/**

 * 使用Scala开发集群测试的spark wordcount程序

 * @author DT_大数据梦工厂

 * 新浪微博:http://weibo.com/ilovepains

 * */

object WordCount_Cluster {

  def main(args:Array[String]){

    /**

     * 1、创建Spark配置对象SparkConf,设置Spark程序的运行时的程序配置信息

     * 例如:通过setMaster来设置程序要连接的Spark集群的url,

     * 如果设置为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如只有1G的内存)的初学者

     */

    val conf = new SparkConf()//创建SparkConf对象

    conf.setAppName("My First Spark App!")//设置应用程序的名称,在程序运行的监控界面可以看到名称

    //conf.setMaster("spark://Master:7077")//此时必须要运行在Master:7077上,之后运行可手工配置

    /**

     * 2、创建SparkContext对象

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

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

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

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

     */

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

    /**

     * 3、根据具体的数据来源(HDFS、HBase、Local FS、DB、S3等)通过SparkContext来创建RDD

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

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

     */

    //sc.textFile(文件路径,最小并行度)

    //val lines: RDD[String],通过类型推断得到lines是String类型的RDD

    //val lines = sc.textFile("/library/wordcount/input/Data", 1)//读取HDFS文件,并切分成不同的Partition

    val lines = sc.textFile("/historyserverforSpark/README.md", 1)

    /**

     * 4、对初始的RDD进行transformation级别的处理,例如map、filter等高阶函数的编程,来进行具体的数据计算

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

     */

    val words = lines.flatMap { line => line.split(" ") }//对每一行的字符串进行单次拆分并把所有行的拆分结果通过flat合并成一个大的单次集合

    /**

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

     */

    val pairs = words.map { word => (word,1) }//其实编程了Tuple  (word,1)

    /**

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

     */

    val wordCounts = pairs.reduceByKey(_+_)//对相同的key,进行value的累加(包括local和Reducer级别同时reduce)

    //集群中wordCounts.collect

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

    /**

     * 5、释放相关资源

     */

    sc.stop()

  }

}

打包:jar

WordCount.jar

http://spark.apache.org/docs/latest/submitting-applications.html

./bin/spark-submit \
  --class <main-class>
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments]

./spark-submit --class com.dt.spark.WordCount_Cluster --master spark://Master:7077 /root/Documents/SparkApps/WordCount.jar

hadoop fs -put /usr/local/spark-1.6.0-bin-hadoop2.6/README.md hdfs://192.168.145.131:9000/historyserverforSpark/  可以上传文件 但是不能用??

作业:在eclipse中写好广告点击排名的程序并测试好


王家林老师名片:

中国Spark第一人

新浪微博:http://weibo.com/ilovepains

微信公众号:DT_Spark

博客:http://blog.sina.com.cn/ilovepains

手机:18610086859

QQ:1740415547

邮箱:[email protected]


本文出自 “一枝花傲寒” 博客,谢绝转载!

以上是关于Eclipse下开发Scala(DT大数据梦工厂)的主要内容,如果未能解决你的问题,请参考以下文章

底层战详解使用Java开发Spark程序(DT大数据梦工厂)

大数据学习:Scala隐式转换和并发编程(DT大数据梦工厂)

Spark高级排序彻底解密(DT大数据梦工厂)

HA下Spark集群工作原理(DT大数据梦工厂)

SparkRDD解密(DT大数据梦工厂)

Master HA彻底解密(DT大数据梦工厂)