1:groupByKey

def groupByKey(): RDD[(K, Iterable[V])]

根据key进行聚集,value组成一个列表,没有进行聚集,所以在有shuffle操作时候避免使用概算子,会增大通信数据量。需要考虑进行一个本地的Combiner,所以可以直接使用reduceByKey

cala> p.collect
res15: Array[(Int, Int)] = Array((1,1), (2,1), (1,1), (2,1), (1,1), (2,1), (3,1), (4,1), (5,1)) scala> p.groupByKey.collect
res16: Array[(Int, Iterable[Int])] = Array((4,CompactBuffer(1)), (2,CompactBuffer(1, 1, 1)), (1,CompactBuffer(1, 1, 1)), (3,CompactBuffer(1)), (5,CompactBuffer(1)))

2:cogroup

def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W]))]

对每一个rdd先进行groupByKey,然后在对相同key的value在进行一个groupByKey,例如

scala> p.collect
res18: Array[(Int, Int)] = Array((1,1), (2,1), (1,1), (2,1), (1,1), (2,1), (3,1), (4,1), (5,1)) scala> p.cogroup(p).collect
res19: Array[(Int, (Iterable[Int], Iterable[Int]))] = Array((4,(CompactBuffer(1),CompactBuffer(1))), (2,(CompactBuffer(1, 1, 1),CompactBuffer(1, 1, 1))), (1,(CompactBuffer(1, 1, 1),CompactBuffer(1, 1, 1))), (3,(CompactBuffer(1),CompactBuffer(1))), (5,(CompactBuffer(1),CompactBuffer(1))))

例如:(1,1)在pair中出现三次,所以cogroup之后(1,1)的结果是:

(1,(CompactBuffer(1, 1, 1),CompactBuffer(1, 1, 1)))

3:aggregateByKey

def aggregateByKey[U](zeroValue: U)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]

说明:这个函数相对有点难懂,其他重载版本与此类似。该函数接受三个参数,一个初始值,两个函数:seqOp和comOp:

seqOp是对分区进行具体的函数,zeroValue值只有在seqOp中有使用,在第二个函数comOp中就不在使用了。

comOp是分区之间的combine函数(combine函数有点类似combiner的作用,进行聚集的函数)。

例如:

    val data = sc.parallelize(List((1, 3), (1, 2), (1, 4), (2, 3), (2, 5)))
def seq(a: Int, b: Int): Int = {
println("seq: " + a + "\t " + b)
math.max(a, b)
}
def comb(a: Int, b: Int): Int = {
println("comb: " + a + "\t " + b)
a + b
}

执行过程输出:

seq: 1     3
seq: 1 2
seq: 1 4
seq: 1 3
seq: 1 5
comb: 3 2
comb: 5 4
comb: 3 5

说明:在调用seqOp函数时候,每一次都向(key,value)中的value和zeroValue进行求最大值,将该最大值作为key的value。执行完毕seqOp函数的状态为:

((1,(3,2,4)),(2(3,5)))

然后调用comOp函数将每一个key的value进行累加,得到最后输出。

输出:

scala> data.aggregateByKey(1)(seq, comb).collect.toBuffer
res36: scala.collection.mutable.Buffer[(Int, Int)] = ArrayBuffer((2,5), (1,7))

4:combineByKey

def combineByKey[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C): RDD[(K, C)]

参数说明:

  • createCombiner, which turns a V into a C (e.g., creates a one-element list)
  • mergeValue, to merge a V into a C (e.g., adds it to the end of a list)
  • mergeCombiners, to combine two C's into a single one.

例如:

val data1 = sc.parallelize(List("a", "b", "c", "c", "b", "a", "b", "a"))
val data2 = sc.parallelize(List(1, 2, 3, 3, 2, 1, 2, 1))
val zip = data2.zip(data1)
val combineByKey = zip.combineByKey(List(_), (x: List[String], y: String) => y :: x, (x: List[String], y: List[String]) => x ::: y)
println(combineByKey.collect().toBuffer)

输出:

ArrayBuffer((1,List(a, a, a)), (2,List(b, b, b)), (3,List(c, c)))

5:flatMapValues

def flatMapValues[U](f: (V) ⇒ TraversableOnce[U]): RDD[(K, U)]

传入一个键值对的值给 (V) ⇒ TraversableOnce[U]函数,返回的是一个集合的函数。将当前的key和当前集合中每一个元素组成元组返回

val a = sc.parallelize(List((1,2),(3,4),(5,6)))
val b = a.flatMapValues(x=>1 to x)
b.collect.foreach(println(_))

输出:

(1,1)
(1,2)
(3,1)
(3,2)
(3,3)
(3,4)
(5,1)
(5,2)
(5,3)
(5,4)
(5,5)
(5,6)

分析:当传入是2的时候,生成一个1 to 2 的序列,然后当前key=1,所以生成(1,1),(1,2)两个元组、

 6:foldByKey

def foldByKey(zeroValue: V)(func: (V, V) ⇒ V): RDD[(K, V)]

fold:折叠的意思,根据key进行折叠,例如:

scala> rdd.collect
res2: Array[String] = Array(hello, world, xiaomi, meizu, meizu) scala> rdd.map((_,1)).foldByKey(1)(_+_).collect
res3: Array[(String, Int)] = Array((meizu,3), (hello,2), (world,2), (xiaomi,2))

此处zeroValue为1,由于meizu这个字符出现两次并加上zeroValue的话恰好是3。hello world xiaomi这些单词都是一次,并加上zeroValue恰好是2。

当zeroValue值为0的时候,我们可以实现一个wordcount。

7:mapValues

def mapValues[U](f: (V) ⇒ U): RDD[(K, U)]

对键值对的每一个value进行处理,key保持不变。例如:我对现有的键值对的值进行乘以10的操作

scala> rdd.map((_,1)).collect
res14: Array[(String, Int)] = Array((hello,1), (world,1), (xiaomi,1), (meizu,1), (meizu,1)) scala> rdd.map((_,1)).mapValues(x=>x*10).collect
res15: Array[(String, Int)] = Array((hello,10), (world,10), (xiaomi,10), (meizu,10), (meizu,10))

8:rightOuterJoin

def rightOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], W))]

右外连接,other中的所有key都会出现在结果中,关联到的是一个Some值,关联不到的是一个None。例如:

scala> rdd1.collect
res26: Array[(Int, Int)] = Array((1,1), (2,1), (3,1), (4,1), (5,1)) scala> rdd2.collect
res27: Array[(Int, Int)] = Array((3,2), (4,2), (5,2), (6,2), (7,2), (8,2)) scala> rdd1.rightOuterJoin(rdd2).collect
res28: Array[(Int, (Option[Int], Int))] = Array((4,(Some(1),2)), (6,(None,2)), (8,(None,2)), (3,(Some(1),2)), (7,(None,2)), (5,(Some(1),2)))

9:leftOuterJoin

def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]

同上面右外连接。示例:

scala> rdd1.collect
res31: Array[(Int, Int)] = Array((1,1), (2,1), (3,1), (4,1), (5,1)) scala> rdd2.collect
res32: Array[(Int, Int)] = Array((3,2), (4,2), (5,2), (6,2), (7,2), (8,2)) scala> rdd1.leftOuterJoin(rdd2).collect
res33: Array[(Int, (Int, Option[Int]))] = Array((4,(1,Some(2))), (2,(1,None)), (1,(1,None)), (3,(1,Some(2))), (5,(1,Some(2))))

rdd中的所有的key都会出现,关联的不到的为None,此时关联的值在value的第二个位置。我们可以通过交换右外连接的两个rdd的位置,实现左外连接,但是区别在于value中元素的位置是逆序的:

scala> rdd2.rightOuterJoin(rdd1).collect //通过右外连接实现左外连接
res34: Array[(Int, (Option[Int], Int))] = Array((4,(Some(2),1)), (2,(None,1)), (1,(None,1)), (3,(Some(2),1)), (5,(Some(2),1)))
05-11 23:01