Skip to content

Latest commit

 

History

History
89 lines (73 loc) · 3.8 KB

0008.md

File metadata and controls

89 lines (73 loc) · 3.8 KB

第8课:彻底实战详解使用IDE开发Spark程序


标签: sparkIMF


###Eclipse for Scala IDE使用

1.修改依赖的scala版本为Scala 2.10.x 2.加入Spark 1.6.0的jar文件依赖 3.找到依赖的Spark jar文件并导入到eclipse中的jar依赖中 4.在src下建立Spark工程包 5.创建Scala入口类 6.把class变成object,并编写main入口方法

####本地运行模式源码

package com.dt.spark.hello

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

/** 
 * 使用Scala开发本地测试的Spark WordCount程序
 */
object WordCount {
  def main(args: Array[String]): Unit = {
     /**
     * 第1步:创建Spark的配置对象SparkConf,设置Spark程序的运行的配置信息,
     * 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,
     * 如果设置为local,则代表Spark程序在本地运行,特别适合于机器配置条件差
     * (例如只有1G的内存)的初学者
     */
     val conf = new SparkConf()
     //设置应用程序的名称,在程序运行的监控界面可以看到名称
     conf.setAppName("My First Spark App")
     conf.setMaster("local") //此时,程序在本地运行,不需要安装spark集群
     
     /**
      * 第2步:创建SparkContext对象
      * SparkContext是Spark程序所有功能的唯一入口,无论是采用Scala、Java、Python、R等都必须有一个SparkContext的实例
      * SparkContext的核心作用:
      * 	初始化Spark应用程序运行所需要的核心组件,包括TAGScheduler、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处理范畴
      */
     val lines = sc.textFile("G:\\runtime\\spark-1.6.0\\README.md",1)  //读取本地文件,并设置为1个Partition
     
     /**
      * 第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) }
     //第4.3步:在每个单词实例计数为1的基础上,统计每个单词在文件中出现的总次数
     val wordCount = pairs.reduceByKey((v1,v2)=>v1+v2) //对相同的Key进行Value的累加(包括Local和Reducer级别同时Reduce)
     
      wordCount.collect().foreach(wordNumberPair => println(wordNumberPair._1+" : "+wordNumberPair._2) )
     
     sc.stop()
  }
}

####集群运行模式

源代码和本地模式一样,但是不指定setMaster()方法即可! 将本地文件改为HDFS上的文件

val lines = sc.textFile("G:\\runtime\\spark-1.6.0\\README.md",1)  //读取HDFS,并设置为1个Partition

使用spark/bin/spark-submit命令执行:

/usr/local/spark-1.6.0/bin/spark-submit --master spark://master:7077 --class com.dt.spark.hello.WordCountCluster wordCount.jar