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

本地安装好java和scala并创建工程WordCount

第一步：修改依赖的Scala版本为Scala 2.10.x(改为本地对应版本)

第二步： 加入spark 1.6 jar文件依赖

第三步： 找到依赖的spark jar文件 ，并导入到eclipse中的jar依赖

第四步：在src下建立spark工程包

第五步： 创建Scala入口类

第六步： 把class 变成 object 并编写main入口方法

使用Scala开发本地测试的Spark WordCount程序

package spark
//使用Scala开发本地测试的Spark WordCount程序
import org.apache.spark.SparkConf

object workcount {
def main(args:Array[String]){
// 第一步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息。
//例如说通过setMaster来设置程序要链接的Spark集群的Master的URL，如果设置为local，则代替Spark程序在本地运行，特别适合于机器配置条件非常差（只有1G的内存）的初学者。

val conf = new SparkConf();//创建SparkConf对象
conf.setAppName("First Spark App!!!") //设置应用程序的名称，在程序运行的监控界面可以看到名称
//conf.setMaster("local") //此时，程序在本地运行，不需要安装Spark集群
//第二步：创建SparkContext对象SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala、Java、Python、R等都必须有一个SparkContext。
//SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler、TaskScheduler、Schedulerbackend同时还会负责Spark程序往Master注册程序等。
//SparkContext是整个Spark应用程序中最为至关重要的一个对象。
val sc = new SparkContext(conf) //创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息

//第三步: 根据具体的数据来源（HDFS、HBase、Local FS 、DB、S3等）通过SparkContext来创建RDD。
//RDD的创建基本有三种方式：根据外部的数据来源（例如HDFS）、根据Scala集合、由其他的RDD操作。
//数据会被RDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴。

//val lines:RDD[String] = sc.textFile("hdfs://Master:9000/README")//读取hdfs并切分成不同的partition
val lines = sc.textFile("F://spark-2.0.0-bin-

hadoop2.7//README",1) //读取本地文件并设置为一个partition。第一个参数为本地文件路径，第二个参数为并行度

//第四步: 对初始的RDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
//4.1 将每一行的字符串拆分成单个的单词,并把所有行的拆分结果通过flat合并成一个大的单词集合
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 wordCounts = pairs.reduceByKey(_+_)//对相同的Key，进行Value的累加(包括Local和Reducer级别同时Reduce)
wordCounts.collect.foreach(wordNumberPair => println(wordNumberPair._1 + " : " + wordNumberPair._2))
//1为key，2为value。打印结果
sc.stop()
}
}

在集群上运行需要修改文件路径

export => java / JAR file =>选择输出目录

传入ubuntu系统中

./spark-submit  --class WordCount  --master spark://Master:7077 /root/../WordCount.jar

或创建脚本 wordcount.sh 直接执行./wordcount.sh 即可

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

package spark
import org.apache.spark.SparkConf

object workcount {
def main(args:Array[String]){
val conf = new SparkConf();
//创建SparkConf对象
conf.setAppName("First Spark App!!!")
//设置应用程序的名称
val sc = new SparkContext(conf)
//创建SparkContext对象，
val lines:RDD[String] = sc.textFile("hdfs://Master:9000/README")
//读取hdfs并切分成不同的partition
val words = lines.flatMap{ line => line.split(" ")}
//单词拆分并合并成单词集合
val pairs = words.map { word => (word,1)}
val wordCounts = pairs.reduceByKey(_+_)
//对相同的Key，进行Value的累加

//排序
val wordCounts1 = wordCounts.map(pairs => (pairs._2,pairs._1))
//交换Key和Value的位置
val wordCounts2 = wordCounts1.sortByKey(false,1)
//按Valve的值递减排序
val wordCounts3 = wordCounts2.map(pairs => (pairs._1,pairs._2))
//还原位置

wordCounts3.collect.foreach(wordNumberPair => println(wordNumberPair._1 + " : " + wordNumberPair._2))
//1为key，2为value
sc.stop()
}
}