Spark Pair RDD键值对操作

￡神魔★判官ぃ 2022-04-12 05:08 262阅读 0赞

# Spark Pair RDD键值对操作 #

1 PairRDD简介

2 创建Pair RDD

3 Pair RDD的转化操作

3.1 聚合操作

对人工智能感兴趣的同学，可以点击以下链接：

[现在人工智能非常火爆，很多朋友都想学，但是一般的教程都是为博硕生准备的，太难看懂了。最近发现了一个非常适合小白入门的教程，不仅通俗易懂而且还很风趣幽默。所以忍不住分享一下给大家。点这里可以跳转到教程。][Link 1]

[https://www.cbedai.net/u014646662][Link 1]

## 1 PairRDD简介 ##

Spark 为包含键值对类型的RDD 提供了一些专有的操作。这些RDD 被称为pair RDD1。PairRDD 是很多程序的构成要素，因为它们提供了并行操作各个键或跨节点重新进行数据分组的操作接口。例如，pair RDD 提供reduceByKey() 方法，可以分别归约每个键对应的数据，还有join() 方法，可以把两个RDD 中键相同的元素组合到一起，合并为一个RDD。我们通常从一个RDD 中提取某些字段（例如代表事件时间、用户ID 或者其他标识符的字段），并使用这些字段作为pair RDD 操作中的键。

## 2 创建Pair RDD ##

在Spark 中有很多种创建pair RDD 的方式，很多存储键值对的数据格式会在读取时直接返回由其键值对数据组成的pair RDD。此外，当需要把一个普通的RDD 转为pair RDD 时，可以调用map() 函数来实现，传递的函数需要返回键值对。后面会展示如何将由文本行组成的RDD 转换为以每行的第一个单词为键的pair RDD。

在Scala 中，为了让提取键之后的数据能够在函数中使用，需要返回二元组。隐式转换可以让二元组RDD 支持附加的键值对函数。

scala> var lines = sc.textFile("passwd")
    lines: org.apache.spark.rdd.RDD[String] = passwd MapPartitionsRDD[1] at textFile at <console>:27
    
    scala> var pair = lines.map(x => (x.split(":")(0),x))
    pair: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[2] at map at <console>:29
    
    scala> pair.take(3).foreach(println)
    (root,root:x:0:0:root:/root:/bin/bash)
    (bin,bin:x:1:1:bin:/bin:/sbin/nologin)
    (daemon,daemon:x:2:2:daemon:/sbin:/sbin/nologin)

当用Scala 从一个内存中的数据集创建pair RDD 时，只需要对这个由二元组组成的集合调用SparkContext.parallelize() 方法。

## 3 Pair RDD的转化操作 ##

Pair RDD 可以使用所有标准RDD 上的可用的转化操作。由于pair RDD 中包含二元组，所以需要传递的函数应当操作二元组而不是独立的元素。

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

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

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

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

scala> var lines = sc.textFile("passwd")
    lines: org.apache.spark.rdd.RDD[String] = passwd MapPartitionsRDD[6] at textFile at <console>:27
    
    scala> var pair = lines.map(x => (x.split(":")(0),x.split(":")(6)))
    pair: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[7] at map at <console>:29
    
    scala> pair.take(3).foreach(println)
    (root,/bin/bash)
    (bin,/sbin/nologin)
    (daemon,/sbin/nologin)
    
    scala> var bash = pair.filter(u => u._2.contains("bash"))
    bash: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[8] at filter at <console>:31
    
    scala> bash.take(3).foreach(println)
    (root,/bin/bash)
    (hdfs,/bin/bash)
    (yarn,/bin/bash)

### 3.1 聚合操作 ###

当数据集以键值对形式组织的时候，聚合具有相同键的元素进行一些统计是很常见的操作。之前讲解过基础RDD 上的fold()、combine()、reduce() 等行动操作，pair RDD 上则有相应的针对键的转化操作。Spark 有一组类似的操作，可以组合具有相同键的值。这些操作返回RDD，因此它们是转化操作而不是行动操作。  
reduceByKey() 与reduce() 相当类似；它们都接收一个函数，并使用该函数对值进行合并。reduceByKey() 会为数据集中的每个键进行并行的归约操作，每个归约操作会将键相同的值合并起来。因为数据集中可能有大量的键，所以reduceByKey() 没有被实现为向用户程序返回一个值的行动操作。实际上，它会返回一个由各键和对应键归约出来的结果值组成的新的RDD。  
foldByKey() 则与fold() 相当类似；它们都使用一个与RDD 和合并函数中的数据类型相同的零值作为初始值。与fold() 一样，foldByKey() 操作所使用的合并函数对零值与另一个元素进行合并，结果仍为该元素。

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

scala> val rdd = sc.parallelize(List(("panda",0),("pink",3),("pirate",3),("panda",1),("pink",4)))
    rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[9] at parallelize at <console>:27
    
    
    scala> val kv = rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).map(v => (v._1,v._2._1/v._2._2.toDouble))
    kv: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[17] at map at <console>:29
    
    scala> kv.collect().foreach(println)
    (panda,0.5)
    (pink,3.5)
    (pirate,3.0)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2NDY2NjI_size_16_color_FFFFFF_t_70][]

熟悉MapReduce 中的合并器（combiner）概念的读者可能已经注意到，调用reduceByKey() 和foldByKey() 会在为每个键计算全局的总结果之前先自动在每台机器上进行本地合并。用户不需要指定合并器。更泛化的combineByKey() 接口可以让你自定义合并的行为。

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

<table> 
 <tbody> 
  <tr> 
   <td style="text-align:center;">函数</td> 
   <td style="text-align:center;">描述</td> 
   <td style="text-align:center;">示例</td> 
   <td style="text-align:center;">结果</td> 
  </tr> 
  <tr> 
   <td>countByKey()</td> 
   <td>对每个键对应的元素分别计数</td> 
   <td>rdd.countByKey()</td> 
   <td>{(1,1),(3,2)}</td> 
  </tr> 
  <tr> 
   <td>collectAsMap()</td> 
   <td>将结果以映射表的形式返回，以便查询</td> 
   <td>rdd.collectAsMap()</td> 
   <td>Mqp{(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>

[Link 1]: https://www.cbedai.net/u014646662
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2NDY2NjI_size_16_color_FFFFFF_t_70]: /images/20220412/cad93582deed4e1bbd2bfa19ae39246f.png