键值对操作 上（Spark自学五）

键值对RDD是Spark中许多操作所需要的常见数据类型。

“分区”是用来让我们控制键值对RDD在各节点上分布情况的高级特性。使用可控的分区方式把常在一起被访问的数据放在同一个节点上，可以大大减少应用的通信开销，带来明显的性能提升。

4.1 动机

　　Spark为pair RDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口。

4.2 创建Pair RDD

　　当需要把一个普通的RDD转为pair RDD时，可以调用map()函数来实现。下例为如何将由文本行组成的RDD转换为以每行的第一个单词为键的pair RDD。

　　例4-1： 在Python中使用第一个单词作为键创建出一个pair RDD

pairs = lines.map(lambda x: (x.split(" ")[0], x))

　　例4-2： 在Scala中使用第一个单词作为键创建出一个pair RDD

val pairs = lines.map(x => (x.split(" ")(0), x))

　　4.3 Pair RDD的转化操作

　　　　Pair RDD可以使用所有标准RDD上的可用的转化操作。

　　　　Pair RDD的转化操作（以键值对集合{（1，2），（3，4），（3，6）}）为例:

　　　　　　rdd.reduceByKey((x, y)=>x+y) 结果：{(1,2), (3,10)}

　　　　　　rdd.groupByKey()                    结果：{(1,[2]),(3,[4,6])}

　　　　　　rdd.mapValues(x => x+1)        结果：{(1,3),(3,5),(3,7)}

　　　　　　rdd.keys()                                结果：{1，3，3}

　　　　　　rdd.values()                             结果：{2,4,6}

　　　　　　rdd.sortByKey()                       结果：{（1，2），（3，4），（3，6）}

　　　　筛选掉长度超过20个字符的行，如下：

　　　　例4-4： 用Python对第二个元素进行筛选

result = pairs.filter(lambda keyValue: len(keyValue[1]<20))

　　　　例4-5： 用Scala对第二个元素进行筛选、

pairs.filter{case (key, value) => value.length < 20}

　　　　　　4.3.1 聚合操作

　　　　　　使用reduceByKey()和mapValues()来计算每个键的对应值的均值。

　　　　　　例4-7：在Python中使用reduceByKey()和mapValues()计算每个键对应的平均值

rdd.mapValues(lambda x:(x, 1)).reduceByKey(lambda x, y: (x[0]+y[0], x[1]+y[1]))

　　　　　　例4-8：在Python中使用reduceByKey()和mapValues()计算每个键对应的平均值

rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))

　　　　　　解决分布式单词计数问题如下

　　　　　　例4-9： 用Python实现单词计数

rdd = sc.textFile("s3://...")
words = rdd.flatMap(lambda x: x.split(" "))
result = words.map(lambda x: (x, 1)).reduceByKey(lambda x, y:x+y)

　　　　　　例4-10： 用Scala实现单词计数

val input = sc.textFile("s3://...")
val words = input.flatMap(x => x.split(" "))
val result = words.map(x => (x, 1)).reduceByKey((x, y) => x+y)

　　　　　　我们可以使用countByValue()函数，以更快地实现单词计数：input.flatMap(x => x.split(" ")).countByValue().

　　　　　　Spark怎样确定如何分割工作：每个RDD都有固定数目的分区，分区数决定了在RDD上执行操作时的并行度。

　　　　　　例4-15: 在Python中自定义reduceByKey()的并行度

data = [("a", 3), ("b", 4), ("a", 1)]
sc.parallelize(data).reduceByKey(lambda x, y: x+y, 10)

　　　　4.3.2 数据分组

 　　　　　　略

　　　　4.3.3 连接

　　　　　　连接数据操作：将有键的数据与另一组有键的数据一起使用。连接方式：右外连接、左外连接、交叉连接以及内连接。

　　　　　　例4-17：在Scala shell中进行内连接

storeAddress = {
    (Store("Ritual"), "AAA"), (Store("Philz"), "BBB"), 
    (Store("Philz"), "CCC"), (Store("Starbucks"), "DDD")}

storeRating = {
    (Store("Ritual"), 4.9), (Store("Philz"), 4.8)}

storeAddress.rightOuterJoin(storeRating) == {
    (Store("Ritual"), "AAA", 4.9)),
    (Store("Philz"), "BBB", 4.8)),
    (Store("Philz"), "CCC", 4.8))}

　　　　　　例4-18：leftOuterJoin()与rightOuterJoin()

storeAddress = {
    (Store("Ritual"), "AAA"), (Store("Philz"), "BBB"), 
    (Store("Philz"), "CCC"), (Store("Starbucks"), "DDD")}

storeRating = {
    (Store("Ritual"), 4.9), (Store("Philz"), 4.8)}

storeAddress.rightOuterJoin(storeRating) == {
    (Store("Ritual"), (Some("AAA"), 4.9)),
    (Store("Philz"), (Some("BBB"), 4.8)),
    (Store("Philz"), (Some("CCC"), 4.8))}

storeAddress.leftOuterJoin(storeRating) == {
    (Store("Ritual"), ("AAA", Some(4.9))),
    (Store("Starbucks"),("DDD",None)),     
    (Store("Philz"), ("BBB", Some(4.8))),
    (Store("Philz"), ("CCC", Some(4.8)))}

　　　　　4.3.4 数据排序

　　　　　　略