[Spark] Pair RDD常见转化操作

本篇博客中的操作都在 ./bin/pyspark 中执行。

对单个 Pair RDD 的转化操作

下面会对 Pair RDD 的一些转化操作进行解释。先假设我们有下面这些RDD（在pyspark中操作）：

nums = sc.parallelize( [ (1,2) ,(3,4) ,(3,6) ] )
x = sc.parallelize( [ (1,[2,4,5]) ,(4,[7,8,0]) ,(4,[6,7,5])] )


reduceByKey

概述：合并具有相同键值的值。

例子：

>>> nums.reduceByKey(lambda x, y : x + y).collect()
[(1, 2), (3, 10)]
>>>
>>> x.reduceByKey(lambda x, y: x + y).collect()
[(1, [2, 4, 5]), (4, [7, 8, 0, 6, 7, 5])]


这个方法操作的是值(Values)，对上面的两个RDD的操作，第一个是对值做加法，第二个是对列表合并；这两个操作都可以使用lambda x, y : x + y来完成。

再来一个例子，求平均值，（下面的这个RDD的键值中，第一个值是总和，第二个值是数量）：

>>> test = sc.parallelize([('panda', (1,2)), ('pink',(7,2)), ('pirate',(3,1))])
>>> test.mapValues(lambda (x,y): x / (y* 1.0)).collect()
[('panda', 0.5), ('pink', 3.5), ('pirate', 3.0)]


groupByKey

groupByKey 方法的目的是对具有相同键值的数据进行分组，比如说：

>>> l = nums.groupByKey().collect()[1][1]
>>> l
<pyspark.resultiterable.ResultIterable object at 0x109320f10>
>>> for i in l:
... print i
...
4
6


直观地来说，对nums这个RDD的groupByKey操作可以表示为：

[(1,2),(3,4),(3,6)] -> [ (1,[2]), (3, [4,6] )]


然后是对于x这个RDD的：

>>> x = sc.parallelize( [ (1,[2,4,5]) ,(4,[7,8,0]) ,(4,[6,7,5])] )>>> l = x.groupByKey().collect()
>>> l
[(1, <pyspark.resultiterable.ResultIterable object at 0x109310690>), (4, <pyspark.resultiterable.ResultIterable object at 0x109310050>)]>>> l2 = l[1][1]
>>> l2
<pyspark.resultiterable.ResultIterable object at 0x109310050>>>> for i in l2:
... print i
...
[7, 8, 0]
[6, 7, 5]


直观的来说：

[ (1,[2,4,5]), (4,[7,8,0]) ,(4,[6,7,5] ) ]
+
| +-------------+
| | RDD.join |
| +-------------+
v
[ (1,[2,4,5]), (4, [ [6,7,5], [7,8,0] ] ) ]


mapValues

这个比较好理解，对每个键值进行操作：

>>> nums.mapValues(lambda x : x+ 3).collect()
[(1, 5), (3, 7), (3, 9)]


flatMapValues

这个方法的作用是对pair RDD 的每个值(values)生成一个与原键(key)对应的键值对记录。

x = sc.parallelize( [ (1,[2,4,5]) ,(4,[7,8,0]) ,(4,[6,7,5])] )>>> def f(x):
... return x
...>>> x.flatMapValues(f).collect()
[(1, 2), (1, 4), (1, 5), (4, 7), (4, 8), (4, 0), (4, 6), (4, 7), (4, 5)]


这个可以用"flat"这个英文单词的意思来大致理解一下，flat有使变平，拍扁的意思。

对于 nums 这种RDD是进行不了这个方法的。

keys()

返回所有的键值：

>>> nums.keys().collect()
[1, 3, 3]
>>> x.keys().collect()
[1, 4, 4]


values

返回所有的值：

>>> nums.values().collect()
[2, 4, 6]
>>> x.values().collect()
[[2, 4, 5], [7, 8, 0], [6, 7, 5]]


sortByKey

按照键值排序，这个比较好理解：

>>> notSorted = sc.parallelize( [ (7,[2,4,5]) ,(9,[7,8,0]) ,(4,[6,7,5])] )>>> notSorted.sortByKey().collect()
[(4, [6, 7, 5]), (7, [2, 4, 5]), (9, [7, 8, 0])]


对2个Pair RDD的转化操作

这里我们有：

other = sc.parallelize([(3,9)])
nums = sc.parallelize([(1,2),(3,4),(3,6)])


subtractByKey

返回一个减去两个RDD中一样的Key的RDD，可以理解为除去下图中重合的部分：

一个例子：

>>> nums.subtractByKey(other).collect()
[(1, 2)]


join

这个操的作描为：返回一个RDD，返回的RDD只包含输入的两个RDD都包含的键值，每个键值对形如(k, (v1, v2))，其中v1被自己包含，v2被另一个RDD包含。一个例子：

>>> x = sc.parallelize([("a", 1), ("b", 4)])
>>> y = sc.parallelize([("a", 2), ("a", 3)])
>>> x.join(y).collect()
[('a', (1, 2)), ('a', (1, 3))]


可以理解为：

同理换做我们开始提到的两个RDD：

>>> nums.join(other).collect()
[(3, (4, 9)), (3, (6, 9))]


rightOuterJoin && leftOuterJoin

这两个方法的Join的原理和上面的join一样，关于left 和 right 的说明是：

• right：确保右边的RDD的键必须存在
• left：确保左边的RDD的键必须存在

一个例子：

>>> nums.rightOuterJoin(other).collect()
[(3, (4, 9)), (3, (6, 9))]>>> nums.leftOuterJoin(other).collect()
[(1, (2, None)), (3, (4, 9)), (3, (6, 9))]