浅谈Spark RDD API中的Map和Reduce

rdd是什么？

rdd是spark中的抽象数据结构类型，任何数据在spark中都被表示为rdd。从编程的角度来看，rdd可以简单看成是一个数组。和普通数组的区别是，rdd中的数据是分区存储的，这样不同分区的数据就可以分布在不同的机器上，同时可以被并行处理。因此，spark应用程序所做的无非是把需要处理的数据转换为rdd，然后对rdd进行一系列的变换和操作从而得到结果。本文为第一部分，将介绍spark rdd中与map和reduce相关的api中。

如何创建rdd？

rdd可以从普通数组创建出来，也可以从文件系统或者hdfs中的文件创建出来。

举例：从普通数组创建rdd，里面包含了1到9这9个数字，它们分别在3个分区中。

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.rdd[int] = parallelcollectionrdd[1] at parallelize at <console>:12

举例：读取文件readme.md来创建rdd，文件中的每一行就是rdd中的一个元素

scala> val b = sc.textfile("readme.md")
b: org.apache.spark.rdd.rdd[string] = mappedrdd[3] at textfile at <console>:12

虽然还有别的方式可以创建rdd，但在本文中我们主要使用上述两种方式来创建rdd以说明rdd的api。

map

map是对rdd中的每个元素都执行一个指定的函数来产生一个新的rdd。任何原rdd中的元素在新rdd中都有且只有一个元素与之对应。

举例：

scala> val a = sc.parallelize(1 to 9, 3)
scala> val b = a.map(x => x*2)
scala> a.collect
res10: array[int] = array(1, 2, 3, 4, 5, 6, 7, 8, 9)
scala> b.collect
res11: array[int] = array(2, 4, 6, 8, 10, 12, 14, 16, 18)

上述例子中把原rdd中每个元素都乘以2来产生一个新的rdd。

mappartitions

mappartitions是map的一个变种。map的输入函数是应用于rdd中每个元素，而mappartitions的输入函数是应用于每个分区，也就是把每个分区中的内容作为整体来处理的。
它的函数定义为：

def mappartitions[u: classtag](f: iterator[t] => iterator[u], preservespartitioning: boolean = false): rdd[u]

f即为输入函数，它处理每个分区里面的内容。每个分区中的内容将以iterator[t]传递给输入函数f，f的输出结果是iterator[u]。最终的rdd由所有分区经过输入函数处理后的结果合并起来的。

举例：

scala> val a = sc.parallelize(1 to 9, 3)
scala> def myfunc[t](iter: iterator[t]) : iterator[(t, t)] = {
  var res = list[(t, t)]() 
  var pre = iter.next while (iter.hasnext) {
    val cur = iter.next; 
    res .::= (pre, cur) pre = cur;
  } 
  res.iterator
}
scala> a.mappartitions(myfunc).collect
res0: array[(int, int)] = array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8))

上述例子中的函数myfunc是把分区中一个元素和它的下一个元素组成一个tuple。因为分区中最后一个元素没有下一个元素了，所以(3,4)和(6,7)不在结果中。

mappartitions还有些变种，比如mappartitionswithcontext，它能把处理过程中的一些状态信息传递给用户指定的输入函数。还有mappartitionswithindex，它能把分区的index传递给用户指定的输入函数。

mapvalues

mapvalues顾名思义就是输入函数应用于rdd中kev-value的value，原rdd中的key保持不变，与新的value一起组成新的rdd中的元素。因此，该函数只适用于元素为kv对的rdd。

举例：

scala> val a = sc.parallelize(list("dog", "tiger", "lion", "cat", "panther", " eagle"), 2)
scala> val b = a.map(x => (x.length, x))
scala> b.mapvalues("x" + _ + "x").collect
res5: array[(int, string)] = array((3,xdogx), (5,xtigerx), (4,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))

mapwith

mapwith是map的另外一个变种，map只需要一个输入函数，而mapwith有两个输入函数。它的定义如下：

def mapwith[a: classtag, u: ](constructa: int => a, preservespartitioning: boolean = false)(f: (t, a) => u): rdd[u]

第一个函数constructa是把rdd的partition index（index从0开始）作为输入，输出为新类型a；

第二个函数f是把二元组(t, a)作为输入（其中t为原rdd中的元素，a为第一个函数的输出），输出类型为u。

举例：把partition index 乘以10，然后加上2作为新的rdd的元素。

val x = sc.parallelize(list(1,2,3,4,5,6,7,8,9,10), 3) 
x.mapwith(a => a * 10)((a, b) => (b + 2)).collect 
res4: array[int] = array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)

flatmap

与map类似，区别是原rdd中的元素经map处理后只能生成一个元素，而原rdd中的元素经flatmap处理后可生成多个元素来构建新rdd。

举例：对原rdd中的每个元素x产生y个元素（从1到y，y为元素x的值）

scala> val a = sc.parallelize(1 to 4, 2)
scala> val b = a.flatmap(x => 1 to x)
scala> b.collect
res12: array[int] = array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4)

flatmapwith

flatmapwith与mapwith很类似，都是接收两个函数，一个函数把partitionindex作为输入，输出是一个新类型a；另外一个函数是以二元组（t,a）作为输入，输出为一个序列，这些序列里面的元素组成了新的rdd。它的定义如下：

def flatmapwith[a: classtag, u: classtag](constructa: int => a, preservespartitioning: boolean = false)(f: (t, a) => seq[u]): rdd[u]

举例：

scala> val a = sc.parallelize(list(1,2,3,4,5,6,7,8,9), 3)
scala> a.flatmapwith(x => x, true)((x, y) => list(y, x)).collect
res58: array[int] = array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2,
8, 2, 9)

flatmapvalues

flatmapvalues类似于mapvalues，不同的在于flatmapvalues应用于元素为kv对的rdd中value。每个一元素的value被输入函数映射为一系列的值，然后这些值再与原rdd中的key组成一系列新的kv对。

举例

scala> val a = sc.parallelize(list((1,2),(3,4),(3,6)))
scala> val b = a.flatmapvalues(x=>x.to(5))
scala> b.collect
res3: array[(int, int)] = array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))

上述例子中原rdd中每个元素的值被转换为一个序列（从其当前值到5），比如第一个kv对(1,2), 其值2被转换为2，3，4，5。然后其再与原kv对中key组成一系列新的kv对(1,2),(1,3),(1,4),(1,5)。

reduce

reduce将rdd中元素两两传递给输入函数，同时产生一个新的值，新产生的值与rdd中下一个元素再被传递给输入函数直到最后只有一个值为止。

举例

scala> val c = sc.parallelize(1 to 10)
scala> c.reduce((x, y) => x + y)
res4: int = 55

上述例子对rdd中的元素求和。

reducebykey

顾名思义，reducebykey就是对元素为kv对的rdd中key相同的元素的value进行reduce，因此，key相同的多个元素的值被reduce为一个值，然后与原rdd中的key组成一个新的kv对。

举例:

scala> val a = sc.parallelize(list((1,2),(3,4),(3,6)))
scala> a.reducebykey((x,y) => x + y).collect
res7: array[(int, int)] = array((1,2), (3,10))

上述例子中，对key相同的元素的值求和，因此key为3的两个元素被转为了(3,10)。

reference

本文中的部分例子来自:http://homepage.cs.latrobe.edu.au/zhe/zhenhesparkrddapiexamples.html

总结

以上就是本文关于浅谈spark rdd api中的map和reduce的全部内容，希望对大家有所帮助。感兴趣的朋友可以继续参阅本站：spark三种属性配置方式详解、浅谈七种常见的hadoop和spark项目案例等，有什么问题可以随时留言，小编会及时回复大家的。感谢朋友们对本站的支持！