• RDD可以被cache在内存中。这种机制无疑可以简单有效地解决MapReduce在迭代计算时反复读取写出磁盘的问题，但同时大大增大了内存开销。
• RDD帮助Fault Recovery. RDD可存储lineage信息来重建lost partitions. 比如记录RDD的转化过程。MapReduce采用的是checkpointing机制，代价要大很多。
• RDD是一个logical单元，甚至不需要实例化，而只需包含从磁盘重建workset的信息。

RDD组成

RDD，弹性分布式数据集。是设计用来表示数据集的一种数据结构。为了让 RDD 能 handle 更多的问题，规定 RDD 应该是只读的，分区记录的一种数据集合。

每个RDD都包含：

• 一组RDD分区（partition），即数据集的原子组成部分
• 一个函数，即在父RDD上执行何种计算；
• 对父RDD的一组依赖，这些依赖描述了RDD的Lineage
• 元数据，描述分区模式和数据存放的位置，可选择的。
• 一个对于key-value类型的RDD的Partitioner，可选择的。

例如，一个表示HDFS文件的RDD包含：各个数据块的一个分区，并知道各个数据块放在哪些节点上。而且这个RDD上的map操作结果也具有同样的分区，map函数是在父数据上执行的。

RDD 产生

RDD产生有两种方式：

• 通过集合类生成：sc.parallelize。把 driver 端定义的一个数据集，或者一个获取数据集的生成器，分发到 worker 上的 executor 中，以供后续分析。这种方式在测试代码逻辑时经常用到。
• 从数据源读取：本地文件，HDFS(hdfs://), Cassandra, HBase, Amazon S3(s3n://)等
  • 文本文件：textFile方法
  • SequenceFiles： sequenceFile[K, V]方法读取
  • 其他Hadoop输入格式：hadoopRDD方法

RDD保存：

• 文本：JavaRDD.saveAsTextFile。
  • 会按照执行task的多少生成多少个文件，比如part-00000到part-0000n，n是task的个数，亦是最后的stage的分区数。coalesce可以调整生成的文件数。
• 对象：JavaRDD.saveAsObjectFile、JavaSparkContext.objectFile

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

public class RDDInit {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("RDDInit").setMaster("local").set("spark.some.config","forRunInYarn");
        JavaSparkContext sc = new JavaSparkContext(conf);
        //RDD生成
        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
        JavaRDD<Integer> distData = sc.parallelize(data);
        System.out.println(distData.take(5));
        //RDD保存
        distData.coalesce(1).saveAsTextFile("output/RDDWordCount/textFile");
        distData.saveAsObjectFile("output/RDDWordCount/object");
        //读取保存的RDD
        JavaRDD<Integer> readObject = sc.objectFile("output/RDDWordCount/object/part-00000");
        System.out.println(readObject.take(3));
        sc.stop();
    }
}

RDD 操作

两类算子

RDD存在两类算子:

• transformation,主要是用于对RDD进行数据转换，生成一个新的RDD。包括：map, flatMap, filter, union, sample, join, groupByKey, cogroup, ReduceByKey, cros, sortByKey, mapValues等
• action，是要将转换好的RDD再转换成原始数据。 包括：collect（返回元素本身）, reduce, count（返回RDD中的元素个数）, save（将RDD输出到存储系统）,first，take， lookupKey等。

惰性运算

spark运算是一种懒惰运算，其程序执行逻辑是：

• 对于一系列transformation算子，在遇到action算子之前，这些transformation是不会执行的，而会生成一个运算逻辑图。
• 当遇到一个action算子的时候，才根据前面的运算逻辑图执行程序。
• 这样如果有台机器宕机了之后，只需要根据逻辑图将宕机中的RDD进行重新计算就可以了（窄依赖的情况下，宽依赖的情况下计算代价要更高。）

下面有一个求所有字符长度的例子：

1
2
3
4
5
6
7
8
9
10
11

/**
 * length count:所有字符的length
 */
//生成RDD
List<String> data = Arrays.asList("hello world", "hello spark","hello rdd", "hello spark rdd");
JavaRDD<String> lines = sc.parallelize(data);
//transformation算子
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
//action算子
int totalLength = lineLengths.reduce((a, b) -> a + b);
System.out.println(totalLength);//输出：46

RDD关系

DAG有向无环图

DAG - Direct Acyclic Graph，有向无环图。假设图二中的A,B,C,D,E都代表spark里不同的RDD。

说明：

• 图：图是表达RDD Lineage信息的一个结构。在spark中，大部分 RDD 都是通过其他 RDD 进行转换而来的，比如说图二中，B和D都是通过A转换而来的，而C是通过B转换而来，E的话是通过B和D一起转换来的。
• 有向：有向就更容易理解了，简单来说就是 lineage 是一个 top-down 的结构，而且是时间序列上的 top-down 结构。一个时间上的先来后到，即祖先与子孙的关系，是不可逆的。
• 无环：spark 的优化器在这里也发挥了很大的作用。假设有图三中左下 B,D,E 这样一个 RDD 转换图，那当我们的需要执行 D.collect 操作的时候，就会引发一个死循环了。

RDD依赖

依赖主要分为两类：窄依赖和宽依赖。

如下图：方框表示RDD，实心矩形表示分区

• 对于map和filter形式的转换来说，它们只是将Partition的数据根据转换的规则进行转化，并不涉及其他的处理，可以简单地认为只是将数据从一个形式转换到另一个形式。
• 对于union，只是将多个RDD合并成一个，parent RDD的Partition(s)不会有任何的变化，可以认为只是把parent RDD的Partition(s)简单进行复制与合并。
• 对于join，如果每个Partition仅仅和已知的、特定的Partition进行join，那么这个依赖关系也是窄依赖。对于这种有规则的数据的join，并不会引入昂贵的Shuffle。对于窄依赖，由于RDD每个Partition依赖固定数量的parent RDD(s)的Partition(s)，因此可以通过一个计算任务来处理这些Partition，并且这些Partition相互独立，这些计算任务也就可以并行执行了。
• 对于groupByKey，子RDD的所有Partition(s)会依赖于parent RDD的所有Partition(s)，子RDD的Partition是parent RDD的所有Partition Shuffle的结果，因此这两个RDD是不能通过一个计算任务来完成的。同样，对于需要parent RDD的所有Partition进行join的转换，也是需要Shuffle，这类join的依赖就是宽依赖而不是前面提到的窄依赖了。

任务调度

Spark调度器根据目标RDD的Lineage图创建一个由stage构成的无回路有向图（DAG）。每个stage内部尽可能多地包含一组具有窄依赖关系的转换，并将它们流水线并行化（pipeline）。stage的边界有两种情况：一是宽依赖上的Shuffle操作；二是已缓存分区，它可以缩短父RDD的计算过程。

实线方框表示RDD，实心矩形表示分区（黑色表示该分区被缓存）。要在RDD G上执行一个动作，调度器根据宽依赖创建一组stage，并在每个stage内部将具有窄依赖的转换流水线化（pipeline）。 本例不用再执行stage 1，因为B已经存在于缓存中了，所以只需要运行2和3。

故障恢复

Spark故障恢复提供两种方式

• Linage：通过数据的血缘关系，再执行一遍前面的处理
• Checkpoint：Lineage链较长、宽依赖的RDD需要采用检查点机制。将数据集存储到持久存储中。

参考

博客：spark 之 RDD

博客：RDD论文中文版

知乎：与 Hadoop 对比，如何看待 Spark 技术？

博客：RDD依赖关系

博客：Job 逻辑执行图