二、spark--spark core原理与使用

[TOC]

一、spark中一些基本术语

RDD：弹性分布式数据集，是spark的核心重点
算子：操作RDD的一些函数
application：用户的写的spark程序（DriverProgram + ExecutorProgram）
job：一个action类算子触发的操作
stage：一组任务，会根据依赖关系将job划分成若干个stage
task：同一个stage内部有多个同样操作的task（但处理的数据不同），是集群中最小的执行单元

可能说完这些概念，其实还是不太懂，没关系，这只是先有点印象

二、RDD基本原理和使用

2.1 什么是RDD

​ RDD，全称：Resilient Distributed Dataset，也就是弹性分布式数据集。是spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。可能这还不清晰，我举个例子：
假设我使用sc.textFile(xxxx)从hdfs中读取一个文件的数据，那么文件的数据就相当于一个RDD，但是事实上这个文件的数据在处理时是处于多个不同的worker节点上的，但是在逻辑上，在这个spark集群，这些数据都是属于一个RDD中的。这也就是为什么说RDD是个逻辑概念，它是整个集群的一个抽象对象，分布在集群中。由此看出，RDD是spark实现数据分布式计算处理的关键。例如：

​ 图 2.1 RDD原理

2.2 RDD的属性

关于RDD的属性，源码中有一段注释，如下：

* Internally, each RDD is characterized by five main properties:*  - A list of partitions1、是一组分区理解：RDD是由分区组成的，每个分区运行在不同的Worker上，通过这种方式，实现分布式计算，即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。*  - A function for computing each split2、split理解为分区在RDD中，有一系列函数，用于处理计算每个分区中的数据。这里把函数叫做算子。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。算子类型：transformation   action*  - A list of dependencies on other RDDs3、RDD之间存在依赖关系。窄依赖，宽依赖。需要用依赖关系来划分Stage，任务是按照Stage来执行的。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。*  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)4、可以自动以分区规则来创建RDD创建RDD时，可以指定分区，也可以自定义分区规则。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。*  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for*    an HDFS file)5、优先选择离文件位置近的节点来执行任务。移动计算，不移动数据这点需要解释下：一般来说spark是构建在hdfs之上，从hdfs中读取数据进行处理的。而hdfs是一个分布式存储，比如有A、B、C三个datanode，假设spark要处理的数据刚好存储在C节点上。如果spark此时将任务放在B节点或者A节点上执行，那么就得先从C节点读取数据，然后经过网络传输到A或B节点，然后才能处理，这其实是很耗费性能。而这里spark的意思是优先在离处理数据比较近的节点上执行任务，也就是优先在C节点上执行任务。这就可以节省数据传输所耗费的时间和性能。也就是移动计算而不移动数据。

2.3 创建RDD

创建RDD首先需要创建 SparkContext对象：//创建spark配置文件对象.设置app名称，master地址,local表示为本地模式。//如果是提交到集群中，通常不指定。因为可能在多个集群汇上跑，写死不方便val conf = new SparkConf().setAppName("wordCount").setMaster("local")//创建spark context对象val sc = new SparkContext(conf)

通过sc.parallelize() 创建RDD：

sc.parallelize(seq,numPartitions)seq：为序列对象，如list，array等numPartitions：分区个数，可以不指定，默认是2例子：val rdd1 = sc.parallelize(Array(1,2,3,4,5),3)rdd1.partitions.length

通过外部数据源创建

val rdd1 = sc.textFile("/usr/local/tmp_files/test_WordCount.txt")

2.4 算子类型

算子有分为transformation和action类型，
transformation：

延迟计算，lazy懒值，不会触发计算。只有遇到action算子才会触发计算。它们只是记住这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给Driver的动作时，这些转换才会真正运行。这种设计让Spark更加有效率地运行

action：

和transformation类似，但是是直接触发计算的，不会等待

2.5 transformation算子

这里为了方便讲解，实现创建一个rdd，都使用spark-shell中进行演示：

scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,8,34,100,79))rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24

2.5.1 map(func)

map[U](f:T=>U)()参数是一个函数，并且要求函数参数是单个，返回值也是单个。用函数处理传入的数据，然后返回处理后的数据例子：//这里传入一个匿名函数，将rdd1中的每个值*2，并返回处理的新数组scala> val rdd2 = rdd1.map(_*2)rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at :26//这里collect是一个action算子，触发计算并打印结果scala> rdd2.collectres0: Array[Int] = Array(2, 4, 6, 8, 10, 16, 68, 200, 158)

2.5.2 filter

filter(f:T=>boolean)参数是一个判断函数，判断传入的参数，然后返回true还是false，常用来过滤数据。最后将true的数据返回例子：//过滤出大于20的数据scala> rdd2.filter(_>20).collectres4: Array[Int] = Array(68, 200, 158)

2.5.3 flatMap

flatMap(f:T=>U)先map后flat，flat就是将多个列表等对象展开合并成一个大列表。并返回处理后的数据。这个函数一般用来处理多个一个列表中还包含多个列表的情况例子：scala> val rdd4 = sc.parallelize(Array("a b c","d e f","x y z"))rdd4: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[18] at parallelize at :24//处理逻辑是：将array中每个字符串按空格切割，然后生成多个array，接着将多个array展开合并一个新的arrayscala> val rdd5 = rdd4.flatMap(_.split(" "))rdd5: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[19] at flatMap at :26scala> rdd5.collectres5: Array[String] = Array(a, b, c, d, e, f, x, y, z)

2.5.4 集合操作

union(otherDataset) 并集intersection(otherDataset) 交集distinct([numTasks]))去重例子：scala> val rdd6 = sc.parallelize(List(5,6,7,8,9,10))rdd6: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at :24scala> val rdd7 = sc.parallelize(List(1,2,3,4,5,6))rdd7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[21] at parallelize at :24//并集scala> val rdd8 = rdd6.union(rdd7)rdd8: org.apache.spark.rdd.RDD[Int] = UnionRDD[22] at union at :28scala> rdd8.collectres6: Array[Int] = Array(5, 6, 7, 8, 9, 10, 1, 2, 3, 4, 5, 6)//去重scala> rdd8.distinct.collectres7: Array[Int] = Array(4, 8, 1, 9, 5, 6, 10, 2, 7, 3)                         //交集scala> val rdd9 = rdd6.intersection(rdd7)rdd9: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[31] at intersection at :28scala> rdd9.collectres8: Array[Int] = Array(6, 5)

2.5.5 分组操作

groupByKey([numTasks])：只是将同一key的进行分组聚合reduceByKey(f:（V,V）=>V, [numTasks]) 首先是将同一key的KV进行聚合，然后对value进行操作。scala> val rdd1 = sc.parallelize(List(("Tom",1000),("Jerry",3000),("Mary",2000)))rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[32] at parallelize at :24scala> val rdd2 = sc.parallelize(List(("Jerry",1000),("Tom",3000),("Mike",2000)))rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[33] at parallelize at :24scala> val rdd3 = rdd1 union rdd2rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[34] at union at :28scala> rdd3.collectres9: Array[(String, Int)] = Array((Tom,1000), (Jerry,3000), (Mary,2000), (Jerry,1000), (Tom,3000), (Mike,2000))scala> val rdd4 = rdd3.groupByKeyrdd4: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[35] at groupByKey at :30//分组scala> rdd4.collectres10: Array[(String, Iterable[Int])] = Array((Tom,CompactBuffer(1000, 3000)), (Jerry,CompactBuffer(3000, 1000)), (Mike,CompactBuffer(2000)), (Mary,CompactBuffer(2000)))注意：使用分组函数时，不推荐使用groupByKey，因为性能不好，官方推荐reduceByKey//分组并聚合scala> rdd3.reduceByKey(_+_).collectres11: Array[(String, Int)] = Array((Tom,4000), (Jerry,4000), (Mike,2000), (Mary,2000))

2.5.6 cogroup

这个函数的功能不太好总结，直接看例子吧scala> val rdd1 = sc.parallelize(List(("Tom",1),("Tom",2),("jerry",1),("Mike",2)))rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[37] at parallelize at :24scala> val rdd2 = sc.parallelize(List(("jerry",2),("Tom",1),("Bob",2)))rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[38] at parallelize at :24scala> val rdd3 = rdd1.cogroup(rdd2)rdd3: org.apache.spark.rdd.RDD[(String, (Iterable[Int], Iterable[Int]))] = MapPartitionsRDD[40] at cogroup at :28scala> rdd3.collectres12: Array[(String, (Iterable[Int], Iterable[Int]))] = Array((Tom,(CompactBuffer(1, 2),CompactBuffer(1))), (Mike,(CompactBuffer(2),CompactBuffer())), (jerry,(CompactBuffer(1),CompactBuffer(2))), (Bob,(CompactBuffer(),CompactBuffer(2))))

2.5.7 排序

sortByKey(acsending:true/false) 根据KV中的key排序sortBy(f:T=>U,acsending:true/false) 一般排序，且是对处理后的数据进行排序，可以用来给KV的，按照value进行排序例子：scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,8,34,100,79))rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24scala> val rdd2 = rdd1.map(_*2)rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at :26scala> rdd2.collectres0: Array[Int] = Array(2, 4, 6, 8, 10, 16, 68, 200, 158)scala> rdd2.sortBy(x=>x,true)res1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at sortBy at :29scala> rdd2.sortBy(x=>x,true).collectres2: Array[Int] = Array(2, 4, 6, 8, 10, 16, 68, 158, 200)                      scala> rdd2.sortBy(x=>x,false).collectres3: Array[Int] = Array(200, 158, 68, 16, 10, 8, 6, 4, 2)

另外一个例子：

需求：我们想按照KV中的value进行排序，但是SortByKey按照key排序的。做法一：1、第一步交换，把key value交换，然后调用sortByKey2、KV再次调换位置scala> val rdd1 = sc.parallelize(List(("tom",1),("jerry",1),("kitty",2),("bob",1)))rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[42] at parallelize at :24scala> val rdd2 = sc.parallelize(List(("jerry",2),("tom",3),("kitty",5),("bob",2)))rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[43] at parallelize at :24scala> val rdd3 = rdd1 union(rdd2)rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[44] at union at :28scala> val rdd4 = rdd3.reduceByKey(_+_)rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[45] at reduceByKey at :30scala> rdd4.collectres13: Array[(String, Int)] = Array((bob,3), (tom,4), (jerry,3), (kitty,7))//调换位置再排序，然后再调回来scala> val rdd5 = rdd4.map(t => (t._2,t._1)).sortByKey(false).map(t=>(t._2,t._1))rdd5: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[50] at map at :32scala> rdd5.collectres14: Array[(String, Int)] = Array((kitty,7), (tom,4), (bob,3), (jerry,3)) 做法二：直接使用sortBy 这个算子，可以直接按照value排序

2.6 action算子

reduce

类似前面的reduceByKey，但是用于非KV的数据合并，且是action算子scala> val rdd1 = sc.parallelize(List(1,2,3,4,5))rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[41] at parallelize at :24scala> val rdd2 = rdd1.reduce(_+_)rdd2: Int = 15

还有一些action算子：

reduce(func)    通过func函数聚集RDD中的所有元素，这个功能必须是可交换且可并联的collect()   在驱动程序中，以数组的形式返回数据集的所有元素。通常只是用于触发计算count()  返回RDD的元素个数first()   返回RDD的第一个元素（类似于take(1)）take(n)   返回一个由数据集的前n个元素组成的数组takeSample(withReplacement,num, [seed]) 返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子takeOrdered(n, [ordering])  ，返回一个由数据集的前n个元素组成的数组，并排序saveAsTextFile(path)    将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本saveAsSequenceFile(path)    将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。saveAsObjectFile(path)  countByKey()    针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。foreach(func)   在数据集的每一个元素上，运行函数func进行更新。

2.7 RDD的缓存特性

RDD也存在缓存机制，也就是将RDD缓存到内存或者磁盘中，不用重复计算。
这里涉及到几个算子：

cache()   标识该rdd可以被缓存，默认缓存在内存中，底层其实是调用persist() persist() 标识该rdd可以被缓存，默认缓存在内存中persist(newLevel : org.apache.spark.storage.StorageLevel) 和上面类似，但是可以指定缓存的位置可以缓存的位置有：val NONE = new StorageLevel(false, false, false, false)val DISK_ONLY = new StorageLevel(true, false, false, false)val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)val MEMORY_ONLY = new StorageLevel(false, true, false, true)val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)val OFF_HEAP = new StorageLevel(true, true, true, false, 1)基本就分为两类：纯内存缓存纯磁盘缓存磁盘+内存缓存

一般来说，直接是默认的位置，也就是缓存在内存中性能较好，但是会耗费很多内存，这点要注意，如无需要，就不要缓存了。
举例：

读取一个大文件，统计行数scala> val rdd1 = sc.textFile("hdfs://192.168.109.132:8020/tmp_files/test_Cache.txt")rdd1: org.apache.spark.rdd.RDD[String] = hdfs://192.168.109.132:8020/tmp_files/test_Cache.txt MapPartitionsRDD[52] at textFile at :24scala> rdd1.countres15: Long = 923452 触发计算，统计行数scala> rdd1.cacheres16: rdd1.type = hdfs://192.168.109.132:8020/tmp_files/test_Cache.txt MapPartitionsRDD[52] at textFile at :24标识这个RDD可以被缓存，不会触发计算scala> rdd1.countres17: Long = 923452 触发计算，并把结果进行缓存scala> rdd1.countres18: Long = 923452直接从缓存中读取数据。

要注意一个点：调用cache方法的时候，只是说标识该rdd在后续触发计算的时候，结果可以被缓存，而不是说当前rdd就被缓存了，这点要分清楚

2.8 RDD的容错机制--checkpoint

​ spark在计算时，中间涉及到RDD的多个转换过程，如果这时候RDD的某个分区计算故障，导致结果丢失了。最简单的办法自然是从头开始重新计算，但是这样很浪费时间。而checkpoint就是在触发计算的时候，对RDD进行检查点状态保存，如果后面的计算出错了，还可以从检查点开始重新计算。
​ checkpoint一般是在具有容错能力，高可靠的文件系统上(比如HDFS, S3等)设置一个检查点路径，用于保存检查点数据。出错的时候直接从检查点目录读取。有本地目录和远程目录两种模式。

2.8.1 本地目录

这种模式下，要求运行在本地模式下，不能用在集群模式下，一般用于测试开发

sc.setCheckpointDir(本地路径)   设置本地checkpoint路径rdd1.checkpoint    设置检查点rdd1.count         遇到action类算子，触发计算，就会在checkpoint目录生成检查点

2.8.2 远程目录（hdfs为例）

这种模式要求运行在集群模式下，用于生产环境

scala> sc.setCheckpointDir("hdfs://192.168.109.132:8020/sparkckpt0619")scala> rdd1.checkpointscala> rdd1.countres22: Long = 923452用法都是类似，只是目录不一样

要注意，使用checkpoint的时候，源码中有一段话如下：

this function must be called before any job has been executed on this RDD. It is strongly recommend that  this rdd ispersisted in memory，otherwise saving it on a file will require recomputation.大致意思就是：要在开始计算前就调用这个方法，也就是action算子之前。而且最好设置这个rdd缓存在内存中，否则保存检查点的时候，需要重新计算一次。

2.9 RDD中的依赖与stage原理

2.9.1 RDD依赖

​ 这个是RDD运行原理中的一个重点概念。
​ 首先要说说依赖，依赖的意思就是RDD之间是有依赖关系的，因为spark计算过程中涉及到多个RDD的转换。RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。看图

​ 图2.2 RDD的宽窄依赖

宽依赖：
一个父RDD的partition被多个子RDD的partition依赖。其实就是将父RDD的数据进行shuffle的过程，因为父RDD一个partition被多个RDD的partition依赖，意味着需要将父RDD的数据打乱分配给多个RDD，打乱的过程其实就是shuffle。一般实际情况是多个父RDD的partition和多个子RDD的partition互相交错的依赖。

窄依赖：
一个父RDD的partition最多被一个子RDD的partition依赖

2.9.2 stage划分

​ 图2.3 RDD依赖

​ DAG(Directed Acyclic Graph)叫做有向无环图，原始的RDD通过一系列的转换就就形成了DAG，根据RDD之间的依赖关系的不同将DAG划分成不同的Stage。宽窄依赖的作用就是用划分stage，stage之间是宽依赖，stage内部是窄依赖。
​ 对于窄依赖，由于父和子RDD的partition都是一对一的依赖关系，所以可以父和子的转换可以放在一个task中执行，例如上面的task0，CDF都是窄依赖，所以CDF直接的转换是可以放在一个task中执行的。一个stage内部都是窄依赖
​ 对于宽依赖，由于有shuffle的存在，那么就要求所有父RDD都处理完成后，才能执行执行shuffle，接着子RDD才能进行处理。由于shuffle的存在，导致task任务链必定不是连续的了，需要重新开始规划task任务链，所以宽依赖是划分stage的依据。
​ 再往深的说，为什么要划分stage？
​ 根据宽依赖划分了stage之后，因为宽依赖的shuffle，所以导致task链是无法连续。而划分stage就是让一个stage内部只有窄依赖，窄依赖是一对一的关系，那么task链就是连续的，没有shuffle，就比如上面task0中，C->D->F，中的一个分区，转换过程都是一对一的，所以是一个连续的task链，放在一个task中，而另外一个分区类似，就放在task1中。因为F->G是宽依赖，需要shuffle，所以task链无法连续。像这种一条线把RDD转换逻辑串起来，直到遇到宽依赖，就是一个task，而一个task处理的实际上是一个分区的数据转换过程。而在spark中，task是最小的调度单位，spark会将task分配到离分区数据近的worker节点上执行。所以其实spark调度是task。
​ 那么回到最开始的问题上，为什么要划分stage，因为根据宽窄依赖划分出stage之后，stage内部就可以很方便划分出一个个task，每个task处理一个分区的数据，然后spark就将task调度到对应的worker节点上执行。所以从划分stage到划分task，核心就在于实现并行计算。
​ 所以，最后就是一句话，划分stage的目的是为了更方便的划分task

2.9.3 RDD存储的是什么？

​ 说到这里，其实我们想到一个问题，RDD里面存储的是数据吗？实际上并不是，它存储的实际上对数据的转换链，说的具体点是对分区的转换链，也就是task中包含的算子。而当划分stage，接着划分task之后，一个task内部有什么算子就已经很清楚了，接着就是把计算任务发送到worker节点执行。这种计算我们称为 pipeline计算模式，算子就是在管道中的。
​ 由此，其实RDD叫做弹性分布式数据集，并不是说它存储数据，而是存储了操作数据的方法函数，也就是算子。

2.10 RDD高级算子

2.10.1 mapPartitionsWithIndex

def mapPartitionsWithIndex[U](f: (Int, Iterator[T]) ⇒ Iterator[U])参数说明：f是一个函数参数，需要自定义。f 接收两个参数，第一个参数是Int，代表分区号。第二个Iterator[T]代表该分区中的所有元素。通过这两个参数，可以定义处理分区的函数。Iterator[U] ： 操作完成后，返回的结果。举例：将每个分区中的元素，包括分区号，直接打印出来。//先创建一个rdd，指定分区数为3scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8),3)rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24scala> def fun1(index:Int,iter:Iterator[Int]):Iterator[String]={| iter.toList.map( x => "[PartId: " + index + " , value = " + x + " ]").iterator| }fun1: (index: Int, iter: Iterator[Int])Iterator[String]scala> rdd1.mapPartitionsWithIndex(fun1).collectres0: Array[String] = Array([PartId: 0 , value = 1 ], [PartId: 0 , value = 2 ], [PartId: 1 , value = 3 ], [PartId: 1 , value = 4 ], [PartId: 1 , value = 5 ], [PartId: 2 , value = 6 ], [PartId: 2 , value = 7 ], [PartId: 2 , value = 8 ])

2.10.2 aggregate

聚合操作，类似于分组（group by）.但是aggregate是先局部聚合（类似于mr中的combine），然后再全局聚合。性能比直接使用reduce算子要好，因为reduce是直接全局聚合的。def aggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)参数说明：zeroValue：初始值，这个初始值会加入到每一个分区中，最后也会加入到全局操作中seqOp: (U, T) ⇒ U：局部聚合操作函数combOp: (U, U) ⇒ U：全局聚合操作函数=================================================例子1：初始值是10scala> val rdd2 = sc.parallelize(List(1,2,3,4,5),2)rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at :27//打印看看分区情况scala> rdd2.mapPartitionsWithIndex(fun1).collectres7: Array[String] = Array([PartId: 0 , value = 1 ], [PartId: 0 , value = 2 ], [PartId: 1 , value = 3 ], [PartId: 1 , value = 4 ], [PartId: 1 , value = 5 ])//求出每个分区的最大值，最后得出每个分区最大值，然后全局之后将每个最大值相加scala> rdd2.aggregate(10)(max(_,_),_+_)res8: Int = 30为什么这里是10呢？初始值是10 代表每个分区都多了一个10局部操作，每个分区最大值都是10全局操作，也多一个10 即 10（局部最大） + 10（局部最大） + 10（全局操作默认值） = 30=================================================例子2：使用aggregate将所有分区全局数据求和，有两种方式：1、reduce(_+_)2、aggregate(0)(_+_,_+_)

2.10.3 aggregateByKey

类似于aggregate操作，区别：操作的  的数据,只操作同一key的中的value。是将同一key的KV先局部分组，然后对value聚合。然后再全局分组，再对value聚合。 aggregateByKey和reduceByKey实现的类似的功能，但是效率比reduceByKey高例子：val pairRDD = sc.parallelize(List(("cat",2),("cat",5),("mouse",4),("cat",12),("dog",12),("mouse",2)),2)def fun1(index:Int,iter:Iterator[(String,Int)]):Iterator[String]={iter.toList.map( x => "[PartId: " + index + " , value = " + x + " ]").iterator}pairRDD.mapPartitionsWithIndex(fun1).collectscala> val pairRDD = sc.parallelize(List(("cat",2),("cat",5),("mouse",4),("cat",12),("dog",12),("mouse",2)),2)pairRDD: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[16] at parallelize at :27scala> def fun1(index:Int,iter:Iterator[(String,Int)]):Iterator[String]={| iter.toList.map( x => "[PartId: " + index + " , value = " + x + " ]").iterator| }fun1: (index: Int, iter: Iterator[(String, Int)])Iterator[String]scala> pairRDD.mapPartitionsWithIndex(fun1).collectres31: Array[String] = Array([PartId: 0 , value = (cat,2) ], [PartId: 0 , value = (cat,5) ], [PartId: 0 , value = (mouse,4) ],[PartId: 1 , value = (cat,12) ], [PartId: 1 , value = (dog,12) ], [PartId: 1 , value = (mouse,2)])需求：找到每个分区中，动物最多的动物，进行就和pairRDD.aggregateByKey(0)(math.max(_,_),_+_).collect0:(cat,2)和(cat,5) --> (cat,5)  (mouse,4)1:(cat,12)   (dog,12)   (mouse,2)求和：(cat,17)  (mouse,6)   (dog,12) 

2.10.4 coalesce 和 repartition

这两者都用于重分区 repartition(numPartition)  指定重分区个数，一定会发生shuffle coalesce(numPartition,shuffleOrNot) 指定重分区个数，默认不会发生shuffle，可以指定shuffle

要看更多算子的用法，
写的很详细

2.11 分区

spark自带了两个分区类：
HashPartitioner：这个是默认的partitioner，在一些涉及到shuffle的算子会用到。在一些可以指定最小分区数量的算子中，就会涉及到分区。这些分区只能用于KV对
RangePartitioner：按照key的范围进行分区，比如1~100,101~200分别是不同分区的
用户也可以自己自定义分区，步骤如下：
1、先继承Partitioner类，里面写分区逻辑，形成一个新的分区类
2、rdd.partitionBy(new partiotionerClassxxx())
例子：

数据格式如下：192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/web.jsp HTTP/1.1" 200 239192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/hadoop.jsp HTTP/1.1" 200 242需求：将同一个页面的访问日志各自写到一个单独的文件中 代码：package SparkExerimport org.apache.spark.{Partitioner, SparkConf, SparkContext}import scala.collection.mutable/**  * 自定义分区：  * 1、先继承Partitioner类，里面写分区逻辑，形成一个新的分区类  * 2、rdd.partitionBy(new partiotionerClassxxx())  */object CustomPart {  def main(args: Array[String]): Unit = {    //指定hadoop的家目录，写入文件到本地时需要用到hadoop的一些包    System.setProperty("hadoop.home.dir","F:\\hadoop-2.7.2")    val conf = new SparkConf().setAppName("Tomcat Log Partitioner").setMaster("local")    val sc = new SparkContext(conf)    //切割文件    val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt").map(      line => {        val jspName = line.split(" ")(6)        (jspName,line)      }    )    //提取出所有key，也就是网页名    val rdd2 = rdd1.map(_._1).distinct().collect()    //分区    val rdd3 = rdd1.partitionBy(new TomcatWebPartitioner(rdd2))    //将分区数据写到文件中    rdd3.saveAsTextFile("G:\\test\\tomcat_localhost")  }}class TomcatWebPartitioner(jspList:Array[String]) extends Partitioner{  privateval listMap = new mutable.HashMap[String,Int]()  var partitionNum = 0  //根据网页名称，规划整个分区个数  for (s<-jspList) {    listMap.put(s, partitionNum)    partitionNum += 1  }  //返回分区总个数  override def numPartitions: Int = listMap.size  //按照key返回某个分区号  override def getPartition(key: Any): Int = listMap.getOrElse(key.toString, 0)}

2.12 序列化问题

​ 首先我们知道一点，一个spark程序其实是分为两部分的，一部分driver，它也是在executor中运行的，另一部分则是普通的executor，用于运行操作RDD的task的。所以其实也可以看出，只有是对RDD操作的代码才会进行分布式运行，分配到多个executor中运行，但是不属于RDD的代码是不会的，它仅仅是在driver中执行。这就是关键了。
例子：

object test {    val sc = new SparkContext()    print("xxxx1")    val rdd1 = sc.textFile(xxxx)    rdd1.map(print(xxx2)) }

例如上面的例子，rdd1中的print(xxx2)会在多个executor中执行，因为它是在rdd内部执行，而外面的print("xxxx1")只会在driver中执行，也没有实现序列化，所以实际上也不可能通过网络传递。所以这种区别一定要了解。由此我们可以知道，如果变量什么的不是在rdd内部的话，是不可能被多个executor上的程序获得的。但是如果我们想这样呢？而且是不需要定义在rdd内部。那么就得用到下面的共享变量了

2.13 spark中广播变量(共享变量)

广播变量就是可以实现将driver中的变量给在不同的executor中运行的rdd算子调用，而且无需再rdd算子内部定义。常见的比如连接mysql等数据库的连接对象，可以设置为广播变量，这样就可以只创建一个连接了。
用法例子：

//定义共享变量，用于共享从mongodb读取的数据,需要将数据封装成 map(mid1,[map(mid2,score),map(mid3,score)....])的形式    val moviesRecsMap = spark.read      .option("uri", mongoConfig.uri)      .option("collection", MOVIES_RECS)      .format("com.mongodb.spark.sql")      .load().as[MoviesRecs].rdd.map(item=> {      (item.mid, item.recs.map(itemRecs=>(itemRecs.mid,itemRecs.socre)).toMap)    }).collectAsMap()    这是关键的一步，就是广播变量出去    //将此变量广播,后面就可以在任意一个executor中调用了    val moviesRecsMapBroadcast = spark.sparkContext.broadcast(moviesRecsMap)    //因为是懒值加载，所以需要手动调用一次才会真正广播出去    moviesRecsMapBroadcast.id

三、spark小案例

3.1 统计访问量前N名的网站页面

需求：根据网站访问日志统计出访问量前N位的网页名称数据格式如下：192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/web.jsp HTTP/1.1" 200 239192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/hadoop.jsp HTTP/1.1" 200 242代码：package SparkExerimport org.apache.spark.{SparkConf, SparkContext}/**  * 分析tomcat日志  * 日志例子：  * 192.168.88.1 - - [30/Jul/2017:12:53:43 +0800] "GET /MyDemoWeb/ HTTP/1.1" 200 259  *  * 统计每个页面的访问量  */object TomcatLog {  def main(args: Array[String]): Unit = {    val conf = new SparkConf().setAppName("Tomcat Log analysis").setMaster("local")    val sc = new SparkContext(conf)    val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt")      .map(_.split(" ")(6))      .map((_,1))      .reduceByKey(_+_)      .map(t=>(t._2,t._1))      .sortByKey(false)      .map(t=>(t._2,t._1))      .collect()    //也可以使用 sortBy(_._2,false)直接根据value进行排序    //取出rdd中的前N个数据    rdd1.take(2).foreach(x=>println(x._1 + ":" + x._2))    println("=========================================")    //取出rdd中的后N个数据    rdd1.takeRight(2).foreach(x=>println(x._1 + ":" + x._2))    sc.stop()  }}

3.2 自定义分区例子

请看前面2.11中分区的例子

3.3 spark连接mysql

package SparkExerimport java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.spark.{SparkConf, SparkContext}object SparkConMysql {  def main(args: Array[String]): Unit = {    val conf = new SparkConf().setAppName("Tomcat Log To Mysql").setMaster("local")    val sc = new SparkContext(conf)    val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt")      .map(_.split(" ")(6))    rdd1.foreach(l=>{      //jdbc操作需要包含在rdd中才能被所有worker上的executor调用，也就是借用rdd实现序列化      val jdbcUrl = "jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8"      var conn:Connection = null      //sql语句编辑对象      var ps:PreparedStatement = null      conn = DriverManager.getConnection(jdbcUrl, "root", "wjt86912572")      //？是占位符，后面需要ps1.setxxx(rowkey,value)的形式填充值进去的，按先后顺序      ps = conn.prepareStatement("insert into customer values (?,?)")      ps.setString(1,l)      ps.setInt(2,1)    })  }}注意：spark操作jdbc时，如果直接使用jdbc操作数据库，会有序列化的问题。因为在spark分布式框架中，所有操作RDD的对象应该是属于RDD内部的，才有可能在整个分布式集群中使用。也就是需要序列化。通俗来说：5个worker共享一个jdbc连接对象，和5个worker每个单独创建一个连接对象的区别所以在定义jdbc连接对象时，需要在RDD内部定义

上面的方式过于繁琐，而且每个数据都会新建一个jdbc连接对象
优化：使用rdd1.foreachPartition()来对每个分区操作，而不是对每条数据操作
这样可以通过只为每个分区创建一个jdbc连接对象来节省数据库资源

package SparkExerimport java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.spark.{SparkConf, SparkContext}object SparkConMysql {  def main(args: Array[String]): Unit = {    val conf = new SparkConf().setAppName("Tomcat Log To Mysql").setMaster("local")    val sc = new SparkContext(conf)    val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt")      .map(_.split(" ")(6))    rdd1.foreachPartition(updateMysql)    /**      * 上面的方式过于繁琐，而且每个数据都会新建一个jdbc连接对象      * 优化：使用rdd1.foreachPartition()来对每个分区操作，而不是对每条数据操作      * 这样可以通过只为每个分区创建一个jdbc连接对象来节省数据库资源      */  }  def updateMysql(it:Iterator[String]) = {    val jdbcUrl = "jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8"    var conn:Connection = null    //sql语句编辑对象    var ps:PreparedStatement = null    conn = DriverManager.getConnection(jdbcUrl, "root", "wjt86912572")    //conn.createStatement()    //ps = conn.prepareStatement("select * from customer")    //？是占位符，后面需要ps1.setxxx(rowkey,value)的形式填充值进去的，按先后顺序    ps = conn.prepareStatement("insert into customer values (?,?)")    it.foreach(data=>{      ps.setString(1,data)      ps.setInt(2,1)      ps.executeUpdate()    })    ps.close()    conn.close()  }}

另外一种连接mysql的方式就是通过JdbcRDD对象去连接

package SparkExerimport java.sql.DriverManagerimport org.apacheimport org.apache.sparkimport org.apache.spark.rdd.JdbcRDDimport org.apache.spark.{SparkConf, SparkContext}object MysqlJDBCRdd {  def main(args: Array[String]): Unit = {    val conn = () => {      Class.forName("com.mysql.jdbc.Driver").newInstance()      DriverManager.getConnection("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8",      "root",      "wjt86912572")    }    val conf = new SparkConf().setAppName("Tomcat Log To Mysql").setMaster("local")    val sc = new SparkContext(conf)    //创建jdbcrdd对象    val mysqlRdd = new JdbcRDD(sc,conn,"select * from customer where id>? and id {      r.getString(2)    })  }}这个对象的使用局限性很大，只能用于select，而且必须传入where中的两个限制值，还要指定分区

四、 shuffle问题

4.1 shuffle导致的数据倾斜分析

https://www.cnblogs.com/diaozhaojian/p/9635829.html

1、数据倾斜原理（1）在进行shuffle的时候，必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理，此时如果某个key对应的数据量特别大的话，就会发生数据倾斜。（2）由于shuffle之后的分区规则，导致某个分区数据量过多，导致数据倾斜  2、数据倾斜问题发现与定位　　 通过Spark Web UI来查看当前运行的stage各个task分配的数据量，从而进一步确定是不是task分配的数据不均匀导致了数据倾斜。    知道数据倾斜发生在哪一个stage之后，接着我们就需要根据stage划分原理，推算出来发生倾斜的那个stage对应代码中的哪一部分，这部分代码中肯定会有一个shuffle类算子。通过countByKey查看各个key的分布。3、数据倾斜解决方案过滤少数导致倾斜的key提高shuffle操作的并行度局部聚合和全局聚合

4.2 shuffle类算子

1、去重：def distinct()def distinct(numPartitions: Int)2、聚合def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]def groupBy[K](f: T => K, p: Partitioner):RDD[(K, Iterable[V])]def groupByKey(partitioner: Partitioner):RDD[(K, Iterable[V])]def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner): RDD[(K, U)]def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int): RDD[(K, U)]def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]3、排序def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]def sortBy[K](f: (T) => K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]4、重分区def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null)5、集合或者表操作def intersection(other: RDD[T]): RDD[T]def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]def intersection(other: RDD[T], numPartitions: Int): RDD[T]def subtract(other: RDD[T], numPartitions: Int): RDD[T]def subtract(other: RDD[T], p: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]def subtractByKey[W: ClassTag](other: RDD[(K, W)]): RDD[(K, V)]def subtractByKey[W: ClassTag](other: RDD[(K, W)], numPartitions: Int): RDD[(K, V)]def subtractByKey[W: ClassTag](other: RDD[(K, W)], p: Partitioner): RDD[(K, V)]def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]