齐齐哈尔天气：spark中的pair rdd，看这一篇就够了

今天是spark专题的第四篇文章，我们一起来看下Pair RDD。

界说

在之前的文章当中，我们已经熟悉了RDD的相关观点，也领会了RDD基本的转化操作和行动操作。今天我们来看一下RDD当中异常常见的PairRDD，也叫做键值对RDD，可以明了成KVRDD。

KV很好明了，就是key和value的组合，好比Python当中的dict或者是C 以及Java当中的map中的基本元素都是键值对。相比于之前基本的RDD，pariRDD可以支持更多的操作，相对来说加倍天真，可以完成加倍庞大的功效。好比我们可以凭据key举行聚合，或者是盘算交集等。

以是自己pairRDD只不过是数据类型是KV结构的RDD而已，并没有太多的内在，人人不需要忧郁。

Pair RDD转化操作

Pair RDD也是RDD，以是之前先容的RDD的转化操作Pair RDD自然也可以使用。它们两者有些像是类继续的关系，RDD是父类，Pair RDD是实现了一些新特征的子类。子类可以挪用父类当中所有的方式，然则父类却不能挪用子类中的方式。

挪用的时刻需要注重，由于我们的Pair RDD中的数据格式是KV的二元组，以是我们传入的函数必须是针对二元组数据的，否则的话可能运算的效果会有问题。下面我们来枚举一些最常用的转化操作。

为了利便演示，我们用一个牢固的RDD来运行种种转化操作，来直观领会一下这些转化操作事实起什么样的作用。

ex1 = sc.parallelize([[1, 2], [3, 4], [3, 5]])

keys，values和sortByKey

这三个转化操作应该是最常用也是最简朴的，简朴到我们通过字面意思就可以猜出它们的意思。

我们先来看keys和values：

我们的RDD当中二元组当中的第一个元素会被当做key，第二个元素当做value，需要注重的是，它并不是一个map或者是dict，以是key和value都是可以重复的。

sortByKey也很直观，我们从字面意思就看得出来是对RDD当中的数据凭据key值举行排序，同样，我们也来看下效果：

mapValues和flatMapValues

mapValues不能直接使用，而必须要传入一个函数作为参数。它的意思是对所有的value执行这个函数，好比我们想把所有的value所有转变成字符串，我们可以这么操作：

flatMapValues的操作和我们的认知有些相反，我们都知道flatMap操作是可以将一个嵌套的数组打散，然则我们怎么对一个value打散嵌套呢？究竟我们的value不一定就是一个数组，这就要说到我们传入的函数了，这个flatMap的操作实在是针对函数返回的效果的，也就是说函数会返回一个迭代器，然后打散的内容实在是这个迭代器当中的值。

我这么表述可能有些死板，我们来看一个例子就明了了：

不知道这个效果有没有出乎人人的意料，它的整个流程是这样的，我们挪用flatMapValues运算之后返回一个迭代器，迭代器的内容是range(x, x 3)。实在是每一个key对应一个这样的迭代器，之后再将迭代器当中的内容打散，和key组成新的pair。

groupByKey，reduceByKey和foldByKey

这两个功效也对照靠近，我们先说第一个，若是学过SQL的同砚对于group by操作的寄义应该异常熟悉。若是没有领会过也没有关系，group by可以简朴明了成合并或者是分桶。也就是说将key值相同的value合并到一起，获得的效果是key-list的Pair RDD，也就是我们把key值相同的value放在了一个list当中。

我们也来看下例子：

我们挪用完groupby之后获得的效果是一个工具，以是需要挪用一下mapValues将它转成list才可以使用，否则的话是不能使用collect获取的。

reduceByKey和groupByKey类似，只不过groupByKey只是合并到一起，然而reduceByKey是传入reduce函数，执行reduce之后的效果。我们来看一个例子：

在这个例子当中我们执行了累加，把key值相同的value加在了一起。

foldByKey和fold的用法差异并不大，唯一差别的是我们加上了凭据key值聚合的逻辑。若是我们把分区的初始值设置成0的话，那么它用起来和reduceByKey几乎没有区别：

我们只需要清晰foldByKey当中的初始值针对的是分区即可。

combineByKey

这个也是一个很焦点而且不太容易明了的转化操作，我们先来看它的参数，它一共接受5个参数。我们一个一个来说，首先是第一个参数，是createCombiner。

它的作用是初始化，将value凭据我们的需要做初始化，好比将string类型的转化成int，或者是其他的操作。我们用记号可以写成是V => C，这里的V就是value，C是我们初始化之后的新值。

它会和value一起被当成新的pair传入第二个函数，以是第二个函数的接受参数是(C, V)的二元组。我们要做的是界说这个二元组的合并，以是第二个函数可以写成(C, V) => C。源码里的注释和网上的教程都是这么写的，但我以为由于泛起了两个C，可能会让人难以明了，我以为可以写成(C, V) => D，对照好。

最后一个函数是将D举行合并，以是它可以写成是(D, D) => D。

到这里我们看似似乎明了了它的原理，然则又似乎有许多问号，总以为那里有些不太对劲。我想了良久，才找到了问题的泉源，出在那里呢，在于合并。有没有发现第二个函数和第三个函数都是用来合并的，为什么我们要合并两次，它们之间的区别是什么？若是这个问题没搞明了，那么对于它的使用一定是错误的，我小我私家以为这个问题才是这个转化操作的焦点，没讲清晰这个问题的博客都是不够清晰的。

实在这两次合并的逻辑大同小异，然则合并的局限不一样，第一次合并是针对分区的，第二次合并是针对key的。由于在spark当中数据可能不止存放在一个分区内，以是我们要合并两次，第一次先将分区内部的数据整合在一起，第二次再跨分区合并。由于差别分区的数据可能相隔很远，以是会导致网络传输的时间过长，以是我们希望传输的数据只管小，这才有了groupby两次的缘故原由。

我们再来看一个例子：

在这个例子当中我们盘算了每个单词泛起的平均个数，我们一点一点来看。首先，我们第一个函数将value转化成了(1, value)的元组，元组的第0号元素示意泛起该单词的文档数，第1号元素示意文档内泛起的次数。以是第二个函数，也就是在分组内聚合的函数，我们对于泛起的文档数只需要加一即可，对于泛起的次数要举行累加。由于这一次聚合的工具都是(1, value)类型的元素，也就是没有聚合之前的效果。

在第三个函数当中，我们对于泛起的总数也举行累加，是由于这一个函数处置的效果是各个分区已经聚合一次的效果了。好比apple在一个分区内泛起在了两个文档内，一共泛起了20次，在一个分区泛起在了三个文档中，一共泛起了30次，那么显然我们一共泛起在了5个文档中，一共泛起了50次。

由于我们要盘算平均，以是我们要用泛起的总次数除以泛起的文档数。最后经由map之后由于我们获得的照样一个二元组，我们不能直接collect，需要用collectAsMap。

我们把上面这个例子用图来展示，会很容易明了：

毗邻操作

在spark当中，除了基础的转化操作之外，spark还提供了分外的毗邻操作给pair RDD。通过毗邻，我们可以很利便地像是操作聚集一样操作RDD。操作的方式也异常简朴，和SQL当中操作数据表的形式很像，就是join操作。join操作又可以分为join(inner join)、left join和right join。

若是你熟悉SQL的话，想必这三者的区别应该异常清晰，它和SQL当中的join是一样的。若是不熟悉也没有关系，注释起来并不庞大。在join的时刻我们往往是用一张表去join另外一张表，就似乎两个数相减，我们用一个数减去另外一个数一样。好比A.join(B)，我们把A叫做左表，B叫做右表。所谓的join，就是把两张表当中某一个字段或者是某些字段值相同的行毗邻在一起。

好比一张表是学生表，一张表是出勤表。我们两张表用学生的id一关联，就获得了学生的出勤纪录。然则既然是聚集关联，就会泛起数据关联不上的情形。好比某个学生没有出勤，或者是出勤内外记错了学生id。对于数据关联不上的情形，我们的处置方式有四种。第一种是全都抛弃，关联不上的数据就不要了。第二种是所有保留，关联不上的字段就记为NULL。第三种是左表关联不上的保留，右表抛弃。第四种是右表保留，左表抛弃。

下图展示了这四种join，异常形象。

我们看几个现实的例子来体会一下。

首先建立数据集：

ex1 = sc.parallelize([['frank', 30], ['bob', 9], ['silly', 3]])
ex2 = sc.parallelize([['frank', 80], ['bob', 12], ['marry', 22], ['frank', 21], ['bob', 22]])

接着，我们划分运行这四种join，考察一下join之后的效果。

从效果当中我们可以看到，若是两个数据集当中都存在多条key值相同的数据，spark会将它们两两相乘匹配在一起。

行动操作

最后，我们看下pair RDD的行动操作。pair RDD同样是rdd，以是通俗rdd适用的行动操作，同样适用于pair rdd。然则除此之外，spark还为它开发了独占的行动操作。

countByKey

countByKey这个操作顾名思义就是凭据Key值盘算每个Key值泛起的条数，它等价于count groupby的SQL语句。我们来看个详细的例子：

collectAsMap

这个也很好明了，实在就是讲最后的效果以map的形式输出：

从返回的效果可以看到，输出的是一个dict类型。也就是Python当中的"map"。

lookup

这个单词看起来对照少见，实在它代表的是凭据key值查找对应的value的意思。也就是常用的get函数，我们传入一个key值，会自动返回key值对应的所有的value。若是有多个value，则会返回list。

总结

到这里，所有的pair RDD相关的操作就算是先容完了。pair rdd在我们一样平常的使用当中泛起的频率异常高，行使它可以异常利便地实现一些对照庞大的操作。

另外，今天的这篇文章内容不少，想要完全吃透，需要一点功夫。这不是看一篇文章就可以实现的，然则也没有关系，我们初学的时刻只需要对这些api和使用方式有一个也许的印象即可，详细的使用细节可以等用到的时刻再去查阅相关的资料。

今天的文章就是这些，若是以为有所收获，请随手点个关注或者转发吧，你们的举手之劳对我来说很主要。