Spark-RDD简单介绍

缺乏、安全感 2022-04-12 21:20 223阅读 0赞

结合Spark官网, 对Spark RDD的一些简单介绍和总结.

RDD是Spark提供的主要抽象, 全称弹性分布式数据集, 它是跨集群节点来分区的元素集合, 可以并行操作, 可以保留在内存, 还可以自动从节点故障中恢复.

## 创建RDD ##

创建RDD有两种方法

*  并行化现有的集合
    
        val data = Array(1, 2, 3, 4, 5)
        val rdd = sc.parallelize(data)
        val rdd2 = sc.parallelize(data, 10)
    
    并行集合的一个重要参数就是将数据集切分的分区数. Spark执行任务时, 为每一个分区产生一个task, 分区数也就是任务执行时的并行度, 所以可以通过第二个参数来手动设置分区数.
 *  引用外部存储系统中的数据集  
    Spark可以从Hadoop支持的任何存储系统创建RDD, 包括本地文件系统, HDFS, HBase等等.
    
        val rdd = sc.textFile("data.text")
    
    关于读取文件的注意事项
    
        - 如果使用的是本地文件系统路径, 要确保该文件已发送到所有worker节点上的相同路径下.
          - 文件的URI支持使用通配符, 如textFile("/my/directory/*.txt")
          - 该方法同样有第二个可选参数来控制分区数, 默认Spark为文件的每个块创建一个分区(HDFS中默认一块128MB), 你只能创建比现有块更多的分区, 不能更少.

## RDD操作 ##

RDD的操作分两种类型: transformation(转换, 从现有数据集创建新的数据集)和action(行动, 在数据集上运行计算后将值返回给driver端).  
Spark中所有的转换都是懒惰的, 所以转换操作并不会触发Spark job的提交, 只有触发action时, 才会提交job运算结果.

常见的转换操作

<table> 
 <thead> 
  <tr> 
   <th align="left">名称</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong>map</strong>(func)</td> 
   <td align="left">将RDD中每个元素一一转换成新元素返回新数据集</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>filter</strong>(func)</td> 
   <td align="left">返回func为true的元素形成的新数据集</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>flatMap</strong>(func)</td> 
   <td align="left">将RDD中的每个元素进行一对多转换形成新的数据集</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>union</strong>(otherDataset)</td> 
   <td align="left">将两个集合中的数据进行合并, 返回两个集合的并集, 不去重</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>join</strong>(otherDataset, [numPartitions])</td> 
   <td align="left">当调用类型(K, V)和(K, W)的数据集时, 返回(K, (V, W))对的数据集以及每个键的所有元素对</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>groupByKey</strong>([numPartitions])</td> 
   <td align="left">在(K, V)对的数据集上调用, 返回(K, Iterable)对的数据集;默认输出的并行度取决于父RDD的分区数, 也可以使用numPartitions参数指定</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>reduceByKey</strong>(func, [numPartitions])</td> 
   <td align="left">当调用(K, V)对的数据集时, 返回(K, V)对数据集, 使用给定的reduce函数func聚合每个键的值, 同样可以通过numPartitions参数指定任务数量</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>sortByKey</strong>([ascending], [numpartitions])</td> 
   <td align="left">返回按Key升序或降序的(K, V)对的数据集</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>repartition</strong>(numPartitions)</td> 
   <td align="left">随机重新调整RDD中的数据以创建更平衡的分区</td> 
  </tr> 
 </tbody> 
</table>

常见行动操作

<table> 
 <thead> 
  <tr> 
   <th align="left">名称</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong>reduce</strong>(func)</td> 
   <td align="left">使用func来聚合数据集的元素</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>collect</strong>()</td> 
   <td align="left">在driver端将数据集所有元素作为数组返回, 注意当结果集很大时十分消耗内存</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>count</strong>()</td> 
   <td align="left">返回数据集中的元素数</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>first</strong>()</td> 
   <td align="left">返回数据集中的第一个元素</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>take</strong>(n)</td> 
   <td align="left">返回数据集中的前n个元素的数组</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>saveAsTextFile</strong>(path)</td> 
   <td align="left">将数据集的元素作为文本文件写入Hadoop支持的文件系统的指定目录中</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>foreach</strong>(func)</td> 
   <td align="left">在每个元素上运行func</td> 
  </tr> 
 </tbody> 
</table>

## Shuffle ##

在Spark中, 单个任务在单个分区上运行, 为了组织执行单个reduce任务的所有数据, 就必须从所有分区中读取所有键的所有值, 然后将各个值组合在一起以计算每个键的结果, 这就是Shuffle.

一般触发shuffle的操作包括重新分区, 如repartition和coalesce; ByKey操作, 如groupByKey和reduceByKey;连接操作, 如join和cogroup.

shuffle操作消耗巨大, 因为它涉及到磁盘I/O, 数据序列化和网络I/O.为了组织shuffle的数据, Spark生成多组map任务以组织数据, 以及一组reduce任务来聚合数据.map任务的结果会保留在内存中, 直到内存放不下, Spark会将这些数据溢出到磁盘, 从而导致磁盘I/O的额外开销和垃圾回收的增加.  
Shuffle还会在磁盘上生成大量中间文件, 从Spark1.3开始, 这些文件直到相关RDD不再使用才会被垃圾回收, 这样做是为了在重新计算时, 不需要重新创建shuffle文件.spark.local.dir可配置临时存储目录.

## RDD持久性 ##

Spark中最重要的功能之一就是跨操作在内存中持久化数据集.当你缓存RDD时, 每个节点都会将它计算的分区数据存储在内存中, 并在该数据集的其他操作中重用它们, 这使得后续操作执行更快.缓存是迭代算法和快速交互式使用的关键工具.

你可以使用persist()或cache()方法标记要缓存的RDD.第一次计算它时, 会将RDD保留在节点内存中.

此外, 每个持久化的RDD都可以使用不同的存储级别进行存储.例如, 存在内存里, 存在磁盘上, 序列化为Java对象等.persist()通过传递StorageLevel对象来设置级别, cache()方法默认实用StorageLevel.MEMORY\_ONLY.  
全部存储级别有:

<table> 
 <thead> 
  <tr> 
   <th align="left">存储级别</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">MEMORY_ONLY</td> 
   <td align="left">默认级别,将RDD存储为JVM中的反序列化Java对象, 如果内存不够将不会被缓存.</td> 
  </tr> 
  <tr> 
   <td align="left">MEMORY_AND_DISK</td> 
   <td align="left">将RDD存储为JVM中的反序列化Java对象, 如果内存不够将溢出到磁盘.</td> 
  </tr> 
  <tr> 
   <td align="left">MEMORY_ONLY_SER(Java和Scala)</td> 
   <td align="left">将RDD存储为序列化Java对象, 这通常比反序列化对象更节省空间, 但是读取CPU密集程度更高.</td> 
  </tr> 
  <tr> 
   <td align="left">MEMORY_AND_DISK_SER(Java和Scala)</td> 
   <td align="left">与MEMORY_ONLY_SER类似, 但将不适合在内存的分区溢出到磁盘.</td> 
  </tr> 
  <tr> 
   <td align="left">DISK_ONLY</td> 
   <td align="left">仅将RDD存储在磁盘上</td> 
  </tr> 
  <tr> 
   <td align="left">MEMORY_ONLY_2,MEMORY_AND_DISK_2等</td> 
   <td align="left">与以上级别相同, 但在集群的两个节点上复制</td> 
  </tr> 
  <tr> 
   <td align="left">OFF_HEAP(实验性)</td> 
   <td align="left">与MEMORY_ONLY_SER类似, 但将数据存储在堆外内存, 这需要启用堆外内存.</td> 
  </tr> 
 </tbody> 
</table>