【赵渝强老师】Spark中的RDD

  RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，它是Spark中最基本、也是最重要的的数据模型。它由分区组成，每个分区被一个Spark的Worker从节点处理，从而支持分布式的并行计算。RDD通过检查点Checkpoint的方式提供自动容错的功能，并且具有位置感知性调度和可伸缩的特性。通过RDD也提供缓存的机制，可以极大地提高数据处理的速度。

  视频讲解如下：

一、RDD的组成

  在WordCount示例中，每一步都是生成一个新的RDD用于保存这一步的结果。创建RDD也可以使用下面的方式：

scala> val myrdd = sc.parallelize(Array(1,2,3,4,5,6,7,8),2)

  这行代码创建了一个名叫myrdd的RDD集合，该集合中包含了一个数组，并且这个RDD由2个分区组成。通过查看RDD的partitions算子可以查看分区的长度。

scala> myrdd.partitions.length
res0: Int = 2

  那么RDD、分区和Worker节点之间又是什么联系呢？这里以刚才创建的myrdd为例来说明它们之间的关系，如下图所示。

  图中假设有两个Worker的从节点。myrdd又包含了两个分区，每个分区会有一个分区号，分区号从零开始。从图9.9可以看出在第一个Worker上处理的分区0中的数据，即：{1,2,3,4}；而在第二个Worker处理的是分区1中的数据，即：{5,6,7,8}。

提示：这里可以把分区理解成是一个物理概念，它里面的数据由Worker上的Executor执行的任务处理。最外层的虚线方框表示的是RDD，可以看出它其实是一个逻辑概念。

二、RDD的特性

  在了解了RDD的基本概念后，那么RDD又具有什么样的特性呢？Spark RDD的源码中关于RDD的特性做了如下的解释。

 * Internally, each RDD is characterized by five main properties:
 *
 *  - A list of partitions
 *  - A function for computing each split
 *  - A list of dependencies on other RDDs
 *  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
 *  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

  通过这段注释可以了解到RDD具备以下5个基本的特性：

• 由一组分区（Partition）组成

  对于RDD来说，每个分区都会被一个计算任务处理并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU内核的数目。

• 一个计算每个分区的函数

  Spark中RDD的计算是以分区为单位。每个RDD都需要实现compute函数，从而达到处理数据的目的。

• RDD之间的依赖关系

  可以把WordCount程序代码拆开，从而单步执行。每一次转换时可以定义一个新的RDD来保存这一步的结果，如下所示。

scala> val rdd1 = sc.textFile("hdfs://bigdata111:9000/input/data.txt")
scala> val rdd2 = rdd1.flatMap(_.split(" "))
scala> val rdd3 = rdd2.map((_,1))
scala> val rdd4 = rdd3.reduceByKey(_+_)
scala> rdd4.saveAsTextFile("hdfs://bigdata111:9000/output/spark/wc")

  这里一共定义了4个RDD，分别是：rdd1、rdd2、rdd3和rdd4，其中：rdd4依赖rdd3，rdd3依赖rdd2，而rdd2依赖rdd1。根据依赖关系的不同，可以划分任务执行的阶段（Stage），从而支持检查点的容错机制。

提示：如果在计算过程中丢失了某个分区的数据，Spark可以通过这个依赖关系重新进行计算，而不是对RDD的所有分区进行重新计算。

• 一个Partitioner

  Partitioner是Spark RDD的分区函数。Spark内部实现了两种类型的分区函数：一种是基于哈希算法的HashPartitioner；另一种则是基于范围的RangePartitioner。通过继承Partitioner也可以实现自定义的分区函数。分区函数不但决定了RDD本身的分区数量，也决定了RDD Shuffle输出时的分区数量。

• 一个存储了读取每个分区优先位置（preferred location）的列表

  根据这个列表的信息，Spark在进行任务调度的时候会尽可能地将计算任务分配到其所要处理数据块的存储位置，这样可以提高处理数据的效率。