前面章节介绍的都是一些RDD中常用的简单算子，本节就来介绍一些RDD中常用的高级算子，比如mapPartitionsWithIndex、aggregate： aggregateByKey、coalesce与repartition等。更多关于RDD中算子的详细内容请参考下面的文章：
1.mapPartitionsWithIndex算子
作用：把RDD中每个分区中的元素和对应的分区号拿出来，进行计算；
定义如下：
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]
参数1：函数 f: (Int, Iterator[T]) => Iterator[U]
第一个参数：Int 分区号
第二个参数：Iterator[T] 该分区元素的集合
返回值：Iterator[U] 是集合
参数2：preservesPartitioning: Boolean = false 是否保留分区，默认不保留</pre>
示例：打印RDD的每个元素，显示分区号。
（1）创建一个包含9个整型元素的RDD，并指定2个分区
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0]
at parallelize at <console>:24
（2）定义函数参数：对每个分区中的元素进行处理，打印分区中的元素和对应的分区号
scala> def func1(index:Int,iter:Iterator[Int]):Iterator[String] = {
| iter.toList.map(x => "[PartID:"+index + ", value=" + x +"]").iterator
| }
func1: (index: Int, iter: Iterator[Int])Iterator[String]</pre>
（3）调用mapPartitionsWithIndex算子并显示结果
scala> rdd1.mapPartitionsWithIndex(func1).collect
res0: Array[String] = Array([PartID:0, value=1],[PartID:0, value=2],
[PartID:0, value=3],[PartID:0, value=4], [PartID:1, value=5], [PartID:1, value=6],
[PartID:1, value=7], [PartID:1, value=8], [PartID:1, value=9])
2.aggregate算子
作用：先对局部（每个分区）进行聚合操作，然后再对全局（所有分区）进行聚合操作。
定义如下：
def aggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)
参数1：zeroValue: U 初始值（每个分区中放一个，全局再放一个）
参数2：seqOp: (U, T) ⇒ U 局部聚合匿名函数
参数3：combOp: (U, U) ⇒ U 全局聚合匿名函数</pre>
示例1：将RDD中每个分区的最大值求和。
（1）创建一个包含9个整型元素的RDD，并指定3个分区
scala> val rdd2 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0]
at parallelize at <console>:24
（2）查看每个分区中对应的元素：
scala> rdd2.mapPartitionsWithIndex(func1).collect
res1: Array[String] = Array(
[PartID:0, value=1], [PartID:0, value=2], [PartID:0, value=3],
[PartID:1, value=4], [PartID:1, value=5], [PartID:1, value=6],
[PartID:2, value=7], [PartID:2, value=8], [PartID:2, value=9])
（3）调用aggregate算子：将每个分区最大值求和
scala> rdd2.aggregate(0)(math.max(,),+)
res2: Int = 18
可以看到三个分区的最大值分别是3，6，9，和是18。如果参数1不是0呢？
scala> rdd2.aggregate(10)(math.max(,),+)
res3: Int = 40
总结：aggregate算子会在每个分区中的所有元素前面添加一个初始值元素，然后在全局中所有元素前面再添加一个初始值元素。为了更好的看到这个效果，请看下面的例子：
示例2：连接RDD中每个元素。
（1）创建一个包含6个字符串元素的RDD，并指定2个分区
scala> val rdd3 = sc.parallelize(List("a","b","c","d","e","f"),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[4]
at parallelize at <console>:24
（2）查看每个分区中对应的元素：
scala> def func2(index:Int,iter:Iterator[String]):Iterator[String] = {
| iter.toList.map(x => "[PartID:"+index + ", value=" + x +"]").iterator
| }
func2: (index: Int, iter: Iterator[String])Iterator[String]scala> rdd3.mapPartitionsWithIndex(func2).collect
res4: Array[String] = Array(
[PartID:0, value=a], [PartID:0, value=b], [PartID:0, value=c],
[PartID:1, value=d], [PartID:1, value=e], [PartID:1, value=f])
（3）调用aggregate算子：连接RDD中的每个元素：
scala> rdd3.aggregate("")(+,+)
res6: String = abcdef
初始值可以设置为分割符：
scala> rdd3.aggregate("|")(+,+)
res7: String = ||abc|def
3.aggregateByKey算子
作用：先对局部（每个分区）进行聚合操作，然后再对全局（所有分区）进行聚合操作。与aggregate算子的区别是，aggregateByKey算子操作的数据类型必须是<key value>，而aggregate算子的操作的数据类型不限。
示例：将RDD每个分区中每个键的最大值求和。
（1）创建一个包含6个<String,Int>类型元素的RDD，并指定2个分区
scala> val rdd4 = sc.parallelize(List(("cat",2),("cat",5),
("mouse",4),("cat",12),("dog",12),("mouse",2)),2)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] =
ParallelCollectionRDD[6] at parallelize at <console>:24
（2）查看每个分区中对应的元素：
scala> def func3(index:Int,iter:Iterator[(String,Int)]):Iterator[String] = {
| iter.toList.map(x => "[PartID:"+index + ", value=" + x +"]").iterator
| }
func3: (index: Int, iter: Iterator[(String, Int)])Iterator[String]scala> rdd4.mapPartitionsWithIndex(func3).collect
res8: 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)])
（3）调用aggregateByKey算子：将RDD每个分区中每个键的最大值求和：
scala> rdd4.aggregateByKey(0)(math.max(,),+).collect
res9: Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,6))
（4）调用aggregateByKey算子：将RDD每个分区中每个键的所有值求和：
scala> rdd4.aggregateByKey(0)(+,+).collect
res10: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))
4.coalesce与repartition算子
相同点：都是对RDD原有的分区进行重新划分。
不同点：coalesce默认没有shuffle操作，repartition有shuffle操作。
二者关系：repartition内部调用的是coalesce 。这可以从repartition的定义式看出：
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
coalesce(numPartitions, shuffle = true)
}
示例1：使用repartition将RDD重新分区。
（1）创建一个包含9个整型元素的RDD，并指定2个分区
scala> val rdd5 = sc.parallelize(Array(1,2,3,4,5,6,7,8,9),2)
rdd5: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[10]
at parallelize at <console>:24
2）查看每个元素原来的分区情况：
scala> rdd5.mapPartitionsWithIndex(func1).collect
res11: Array[String] = Array([PartID:0, value=1],
[PartID:0, value=2], [PartID:0, value=3], [PartID:0, value=4],
[PartID:1, value=5], [PartID:1, value=6], [PartID:1, value=7],
[PartID:1, value=8], [PartID:1, value=9])
（3）使用repartition将RDD重新分区
scala> val rdd6 = rdd5.repartition(3)
rdd6: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[15]
at repartition at <console>:26
（4）查看每个元素新的分区情况：
scala> rdd6.mapPartitionsWithIndex(func1).collect
res12: Array[String] = Array([PartID:0, value=3], [PartID:0, value=7],
[PartID:1, value=1], [PartID:1, value=4], [PartID:1, value=5],
[PartID:1, value=8], [PartID:2, value=2], [PartID:2, value=6],
[PartID:2, value=9])
示例2：使用coalesce将RDD重新分区。
（1）创建一个包含9个整型元素的RDD，并指定2个分区
scala> val rdd7 = sc.parallelize(Array(1,2,3,4,5,6,7,8,9),2)
rdd7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[17]
at parallelize at <console>:24
（2）查看每个元素原来的分区情况：
scala> rdd7.mapPartitionsWithIndex(func1).collect
res13: Array[String] = Array([PartID:0, value=1], [PartID:0, value=2],
[PartID:0, value=3], [PartID:0, value=4], [PartID:1, value=5],
[PartID:1, value=6], [PartID:1, value=7], [PartID:1, value=8],
[PartID:1, value=9])
（3）使用coalesce将RDD重新分区
scala> val rdd8 = rdd7.coalesce(3)
rdd8: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[19]
at coalesce at <console>:26
（4）查看每个元素新的分区情况：
scala> rdd8.mapPartitionsWithIndex(func1).collect
res14: Array[String] = Array([PartID:0, value=1], [PartID:0, value=2],
[PartID:0, value=3], [PartID:0, value=4], [PartID:1, value=5],
[PartID:1, value=6], [PartID:1, value=7], [PartID:1, value=8],
[PartID:1, value=9])
对比示例1和示例2的结果可以发现，repartition算子重新分区后元素的顺序变化了，证明发生了shuffle操作；coalesce算子重新分区后元素保持原来的顺序，证明没有shuffle操作。
本节详细介绍了几个RDD常用的高级算子。祝你玩的愉快！
