转载请注明 : [过把火] https://www.jianshu.com/p/29d17aa23116

序

一直都没有很系统地阅读过RDD的原始论文，最近翻出来研读一遍，并作此记录。

《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》

总

阅读完之后，唯一的感觉就是—RDD（弹性分布式数据集）完全是伯克利的博士起的一个很抽象的名字罢了，换句话说就是为了发论文而起的一个高大上的名字。但是这并不妨碍RDD的思想依然是diaodiao的。

动机

当前很多分布式计算框架无法实现高效的迭代式计算以及交互式数据挖掘，包括Hadoop！，首先为了解决高效这个问题，RDD提出基于内存的迭代思想，直接鄙视了Hadoop要不断进行磁盘Spill的弊端；其次，为了保证大数据场景下迭代计算的正常运转，RDD自身具有高容错快恢复的特点。

背景及意义

1、Hadoop？
Hadoop为分布式大规模数据的计算而生，但别忘了，Hadoop依托于HDFS，在没有与Spark进行对比之前，可能并不会刻意去思考为何需要HDFS，可能这个问题很容易回答，不就是一个存储仓库嘛，ok，就先这样认为。
2、RDD与Spark的关系？
Spark就是RDD的具体实现。
3、Spark VS Hadoop
相同数据集的重复利用的这种特性在很多领域或算法中常见，例如机器学习，就是对同一数据集不断进行收敛或是梯度下降，不管什么方式，对象都是同一数据分片。那么这种业务丢给Hadoop-MR来做的话，无非就是每计算完一次，Spill到HDFS进行多副本存储，单就这个多副本存储来讲，数据量一大，IO及磁盘的负担将会成为整个业务流程的瓶颈，换句话说，做一次PageRank，加入设定10次收敛，那么MR整体会写9次，这还不算，写完后还得读出来继续迭代，我的天，互联网发展到如今地步，数不清的网页，数不清的图节点，照这样进行rank，可想而知背后需要多好的磁盘性能来做保障，当然，运维人员有饭吃那是肯定的。
OK，RDD概念一出，使得这些业务逻辑无需每次都得先保存后计算，而是将数据显式地存储在内存中进行迭代，同时允许用户控制数据分区。这些还都不是最大的特点，RDD最大的特点是能够通过记录数据集的一些列转换方式来执行这些task，这样一来，某一分片若是丢失，则可以从该RDD的记录中去就近恢复该分片，而不是从头执行！最后一点，RDD中所记录的这些所谓的转换方式其实就是该RDD的诞生方式，也称作是“血缘”。因此，这种方式在不用大规模保存副本的同时也能够有很好地容错表现。

RDD—Resilient Distributed Datasets

1 RDD概念

1、RDD是一个只读的、有分区的分布式数据集。其分类主要有两种：transformations和action。这两种RDD负责不同的业务。transformations负责数据分片的转换，而action负责激活整个计算链条的实际计算。
2、RDD运转方式
RDD只需知道自己是怎么诞生的就可以了，这就是RDD的实际工作方式。

2 RDD与传统DSM对比

RDD作为对内存的抽象，与其相类似的就是分布式共享内存年系统（DSM）。

RDDs 只能通过粗粒度的转换被创建（或者被写） , 然而 DSM 允许对每一个内存位置进行读写, 这个是 RDDs 和 DSM 最主要的区别. 这样使都 RDDs在 应用中大量写数据受到了限制, 但是可以使的容错变的更加高效. 特别是, RDDs 不需要发生非常耗时的 checkpoint 操作, 因为它可以根据 lineage 进行恢复数据 . 而且, 只有丢掉了数据的分区才会需要重新计算, 并不需要回滚整个程序, 并且这些重新计算的任务是在多台机器上并行运算的.

RDDs 的第二个好处是：它不变的特性使的它可以和 MapReduce 一样来运行执行很慢任务的备份任务来达到缓解计算很慢的节点的问题. 在 DSM 中, 备份任务是很难实现的, 因为原始任务和备份任务或同时更新访问同一个内存地址和接口.

最后, RDDs 比 DSM 多提供了两个好处. 第一, 在对 RDDs 进行大量写操作的过程中, 我们可以根据数据的本地性来调度 task 以提高性能. 第二, 如果在 scan-base 的操作中, 且这个时候内存不足以存储这个 RDDs, 那么 RDDs 可以慢慢的从内存中清理掉. 在内存中存储不下的分区数据会被写到磁盘中, 且提供了和现有并行数据处理系统相同的性能保证.

3 RDD的表达

RDD的内容究竟该包括哪些内容才能达到轻松跟踪RDD迭代状态以及应对各种业务逻辑的目的呢？理论上讲，RDD的transformation类型算子越多，则代表RDD能够应对的场景就越多，而不同的transformation算子能够由用于任意结合则会将极大提升其应用场景的范围。
Spark中RDD的设计其客观表达是基于DAG图的形式，图中的每个节点表达相互独立的每个RDD，而RDD中的编程实现主要包含的就是5种信息，或叫做5种接口。以下五个信息可以表达 RDDs: 一个分区列表, 每一个分区就是数据集的原子块. 一个父亲 RDDs 的依赖列表. 一个计算父亲的数据集的函数. 分区模式的元数据信息以及数据存储信息. 比如, 基于一个 HDFS 文件创建出来的的 RDD 中文件的每一个数据块就是一个分区, 并且这个 RDD 知道每一个数据块存储在哪些机器上, 同时, 在这个 RDD 上进行 map 操作后的结果有相同的分区数, 当计算元素的时候, 将 map 函数应用到父亲 RDD 数据中的.

那么，我们定义了每个RDD需要实现的接口后，需要考虑的就是如何定义不同RDD之间的依赖关系！当然，对于一个完整通用的系统而言，需要找到具有普适性的定义方式。RDD的图表达中引入两种依赖关系：1）宽依赖 。2）窄依赖。

宽依赖：表示父亲 RDDs 的一个分区可以被子 RDDs 的多个子分区所依赖
窄依赖：表示父亲 RDDs 的一个分区最多被子 RDDs 一个分区所依赖
举例：map 操作是一个窄依赖, join 操作是一个宽依赖操作。

下图是论文中的一个图例，其中蓝色实心矩形表示分区，一个大的空心矩形代表一个RDD。

为何要分为这两种依赖关系？

第一,：
窄依赖允许所有的父节点分区能够在一台节点中完成计算。例如可以将每一个元素进行 map 操作后紧接着执行filter 操作, 与此相反, 宽依赖需要父亲 RDDs 的所有分区数据在不同的节点之间进行重新洗牌和网络传输类似于MR。
第二：
窄依赖从一个失败节点中恢复是非常高效的, 因为只需要重新计算相对应的父亲的分区数据就可以, 而且这个重新计算是在不同的节点进行并行重计算的, 与此相反, 在一个含有宽依赖的血缘关系 RDDs 图中, 一个节点的失败可能导致一些分区数据的丢失, 但是只用重新计算父 RDD 的所有分区的数据。

4 Job调度

RDD的延迟性执行使得其能够实现交互式执行，举个例子，你可以在shell窗口中写一堆transformation的代码，但是此时代码不会执行，你还可以继续写下去，直到你满意为止，而且中间你可以通过重写相同变量名的不同方法来覆盖更新一些变量（RDD），直到最后你使用了一个action算子后，整个代码块才会执行。这就是交互式操作。

那么整个代码块被激活后JOB是如何调度的呢？

当一个用户对某个 RDD 调用了 action 操作（比如 count 或者 save ）的时候调度器会检查这个 RDD 的血缘关系图, 然后根据这个血缘关系图构建一个含有 stages 的有向无环图（ DAG ）, 最后按照步骤执行这个 DAG 中的 stages , 如下图所示。每一个 stage 包含了尽可能多的带有窄依赖的 transformations 操作. 这个 stage 的划分是根据需要 shuffle 操作的宽依赖或者任何可以不依赖父节点的RDD， 然后调度器可以调度启动 tasks 来执行父Stage未被执行的Stage，一直计算到最终的RDD。

上图中空心矩形表示 RDDs ,蓝色的实心方形表示分区, 黑色的是表示这个分区的数据存储在内存中, 最后对 RDD-G 调用 action 操作,。根据宽依赖生成很多 stages , 且将窄依赖的 transformations 操作放在 stage 中。

调度器在分配 tasks 的时候是采用延迟调度来达到数据本地性的目的（说白了, 就是数据在哪里, 计算就在哪里）. 如果某个分区的数据在某个节点上的内存中, 那么将这个分区的计算发送到这个机器节点中. 如果某个 RDD 为它的某个分区提供了这个数据存储的位置节点, 则将这个分区的计算发送到这个节点上.

对于宽依赖（比如 shuffle 依赖）, 中间数据会写入到节点的磁盘中以利于从错误中恢复, 这个和 MapReduce 将 map 后的结果写入到磁盘中是很相似的.

5 RDD的内存管理

既然RDD基于内存迭代，那么内存资源需要一定的管理方式使其更高效地被利用。
目前Spark支持三种RDD存储介质：1）完全内存中的非序列化jvm对象。2）内存中的序列化数据。3）持久化在磁盘。当然，第一种方式是最好的，因为计算速度最快，但是大多时候内存容不下这么多RDD，则会使用第三种。

内存中的RDD怎样被回收来释放资源呢？Spark中采用LRU的回收方式，即：如果新的RDD无法被内存容纳，则内存中会启动LRU策略来将最近很少使用的RDD进行清除。

6 容错的更高保证：checkpointing

如果迭代链条十分长，那么有必要适当进行checkpoint来缓存一些中间结果来保证计算链条的可靠性。
PageRank是一种需要重复很多次迭代的算法，且真实场景下该算法适配的数据集很大，因此很有必要进行checkpoint。checkpoint对于一些有较高存储保障的RDD来讲并没有用，例如textfile，大多都是从HDFS读取的原始数据。