1 package com.zhouls.spark
 2 
 3 /**
 4   * Created by Administrator on 2016/9/18.
 5   */
 6 
 7 package com.zhouls.spark
 8 
 9 import org.apache.spark.SparkConf
10 import org.apache.spark.SparkContext
11 
12 
13 object WordCount {
14   def main(args: Array[String]): Unit = {
15     /*
16      * 第1步：创建spark的配置对象sparkconf，设置spark程序的运行时的配置信息，
17      * 例如说通过setMaster来设置程序要链接的spark集群的master的URL，如果设置为local，则代表spark程序在本地运行，
18      * 特别适合于机器配置条件非常差（例如只有1G内存）的初学者
19      */
20     val conf = new SparkConf()//创建SparkConf对象
21     conf.setAppName("Wow,My First Spark App!")//设置应用程序的名称，在程序运行的监控界面可以看到名称
22     conf.setMaster("local")//此时，程序在本地运行，不需要安装spark集群
23 
24     /*
25      * 第2步：创建SparkContext对象
26      * SparkContext是spark程序所有功能的唯一入口，无论是采用scala、java、pthon、R等都必须有一个SparkContext
27      * SparkContext核心作用：初始化spark应用程序运行所需要的核心组件、包括DAGScheduler、TaskScheduler、SchedulerBackend
28      * 同时还会负责spark程序往master注册程序等
29      * SparkContext是整个spark应用程序中最为至关重要的一个对象
30      */
31     val sc = new SparkContext(conf)
32 
33     /*
34      * 第3步：根据具体的数据源（HDFS、HBase、Local FS、DB、S3等）通过SparkContext来创建RDD
35      * RDD的创建基本有三种方式：根据外部的数据来源（如HDFS）、根据Scala集合、由其它的RDD操作数据会被RDD划分成
36      * 一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
37      */
38     // val lines:RDD[String] = sc.textFile("D://SoftWare//spark-1.6.2-bin-hadoop2.6//README.md", 1)//path指的是文件路径，minPartitions指的是最小并行度
39     val lines = sc.textFile("D://SoftWare//spark-1.6.2-bin-hadoop2.6//README.md", 1)//path指的是文件路径，minPartitions指的是最小并行度
40 
41     /*
42      * 第4步：对初始的RDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
43      *  第4.1步：将每一行的字符串拆分成单个的单词
44      */
45     val words = lines.flatMap{ line => line.split(" ")}//对每一行的字符串进行单词拆分并把所有行的拆分结果通过flat合并成为一个
46 
47     /*
48     * 4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是 word => (word,1)
49     */
50     val pairs = words.map { word => (word,1) }
51 
52     /*
53      * 4.3步:在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
54      */
55     val wordCounts =  pairs.reduceByKey(_+_)//对相同的key，进行value的累计（包括local和reducer级别同时reduce）
56     wordCounts.foreach(wordNumberPair => println(wordNumberPair._1 + ":" + wordNumberPair._2))
57 
58     /*
59      * 第5步
60      */
61     sc.stop();
62   }
63 
64 }

 1 package com.zhouls.spark
 2 
 3 /**
 4   * Created by Administrator on 2016/9/18.
 5   */
 6 
 7 import org.apache.spark.SparkConf
 8 import org.apache.spark.SparkContext
 9 
10 
11 object WordCount_Cluster {
12   def main(args: Array[String]): Unit  = {
13     /*
14      * 第1步：创建spark的配置对象sparkconf，设置spark程序的运行时的配置信息，
15      * 例如说通过setMaster来设置程序要链接的spark集群的master的URL，如果设置为local，则代表spark程序在本地运行，
16      * 特别适合于机器配置条件非常差（例如只有1G内存）的初学者
17      */
18     val conf = new SparkConf()//创建SparkConf对象
19     conf.setAppName("Wow,My First Spark App!")//设置应用程序的名称，在程序运行的监控界面可以看到名称
20     conf.setMaster("spark://SparkSingleNode:7077")//此时，程序在本地运行，不需要安装spark集群
21 
22     /*
23      * 第2步：创建SparkContext对象
24      * SparkContext是spark程序所有功能的唯一入口，无是采用scala、java、pthon、R等都必须有一个SparkContext
25      * SparkContext核心作用：初始化spark应用程序运行所需要的核心组件、包括DAGScheduler、TaskScheduler、SchedulerBackend
26      * 同时还会负责spark程序往master注册程序等
27      * SparkContext是整个spark应用程序中最为至关重要的一个对象
28      */
29     val sc = new SparkContext(conf)
30 
31     /*
32      * 第3步：根据具体的数据源（HDFS、HBase、Local FS、DB、S3等）通过SparkContext来创建RDD
33      * RDD的创建基本有三种方式：根据外部的数据来源（如HDFS）、根据Scala集合、由其它的RDD操作数据会被RDD划分成
34      * 一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
35      */
36     // val lines:RDD[String] = sc.textFile("D://SoftWare//spark-1.6.2-bin-hadoop2.6//README.md", 1)//path指的是文件路径，minPartitions指的是最小并行度
37     //  val lines = sc.textFile("D://SoftWare//spark-1.6.2-bin-hadoop2.6//README.md", 1)//path指的是文件路径，minPartitions指的是最小并行度
38     val lines = sc.textFile("hdfs://SparkSingleNode:9000/README.md", 1)//没必要会感知上下文
39     //        val lines = sc.textFile("/README.md", 1)//path指的是文件路径，minPartitions指的是最小并行度
40     //            val lines = sc.textFile("/README.md")//为什么，这里不写并行度了呢？因为,hdfs会有一个默认的
41     /*
42      * 第4步：对初始的RDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
43      *  第4.1步：将每一行的字符串拆分成单个的单词
44      */
45     val words = lines.flatMap{ line => line.split(" ")}//对每一行的字符串进行单词拆分并把所有行的拆分结果通过flat合并成为一个
46 
47     /*
48     * 4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是 word => (word,1)
49     */
50     val pairs = words.map { word => (word,1) }
51 
52     /*
53      * 4.3步:在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
54      */
55     val wordCounts =  pairs.reduceByKey(_+_)//对相同的key，进行value的累计（包括local和reducer级别同时reduce）
56     wordCounts.collect.foreach(wordNumberPair => println(wordNumberPair._1 + ":" + wordNumberPair._2))
57 
58     /*
59      * 第5步
60      */
61     sc.stop();
62   }
63 
64 }