zipWithIndex:首先基于分区索引  然后基于分区内元素索引 第一个元素是第一个分区的第一个元素 最后一个元素是最后一个分区的最后一个元素 Index的返回类型是Long类型而不是Int 如果RDD不止一个分区，则触发一个spark job，如果是根据groupBy()返回的RDD 不能保证一个分区内的元素排序，所以 如果需要确保每一个元素的索引序列，需要针对RDD使用sortByKey() 算子 进行sort  或者保存进一个文件

val seqRdd = sc.parallelize(List("Mary","Jim","Green","Jack","Tony"))

seqRdd.mapPartitionsWithIndex{(index,iter) => iter.toList.map(x => ("partitionIndex:0"+Index,x)).toIterator}.collect().foreach(println)

// (partitionIndex:00,Mary)

// (partitionIndex:00,Jim)

// (partitionIndex:01,Green)

// (partitionIndex:01,Jack)

// (partitionIndex:01,Tony)

seqRdd.zipWithIndex().collect().foreach(println)

// (Mary,0)

// (Jim,1)

// (Green,2)

// (Jack,3)

// (Tony,4)

seqRdd.zipWithIndex().mapPartitionsWithIndex{(index,iter) => iter.toList.map(x => ("partitionIndex:0"+Index,x)).toIterator}.collect().foreach(println)

// (partitionIndex:00,(Mary,0))

// (partitionIndex:00,(Jim,1))

// (partitionIndex:01,(Green,2))

// (partitionIndex:01,(Jack,3))

// (partitionIndex:01,(Tony,4))

zip:

将当前RDD与另外一个RDD进行zip操作，返回从两个RDD返回的第一个元素对/第二个元素对 作为key-value对，假定两个RDD 具有两个相同数据量的分区且每个分区的元素数量相同，如果存在某一个分区对，其元素数量不相同，则抛出SparkException：Can Only zip RDDs with the same number of elements in each partition

CollectAsMap:将当前RDD的key-value 对 以Map形式返回至master节点，如果一个key有多个value。只能返回一个value

seqRdd.zip(seqRdd).collectAsMap().foreach(println);seqRdd.zip(seqRdd).foreach(println);

// (Jim,Jim) (Jack,Jack) (Green,Green) (Mary,Mary) (Tony,Tony)

// (Mary,Mary)(Jim,Jim)(Green,Green)(Jack,Jack)(Tony,Tony)

zipWithUniqueId:

使用唯一的Long id 与当前RDD 进行Zips操作，在第K个分区内的元素（其分区内索引分别为0，1，2，3，4）的id分别为：k,k+n,k+2*n,k+3*n... n为分区总数，因此在这里存在一定的跳跃，但是与zipWithIndex不同，该方法不会触发spark job，其他与其一致。

seqRdd.zipWithUniqueId().collect.foreach(println)

// (Mary,0)

// (Jim,2)

// (Green,1)

// (Jack,3)

// (Tony,5)

// 诸如 combineByKey等算子，在指定参数名及其具体值时，需要注意其参数名应与声明时参数名一致。

val seq1Rdd = sc.parallelize(List("Mary","Jim","Green","Jack","Tony"))

val seq2Rdd = sc.parallelize(List(1,2,1,2,1))

val zipRdd = seq2Rdd.zip(seq1Rdd)

combineRdd：方法向后兼容，使用一系列常用的聚合函数对RDD的每个Key 联合组成其value值

val combineRdd = zipRdd.combineByKey(createCombiner = (x:String) => List(x),mergeValue = (x:List[String],y:String) => x.:+(y),mergeCombiners = (x:List[String],y:List[String]) => x.:::(y))

combineRdd.foreach(println)

// （2,(List(Jack,Jim))）

// （1,(List(Green,Tony,Mary))）

解析：

combineByKey属于Key-value算子，做的是聚合操作，这种变换不会触发作业的提交，主要有三个参数：

createCombiner function:一个组合函数 用于将RDD[K,V] 中的V转换成一个新的值C1

mergeValue function: 合并值函数,将一个C1类型值和一个V类型值合并成一个C2类型，输入参数为(C1,V) 输出为新的C2

mergeCombiners function：合并组合器函数 用于将两个C2类型值合并成一个C3类型 输入参数为(C2,C2) 输出为C2

val createCombine = (x:String) => List(x)

val mergeValue = (x:List[String],y:String) => y :: x

// Adda an element at the beginning of this list.{{{ 1 :: List(2,3) = List(2,3).::(1) = List(1,2,3)}}}

val mergerCombiners = (x:List[String],y:List[String]) => x ::: y // Adds the elements of a given list in front of this list. {{{ List(1,2) ::: List(3,4) = List(3,4).:::List(1,2) = List(1,2,3,4)}}}

val combineRdd2 = zipRdd.combineByKey(createCombine,mergeValue,mergerCombiners)

combineRdd2.foreach(println)

// （2,(List(Jim，Jack))）

// （1,(List(Mary，Tony，Green))）

aggregateByKey:

使用combine组合函数和一个中性netural“zero value”对每一个key的多个value进行聚合操作，可以返回不同于RDD中固有数据类型V的结果类型U，一个merge操作将V 转换为U，另外一个merge操作将两个U聚合，（集合可反复遍历） 之前的数据类型转换操作是在一个分区内部进行的  之后的将U进行聚合操作是在分区之间进行的。

为避免内存泄漏，这些函数均可以被修改并且返回他们的第一个参数而不是创建一个新的U实例。 计算的时候与分区的关系很大，注意分区的作用。

val seqRdd = sc.parallelize(List(("cat",2),("dog",12),("cat",12),("cat",5),("mouse",4),("mouse",2)))

seqRdd.mapPartitionsWithIndex{(index,iter) => iter.toList.map(x => ("partitionIndex:0"+Index,x)).toIterator}.collect().foreach(println)

// (partitionIndex:00,(cat,2))

// (partitionIndex:00,(dog,12))

// (partitionIndex:00,(cat,12))

// (partitionIndex:01,(cat,5))

// (partitionIndex:01,(mouse,4))

// (partitionIndex:01,(mouse,2))

val aggregateFuncRdd = seqRdd.aggregateByKey[Int](zeroValue = 0)(seqOp = (value:Int,zeroValue:Int) => math.max(value,zeroValue),combOp = (value1:Int,value2:Int) => value1 + value2)

aggregateFuncRdd.collect.foreach(println)

// (dog,12) (cat,17) (mouse,4)

seqRdd.aggregateByKey[Int](zeroValue = 10)(seqOp = (value:Int,zeroValue:Int) => math.max(value,zeroValue),combOp = (value1:Int,value2:Int) => value1 + value2).collect.foreach(println)

// (dog,12) (cat,22) (mouse,10)  首先和当前分区内的同一key的多个value与zeroValue结合进行seqOp运算，分区0内cat 对应的value最大为12，dog为12，分区1内cat对应的value最大为5 mouse为4  再将其与ZeroValue进行计算 0-cat-Max 12 0-dog-Max 12 1-cat-Max 10 1-mouse-Max 10  然后再将两个分区的同一key的多个value进行相加  得到 (dog,12) (cat,22) (mouse,10)

seqRdd.aggregateByKey[Int](zeroValue = 10)(seqOp = (value:Int,zeroValue:Int) => math.max(value,zeroValue),combOp = (value1:Int,value2:Int) => value1 + value2).collect.foreach(println)

// (dog,100) (cat,200) (mouse,100)

countByKey:

计算每一个Key对应的元素数量，并讲结果collect至本地Map格式，结果Map必须相对较小 才能够全部加载进Driver's 内存 处理大数据量的话  可以返回一个RDD[K,Long] 代替一个Map

seqRdd.countByKey() // self.mapValue(_ => 1L).reduceByKey(_+_).collect().toMap

seqRdd.countByValue()// map(value => (value,null)).countByKey()  当前value不是key-value对的value  而是指代整个key-value 以（value，count）对的本地Map形式返回当前RDD的不同value的个数（此处应该是一个Bug）

seqRdd.map(_._2).countByValue()

sortByKey():

进行了shuffle操作，shuffle之后被重新分区，排序靠前的元素在低序号分区，排序靠后的元素在高序号分区，每一个分区内包含一段已经排序好的元素，针对结果如果调用collect 或者save算子，将会返回或者输出一段已经排序好的记录，调用save算子时将会在文件系统内生成多个依照key进行排序的“part-x”的文件

val ze = sc.parallelize(List("dog","tiger","tac","cat","gnu","panther"),2)

val zf = ze.map(x => (x.length,x))

zf.mapPartitionsWithIndex{(index,iter) => iter.toList.map(x => ("partitionIndex:0"+Index,x)).toIterator}.collect().foreach(println)

// (partitionIndex:00,(3,dog))

// (partitionIndex:00,(5,tiger))

// (partitionIndex:00,(3,tac))

// (partitionIndex:01,(3,cat))

// (partitionIndex:01,(3,gnu))

// (partitionIndex:01,(7,panther))

zf.sortByKey().foreachPartition(x => println("sortByKey....",x.toList.mkString(",")))

// (sortByKey....,(5,tiger),(7,panther))

// (sortByKey....,(3,dog),(3,tac),(3,cat),(3,gnu))

zf.sortByKey().mapPartitionsWithIndex{(index,iter) => iter.toList.map(x => ("partitionIndex:0"+Index,x)).toIterator}.collect().foreach(println)

// (partitionIndex:00,(3,dog))

// (partitionIndex:00,(3,tac))

// (partitionIndex:00,(3,cat))

// (partitionIndex:00,(3,gnu))

// (partitionIndex:01,(5,tiger))

// (partitionIndex:01,(7,panther))

foldByKey():使用一个组合函数 和中性值 neutral “zero value” 数据类型与Value类型相同，执行时首先针对某一分区内相关元素进行组合函数计算 然后 结合 zero value 进行计算。。 然后根据Key将不同分区的组合计算结果进行组合函数运算

zf.foldByKey(zeroValue="#")((x:String,y:String) => x + y)..mapPartitionsWithIndex{(index,iter) => iter.toList.map(x => ("foldByKey:"+Index,x)).toIterator}.collect().foreach(println)

// (foldByKey:1,(3,#dogtac#catgnu))

// (foldByKey:1,(7,#panther))

// (foldByKey:1,(5,#tiger))

Join:

fullOuterJoin:全连接 （key,(Some(v1),Some(v2))）

join:连接 (key,(v1,v2))

leftOuterJoin:左连接 （key,(v1,Some(v2))）

rightOuterJoin:右连接 （key,(Some(v1),v2))