四、RDD编程API
Spark支持两个类型（算子）操作：Transformation和Action
4.1 Transformation
4.2 Action
4.3 Spark WordCount代码编写
使用maven进行项目构建
（1）使用scala进行编写
查看官方网站，需要导入2个依赖包
详细代码
SparkWordCountWithScala.scala
import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object SparkWordCountWithScala { def main(args: Array[String]): Unit = { val conf = new SparkConf() /** * 如果这个参数不设置，默认认为你运行的是集群模式 * 如果设置成local代表运行的是local模式 */ conf.setMaster("local") //设置任务名 conf.setAppName("WordCount") //创建SparkCore的程序入口 val sc = new SparkContext(conf) //读取文件 生成RDD val file: RDD[String] = sc.textFile("E:\\hello.txt") //把每一行数据按照，分割 val word: RDD[String] = file.flatMap(_.split(",")) //让每一个单词都出现一次 val wordOne: RDD[(String, Int)] = word.map((_,1)) //单词计数 val wordCount: RDD[(String, Int)] = wordOne.reduceByKey(_+_) //按照单词出现的次数 降序排序 val sortRdd: RDD[(String, Int)] = wordCount.sortBy(tuple => tuple._2,false) //将最终的结果进行保存 sortRdd.saveAsTextFile("E:\\result") sc.stop() }
运行结果
4.4 WordCount执行过程图
五、RDD的宽依赖和窄依赖
5.1 RDD依赖关系的本质内幕
由于RDD是粗粒度的操作数据集，每个Transformation操作都会生成一个新的RDD，所以RDD之间就会形成类似流水线的前后依赖关系；RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。如图所示显示了RDD之间的依赖关系。
从图中可知：
窄依赖：是指每个父RDD的一个Partition最多被子RDD的一个Partition所使用，例如map、filter、union等操作都会产生窄依赖；（独生子女）
宽依赖：是指一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖；（超生）
需要特别说明的是对join操作有两种情况：
（1）图中左半部分join：如果两个RDD在进行join操作时，一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join，那么这种类型的join操作就是窄依赖，例如图1中左半部分的join操作(join with inputs co-partitioned)；
（2）图中右半部分join：其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition进行join的转换，这就涉及到了shuffle，因此这种类型的join操作也是宽依赖。
总结：
在这里我们是从父RDD的partition被使用的个数来定义窄依赖和宽依赖，因此可以用一句话概括下：如果父RDD的一个Partition被子RDD的一个Partition所使用就是窄依赖，否则的话就是宽依赖。因为是确定的partition数量的依赖关系，所以RDD之间的依赖关系就是窄依赖；由此我们可以得出一个推论：即窄依赖不仅包含一对一的窄依赖，还包含一对固定个数的窄依赖。
一对固定个数的窄依赖的理解：即子RDD的partition对父RDD依赖的Partition的数量不会随着RDD数据规模的改变而改变；换句话说，无论是有100T的数据量还是1P的数据量，在窄依赖中，子RDD所依赖的父RDD的partition的个数是确定的，而宽依赖是shuffle级别的，数据量越大，那么子RDD所依赖的父RDD的个数就越多，从而子RDD所依赖的父RDD的partition的个数也会变得越来越多。
5.2 依赖关系下的数据流视图
在spark中，会根据RDD之间的依赖关系将DAG图（有向无环图）划分为不同的阶段，对于窄依赖，由于partition依赖关系的确定性，partition的转换处理就可以在同一个线程里完成，窄依赖就被spark划分到同一个stage中，而对于宽依赖，只能等父RDD shuffle处理完成后，下一个stage才能开始接下来的计算。
因此spark划分stage的整体思路是：从后往前推，遇到宽依赖就断开，划分为一个stage；遇到窄依赖就将这个RDD加入该stage中。因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。
在spark中，Task的类型分为2种：ShuffleMapTask和ResultTask：
简单来说，DAG的最后一个阶段会为每个结果的partition生成一个ResultTask，即每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition的数量所决定的！而其余所有阶段都会生成ShuffleMapTask；之所以称之为ShuffleMapTask是因为它需要将自己的计算结果通过shuffle到下一个stage中；也就是说上图中的stage1和stage2相当于mapreduce中的Mapper,而ResultTask所代表的stage3就相当于mapreduce中的reducer。
在之前动手操作了一个wordcount程序，因此可知，Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不过区别在于：Hadoop中的MapReduce天生就是排序的；而reduceByKey只是根据Key进行reduce，但spark除了这两个算子还有其他的算子；因此从这个意义上来说，Spark比Hadoop的计算算子更为丰富。
