使用mapPartitions进行RDD分区操作

转换操作是Spark RDD的两大核心操作之一，它使用了链式调用的设计模式，对一个RDD进行计算后，变换成另外一个RDD，然后这个RDD又可以进行另外一次转换。这个过程是分布式的，它必须等行动操作（action）出现后，才真正触发Spark提交作业，开始执行计算。对于转换操作，可以分为两类：一类是对Value型数据进行的操作，一类是对Key-Value类型进行的操作。

1.Value型数据转换操作

（1）map方法将原来RDD中类型为T的元素，通过map中的用户自定义函数f一对一地映射为U类型的元素。新产生的RDD的实际类型是MappedRDD。

（2）mapPartitions与map转换操作类似，只不过映射函数的输入参数由RDD中的每一个元素变成了RDD的中的每一个分区的迭代器，在函数中通过这个分区的迭代器对整个分区的元素进行操作。最后生成的MapPartitionRDD。

（3）mapPartitionsWithIndex和mapPartitions基本类似，只是输入参数多了一个分区的TaskContext。一个TaskContext对象包含一个StageId和一个PartitionId。

（4）zip方法的功能是将两个RDD组成后成键值对（Key-Value）形式的RDD，这里默认两个RDD的partition数量以及元素数量都相同。

（5）union方法是将两个RDD集合中的数据进行合并，返回两个RDD的并集。使用u-nion需要注意的是要保证两个RDD的数据类型一致，保存所有的元素，并不会去掉重复的元素。

（6）distinct方法是将RDD执行去重操作，操作的结果是有重复的元素只保留一份。

（7）filter方法是对RDD中的元素进行过滤操作，对RDD中的每个元素应用f函数，返回值为true的元素保留，返回值为false的元素将被过滤掉，最后生成一个FilteredRDD。

（8）toDebugString方法是用来描述RDD和它依赖的父RDD之间的关系的，我们在spark-shell中经常会用到，来查看一个RDD的产生在中间经过了那些显性和隐性的trans-formation操作。

（9）三个substact方法都相当于进行集合的差操作，RDD1去除RDD1和RDD2交集中的所有元素。

第一个subtract中的参数Partitioner表明可以选用Hash或者Range分区，第二个substact方法中的numPartition表示可以指定分区个数，第三个subtract方法使用的是RDD1默认的分区数。

（10）coalesce和repartition方法都是对RDD分区的重新划分。repartition只是coalesce方法中shuffle为true的简易实现。Shuffle为true的情况指的是的需要重新划分成的分区个数比原来的RDD的分区个数多。

（11）sample方法是将RDD集合中的元素进行采样，返回原来元素的一个子集。用户可以通过传参设定是否有放回的抽样、百分比、随机种子，进而决定采用何种抽样方式。

（12）preferredLocations方法可以通过传入的参数split来指定RDD的优先计算位置。

（13）这三个intersection方法都是指返回两个RDD的交集，并且交集中不会包含相同的元素。第一个带参数numPartitions的intersection方法可以指定分区大小，第二个带参数Partitioner的intersection方法可以选择采用Hash分区还是Range分区，第三个intersection方法的分区大小使用的调用它的RDD的分区大小。

（14）groupBy指的是将元素通过函数f生成相应的Key，数据格式就转换为键值对（key-value）形式，之后将Key相同的元素分为一组。(www.xing528.com)

（15）glom方法是将RDD中的每一个分区中类型为T的元素转换成数组Array[T]，最后返回的RDD类型是GlommedRDD。

（16）flatMap方法则是将原来RDD中的每个元素通过函数f转换为新的元素，并将生成的RDD的每个集合中的元素合并为一个集合。

（17）randomSplit方法是根据参数weights权重将一个RDD切分成多个RDD。

2.Key-Value型RDD转换操作

（1）mapValue方法和flatMapValue方法是对[Key-Value]类型的数据中的V值分别进行map操作和flatMap操作。

（2）partitionBy方法是对RDD进行分区操作。如果原有RDD的分区器和现在分区器（partitioner）一致，则不重新分区。如果不一致，则会根据分区器生成一个新的Shuffle-dRDD。

（3）combineByKey方法是将RDD[K，V]转换成返回类型RDD[K，C]，createCombiner负责如何将value转换成combine的输入C，mergeCombiners定义如何合并两个C，mergeValue定义如何将新来的V与已有的C合并，partitioner指定分区，MapSideCombine指定是否需要在Map端进行combine操作。

（4）reduceByKey方法在一个（K，V）对的数据集上使用时，返回一个（K，V）对的数据集，key相同的值，都被使用指定的reduce函数聚合到一起。在reduceByKey方法内部会继续调用combineByKey[C]（createCombiner：（V）⇒C，mergeValue：（C，V）⇒C，mergeCom-biners：（C，C）⇒C，numPartitions：Int）：RDD[（K，C）]方法，numPartitions会被包装成new HashPartitioner（numPartitions）。

（5）foldByKey方法类似于reduceByKey方法，只是在reduce函数聚合的时候可能会加上zeroValue的值。

（6）groupByKey方法在一个由（K，V）对组成的数据集上调用，返回一个（K，Seq[V]）对的数据集。在groupByKey方法内部也是使用了comBineByKey方法来完成操作最终的实现，在groupByKey方法内部首先会将createCombiner、mergeValue和mergeCombiners这三个函数具体化，然后再将这三个函数作为参数传递给groupByKey方法内部的comBineByKey方法。下面是createCombiner、mergeValue和mergeCombiners这三个函数的具体实现：

bufs＝combineByKey[Arraybuffer[V]]（）。逻辑是将同一个Key的value简单地放到一个ArrayBuffer里，最后返回bufs.asInstanceOf[RDD[（K.Seq[V]）]]

（7）当有两个key-Value类型的元素（K，V）和（K，W）时，cogroup方法返回的是（K，Seq[V]，Seq[W]）的集合的迭代器，numPartitions为并发的任务数。

（8）join方法是对两个需要连接的RDD进行cogroup操作，然后对每个key下的元素进行笛卡尔积操作，返回的结果再展平。

（9）leftOutJoin方法与join方法一样，都是针对RDD[K，V]中K值相等的连接操作，分别对应于左外连接和内连接，最终都会调用cogroup函数来实现。

（10）SortByKey方法是按照Key的大小进行排序，默认是升序的方式排序，如果想降序排序，设置Boolean为false。