什么是RDD？

RDD是Spark中的数据抽象，全称弹性分布式数据集（Resilient Distributed Datasets）。RDD可以理解为将一个大的数据集合以分布式的形式保存在集群服务器的内存中。RDD是一个容错的、并行的数据结构，可以让用户显式地将数据存储到磁盘和内存中，并能控制数据的分区。

RDD是Spark的核心，也是整个Spark的架构基础。

RDD的特点：

• 不可变性
  RDD是一种不可变的数据结构。一旦创建，就不能修改。
• 分片
  RDD表示的是一组数据的分区。这些分区分布在集群的多个节点上。RDD中存储了分区和数据物理分区之间关系的映射。
• 容错性
  RDD可以自动处理节点出现故障的情况。当某个节点出现故障时，该节点上存储的数据将无法被访问。Spark会在其他节点上重建丢失的RDD分区数据。RDD中存储了血统信息，通过血统信息，Spark可以恢复RDD的部分信息，当节点出现故障的时候，可以基于血统恢复整个RDD。
• 内存计算
  RDD类提供了一套支持内存计算的API。RDD可以在内存中缓存或长期驻留，从而提升了对RDD操作的效率。

RDD的5个主要属性：

  //只计算一次  
  protected def getPartitions: Array[Partition]  
  //对一个分片进行计算，得出一个可遍历的结果
  def compute(split: Partition, context: TaskContext): Iterator[T]
  //只计算一次，计算RDD对父RDD的依赖
  protected def getDependencies: Seq[Dependency[_]] = deps
  //可选的，分区的方法，针对第4点，类似于mapreduce当中的Paritioner接口，控制key分到哪个reduce
  @transient val partitioner: Option[Partitioner] = None
  //可选的，指定优先位置，输入参数是split分片，输出结果是一组优先的节点位置
  protected def getPreferredLocations(split: Partition): Seq[String] = Nil

1. 一组分片（Partition）。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。
2. 一个计算每个分区的函数。Spark中的RDD的计算是以分片为单位的，每个RDD都会实现compute函数来完成计算任务。
3. RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于刘水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，这样就不用对RDD的所有分区进行重新计算。源RDD没有依赖，通过依赖关系描述血统（lineage）。
4. 一个Partitioner（RDD的分片函数）。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于key-value的RDD，才会有Partitioner，非key-value的RDD的Partitioner的值是None。Partitioner函数决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
5. 一个存储每个Partition的优先位置的列表。对于一个HDFS文件来说，这个列表保存的就是每个partition所在的块的位置。Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置（移动数据不如移动计算）。

RDD的创建

可以通过两种方式创建RDD：

1. 由并行化集合创建
2. 由外部存储系统的数据集创建，包括本地的文件系统、Hadoop文件系统HDFS或从Hadoop接口API创建

// 由并行化集合创建
val list = (1 t0 100).toList
val rdd = sc.parallelize(list)

// 由外部存储系统创建
val rdd = sc.textFile("/path/data")
val rdd = sc.wholeTextFiles("/path/data")

RDD的操作

image

RDD转换操作

转换操作指的是在原RDD实例上进行计算，然后创建一个新的RDD实例。

RDD中的所有的转换操作都是惰性的，在执行RDD的转换操作的时候，并不会直接计算结果，而是记住这些应用到基础数据集上的转换动作，只有行动操作时，这些转换才会真正的去执行。这样设计的好处是更加有效率的运行。

惰性求值的好处：
Spark使用惰性求值，这样就可以把一些操作合并到一起来减少计算数据的步骤。

RDD行动操作

行动操作指的是向驱动器程序返回结果或把结果写入外部系统的操作。

Spark在调用RDD的行动操作的时候，会触发Spark中的连锁反应。当调用的行动操作的时候，Spark会尝试创建作为调用者的RDD。如果这个RDD是从文件中创建的，那么Spark会在worker节点上读取文件至内存中。如果这个RDD是通过其他RDD的转换得到的，Spark会尝试创建其父RDD。这个过程会一直持续下去，直到Spark找到根RDD。然后Spark就会真正执行这些生成RDD所必须的转换计算。最后完成行动操作，将结果返回给驱动程序或者写入外部存储。

img

RDD的缓存

Spark速度非常快的原因之一，就是在不同操作中在内存中持久化一个数据集。当持久化一个RDD后，每一个节点都将把计算的分片结果保存在内存中，并在对此数据集进行的其他动作中重用。这使得后续的动作变得更加迅速。缓存是Spark构建迭代算法和快速交互式查询的关键。所以我们在开发过程中，对经常使用的RDD要进行缓存操作，以提升程序运行效率。

RDD缓存的方法

RDD类提供了两种缓存方法：

rdd.cache()
rdd.persist()

cache方法其实是将RDD存储在集群中Worker的内存中。

persist是一个通用的cache方法。它可以将RDD存储在内存中或硬盘上或者二者皆有。

缓存的容错

缓存是有可能丢失（如机器宕机），或者存储于内存的数据由于内存不足而被删除。RDD的缓存的容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列的转换，丢失的数据会被重新计算。因为RDD的各个Partition是相对独立的，所以在重新计算的时候只需要计算丢失部分Partition即可，不需要重新计算全部的Partition。因此，在一个缓存RDD的节点出现故障的时候，Spark会在另外的节点上自动重新创建出现故障的节点中存储的分区。

RDD的检查点

RDD的缓存能够在第一次计算完成后，将计算结果保存到内存、本地文件系统或者Tachyon中。通过缓存，Spark避免了RDD上的重复计算，能够极大地提升计算速度。但是，如果缓存丢失了，则需要重新计算。如果计算特别复杂或者计算特别耗时，那么缓存丢失对于整个Job的影响是不容忽视的。为了避免缓存丢失重新计算带来的开销，所以Spark引入了检查点（checkpoint）机制。

缓存是在计算结束后，直接将计算结果通过用户定义的存储级别写入不同的介质。而检查点不同，它是在计算完成后，重新建立一个Job来计算。所以为了避免重复计算，推荐先将RDD缓存，这样在进行检查点操作时就可以快速完成。

RDD依赖

Spark会根据用户提交的计算逻辑中的RDD的转换和动作来生动RDD之间的依赖关系，同时这个计算链也就生成了逻辑上的DAG。

RDD之间的依赖关系包括：

1. 一个RDD是从哪些RDD转换而来，即RDD的parent RDD(s)是什么
2. 依赖于parent RDD(s)的哪些分区

Spark中的依赖关系主要体现为两种形式：

• 窄依赖：是指父RDD的每一个分区最多被一个子RDD的分区使用。
• 宽依赖：是指子RDD的每个分区都依赖于所有父RDD的所有所有分区或多个分区。

RDD的依赖关系

1. 窄依赖允许在单个集群节点上流水线式执行，这个节点可以计算所有父级分区。
2. 在窄依赖中，节点失败后的恢复更加高效。因为只有丢失的父级分区需要重新计算，并且这些丢失的父级分区可以并行地在不同节点上重新计算。但是在宽依赖的继承关系中，单个失败的节点可能导致一个RDD的所有先祖RDD中的一些分区丢失，导致计算的重新执行。