Spark学习—PairRDD编程

￡神魔★判官ぃ 2022-06-16 13:14 115阅读 0赞

**PairRDD**:键值对RDD,其是Spark中转化操作常用的数据类型。上一篇文件《》我们就使用了PairRDD。PairRDD是很多程序的构成要素，因为他提供了并行操作或跨节点重新进行数据分组的操作接口。本文将结合Java示例讲解PairRDD相关接口操作。

# 一. PairRDD创建 #

很多存储键值对的数据格式会在读取时直接返回其键值对数据组成的pair RDD，同时也可以调用map()函数来实现，传递的函数需要返回键值对。

对应Java创建PairRDD，由于Java没有自带的二元组类型，因此Spark的Java API中使用scala.Tuple2类来创建二元组，同时使用mapToPair()函数来代替map()函数。

比如上一篇文章的例子：[Spark学习—RDD编程][Spark_RDD]

# 二. PairRDD相关接口![Image 1][] #

PairRDD本质还是RDD，相对RDD的操作方法接口，PairRDD都支持。

## 1. PairRDD的转化操作 ##

<table>       
 <tbody> 
  <tr> 
   <td><strong>函数名</strong></td> 
   <td><strong>目的</strong></td> 
   <td><strong>示例</strong></td> 
   <td><strong>结果</strong></td> 
  </tr> 
  <tr> 
   <td>reduceByKey(func)</td> 
   <td>合并具有相同键的值</td> 
   <td>rdd.reduceByKey((x, y) =&gt; x + y)</td> 
   <td>{(1,2), (3,10)}</td> 
  </tr> 
  <tr> 
   <td>groupByKey()</td> 
   <td>对具有相同键的值进行分组</td> 
   <td>rdd.groupByKey()</td> 
   <td>{(1,[2]),(3, [4,6])}</td> 
  </tr> 
  <tr> 
   <td>combineByKey(<br> createCombiner,<br> mergeValue,<br> mergeCombiners,<br> partitioner)</td> 
   <td>使用不同的返回类型合并具有相同键的值</td> 
  </tr> 
  <tr> 
   <td>mapValues(func)</td> 
   <td>对pair RDD 中的每个值应用一个函数而不改变键</td> 
   <td>rdd.mapValues(x =&gt; x+1)</td> 
   <td>{(1,3), (3,5), (3,7)}</td> 
  </tr> 
  <tr> 
   <td>flatMapValues(func)</td> 
   <td>对pair RDD 中的每个值应用一个返回迭代器的函数，然后对返回的每个元素都生成一个对应原键的键值对记录通常用于符号化</td> 
   <td>rdd.flatMapValues(x =&gt; (x to 5))</td> 
   <td>{(1,2), (1,3), (1,4),<br> (1,5), (3,4), (3,5)}</td> 
  </tr> 
  <tr> 
   <td>keys()</td> 
   <td>返回一个仅包含键的RDD</td> 
   <td>rdd.keys()</td> 
   <td>{1, 3,3}</td> 
  </tr> 
  <tr> 
   <td>values()</td> 
   <td>返回一个仅包含值的RDD</td> 
   <td>rdd.values()</td> 
   <td>{2, 4,6}</td> 
  </tr> 
  <tr> 
   <td>sortByKey()</td> 
   <td>返回一个根据键排序的RDD</td> 
   <td>rdd.sortByKey()</td> 
   <td>{(1,2), (3,4), (3,6)}</td> 
  </tr> 
 </tbody> 
</table>

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

**combineByKey**:最为常用的基于键进行聚合的函数，和aggregate()一样，combineByKey()可以返回与输入数据的类型不一样的返回值。

**createCombiner**：组合器函数，用于将V类型转换成C类型，输入参数为RDD\[K,V\]中的V,输出为C

**mergeValue**：合并值函数，将一个C类型和一个V类型值合并成一个C类型，输入参数为(C,V)，输出为C

**mergeCombiners**：合并组合器函数，用于将两个C类型值合并成一个C类型，输入参数(C,C)，输出为C

**numPartitions**：结果RDD分区数，默认保持原有的分区数

**partitioner**：分区函数,默认为HashPartitioner

<table>       
 <tbody> 
  <tr> 
   <td><strong>函数名</strong></td> 
   <td><strong>目的</strong></td> 
   <td><strong>示例</strong></td> 
   <td><strong>结果</strong></td> 
  </tr> 
  <tr> 
   <td>subtractByKey</td> 
   <td>删掉RDD 中键与other RDD 中的键<br> 相同的元素</td> 
   <td>dd.subtractByKey(other)</td> 
   <td>{(1, 2)}</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>对两个RDD 进行内连接</td> 
   <td>rdd.join(other)</td> 
   <td>{(3, (4, 9)), (3,(6, 9))}</td> 
  </tr> 
  <tr> 
   <td>rightOuterJoin</td> 
   <td>对两个RDD 进行连接操作，<br> 确保第一个RDD 的键必须存在<br> （右外连接）</td> 
   <td>rdd.rightOuterJoin(other)</td> 
   <td>{(3,(Some(4),9)),<br> 3,(Some(6),9))}</td> 
  </tr> 
  <tr> 
   <td>leftOuterJoin</td> 
   <td>对两个RDD 进行连接操作，<br> 确保第二个RDD 的键必须存在<br> （左外连接）</td> 
   <td>rdd.leftOuterJoin(other)</td> 
   <td>{(1,(2,None)), <br> (3,(4,Some(9))),<br> &nbsp;(3,(6,Some(9)))}</td> 
  </tr> 
  <tr> 
   <td>cogroup</td> 
   <td>将两个RDD 中拥有相同键的<br> 数据分组到一起</td> 
   <td>rdd.cogroup(other)</td> 
   <td>{(1,([2],[])), (3,([4, 6],[9]))}</td> 
  </tr> 
 </tbody> 
</table>

针对两个pair RDD的转化操作（rdd = \{(1, 2), (3, 4), (3, 6)\}other = \{(3, 9)\}）

**连接方式**：左外连接、右外连接、交叉连接和内连接；

**join**:内连接，只有在两个pair RDD 中都存在的键才叫输出。当一个输入对应的某个键有多个值时，生成的pair RDD 会包括来自两个输入RDD 的每一组相对应的记录。

**leftOuterJoin**：在使用leftOuterJoin() 产生的pair RDD 中，源RDD 的每一个键都有对应的记录。每个键相应的值是由一个源RDD 中的值与一个包含第二个RDD 的值的Option（在Java 中为Optional）对象组成的二元组。

**rightOuterJoin**：rightOuterJoin() 几乎与leftOuterJoin() 完全一样，只不过预期结果中的键必须出现在第二个RDD 中，而二元组中的可缺失的部分则来自于源RDD 而非第二个RDD。

**cogroup**：对多个共享同一个键的RDD进行分组。对两个键的类型均为K 而值的类型分别为V 和W 的RDD 进行cogroup() 时，得到的结果RDD 类型为\[(K, (Iterable\[V\], Iterable\[W\]))\]。如果其中的一个RDD 对于另一个RDD 中存在的某个键没有对应的记录，那么对应的迭代器则为空。cogroup() 提供了为多个RDD 进行数据分组的方法。

## 2. PairRDD的行动操作 ##

<table>       
 <tbody> 
  <tr> 
   <td><strong>函数名</strong></td> 
   <td><strong>目的</strong></td> 
   <td><strong>示例</strong></td> 
   <td><strong>结果</strong></td> 
  </tr> 
  <tr> 
   <td>countByKey()</td> 
   <td>对每个键对应的元素分别计数</td> 
   <td>rdd.countByKey()</td> 
   <td>{(1, 1), (3, 2)}</td> 
  </tr> 
  <tr> 
   <td>collectAsMap()</td> 
   <td>将结果以映射表的形式返回，<br> 以便查询</td> 
   <td>rdd.collectAsMap()</td> 
   <td>Map{(1, 2), (3,4), (3, 6)}</td> 
  </tr> 
  <tr> 
   <td>lookup(key)</td> 
   <td>返回给定键对应的所有值</td> 
   <td>rdd.lookup(3)</td> 
   <td>[4, 6]</td> 
  </tr> 
 </tbody> 
</table>

Pair RDD的行动操作（以键值对集合\{(1, 2), (3, 4), (3, 6)\}为例）

# 三.示例 #

![1240][]

![Image 1][]

代码地址：[http://git.oschina.net/a123demi/sparklearning][http_git.oschina.net_a123demi_sparklearning]

[Spark_RDD]: http://blog.csdn.net/a123demi/article/details/71453594
[Image 1]: 
[1240]: /images/20220616/bcbacd878fa046cd842c179db6ba72a5.png
[http_git.oschina.net_a123demi_sparklearning]: http://git.oschina.net/a123demi/sparklearning