想了解RDD，看这一篇文章就够了SparkCore之RDD详解（史上最易懂）

1、 什么是RDD

- 为什么要有RDD?
在许多迭代式算法(比如机器学习、图算法等)和交互式数据挖掘中&＃xff0c;不同计算阶段之间会重用中间结果&＃xff0c;即一个阶段的输出结果会作为下一个阶段的输入。但是&＃xff0c;之前的MapReduce框架采用非循环式的数据流模型&＃xff0c;把中间结果写入到HDFS中&＃xff0c;带来了大量的数据复制、磁盘IO和序列化开销。且这些框架只能支持一些特定的计算模式(map/reduce)&＃xff0c;并没有提供一种通用的数据抽象。

AMP实验室发表的一篇关于RDD的论文:《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》就是为了解决这些问题的

RDD提供了一个抽象的数据模型&＃xff0c;让我们不必担心底层数据的分布式特性&＃xff0c;只需将具体的应用逻辑表达为一系列转换操作(函数)&＃xff0c;不同RDD之间的转换操作之间还可以形成依赖关系&＃xff0c;进而实现管道化&＃xff0c;从而避免了中间结果的存储&＃xff0c;大大降低了数据复制、磁盘IO和序列化开销&＃xff0c;并且还提供了更多的API(map/reduec/filter/groupBy…)

- RDD是什么?
RDD(Resilient Distributed Dataset)叫做弹性分布式数据集&＃xff0c;是Spark中最基本的数据抽象&＃xff0c;代表一个不可变、可分区、里面的元素可并行计算的集合。

&＃xff08;Resilient Distributed Dataset&＃xff09;RDD单词拆解

• Resilient:它是弹性的&＃xff0c;RDD中的数据可以保存在内存中或者磁盘里面
• Distributed:它里面的元素师分布式存储的&＃xff0c;可以用于分布式计算
• Dataset:它是一个集合&＃xff0c;可以存放很多元素

2、RDD的主要属性

1、Alist of partitions:
一组分片(Partition)/一个分区(Partition)列表&＃xff0c;即数据集的基本组成单位。
对于RDD来说&＃xff0c;每个分片都会被一个计算任务处理&＃xff0c;分片数决定并行度。
用户可以在创建RDD时指定RDD的分片个数&＃xff0c;如果没有指定&＃xff0c;那么就会采用默认值。

2.A function for computing each split &＃xff1a;
一个函数会被作用在每一个分区。
spark中RDD的计算是一分区为单位的&＃xff0c;compute函数会被作用到每一个分区上。

3.A list of dependencies on other RDDs&＃xff1a;
一个RDD会依赖于其他多个RDD。
RDD的每次转换都会生成一个新的RDD&＃xff0c;所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时&＃xff0c;Spark可以通过这个依赖关系重新计算丢失的分区数据&＃xff0c;而不是对RDD的所有分区进行重新计算。(Spark的容错机制)

4.Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)&＃xff1a;

Spark中的分区函数&＃xff0c;一个是基于哈希的HashPartitioner&＃xff0c;另外一个是基于范围的RangePartitioner。

对于KV类型的RDD会有一个Partitioner函数&＃xff0c;即RDD的分区函数&＃xff08;可选项&＃xff09;
只有对于key-value的RDD&＃xff0c;才会有Partitioner&＃xff0c;非key-value的RDD的Parititioner的值是None。Partitioner函数决定了RDD本身的分区数量&＃xff0c;也决定了parent RDD Shuffle输出时的分区数量。

5.Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)&＃xff1a;
可选项,一个列表&＃xff0c;存储每个Partition的位置(preferred location)。
对于一个HDFS文件来说&＃xff0c;这个列表保存的就是每个Partition所在的块的位置。按照"移动数据不如移动计算"的理念&＃xff0c;Spark在进行任务调度的时候&＃xff0c;会尽可能选择那些存有数据的worker节点来进行任务计算。

●总结
RDD 是一个数据集&＃xff0c;不仅表示了数据集&＃xff0c;还表示了这个数据集从哪来&＃xff0c;如何计算。
主要属性包括
1.多分区
2.计算函数
3.依赖关系
4.分区函数(默认是hash)
5.最佳位置

3、RDD的3种创建方式

1.由外部存储系统的数据集创建&＃xff0c;包括本地的文件系统&＃xff0c;还有所有Hadoop支持的数据集&＃xff0c;比如HDFS、Cassandra、HBase等
val rdd1 &＃61; sc.textFile(“hdfs://node01:8020/wordcount/input/words.txt”)

2.通过已有的RDD经过算子转换生成新的RDD
val rdd2&＃61;rdd1.flatMap(_.split(" "))

3.由一个已经存在的Scala集合创建
val rdd3 &＃61; sc.parallelize(Array(1,2,3,4,5,6,7,8))
或者
val rdd4 &＃61; sc.makeRDD(List(1,2,3,4,5,6,7,8))
makeRDD方法底层调用了parallelize方法

4、RDD的算子分类

●分类
RDD的算子分为两类:
1.Transformation转换操作:返回一个新的RDD
2.Action动作操作:返回值不是RDD(无返回值或返回其他的)

●注意:
RDD不实际存储真正要计算的数据&＃xff0c;而是记录了数据的位置在哪里&＃xff0c;数据的转换关系(调用了什么方法&＃xff0c;传入什么函数)
RDD中的所有转换都是惰性求值/延迟执行的&＃xff0c;也就是说并不会直接计算。只有当发生一个要求返回结果给Driver的Action动作时&＃xff0c;这些转换才会真正运行。
之所以使用惰性求值/延迟执行&＃xff0c;是因为这样可以在Action时对RDD操作形成DAG有向无环图进行Stage的划分和并行优化&＃xff0c;这种设计让Spark更加有效率地运行。

5、RDD的持久化/缓存

在实际开发中某些RDD的计算或转换可能会比较好费时间&＃xff0c;如果这些RDD后续还会频繁的被使用到&＃xff0c;那么可以将这写RDD进行持久化/缓存&＃xff0c;这样下次再使用到的时候就不用再重新计算了&＃xff0c;提高了程序运行的效率

5.1、持久化/缓存API详解

●persist方法和cache方法
RDD通过persist或cache方法可以将前面的计算结果缓存&＃xff0c;但是并不是这两个方法被调用时立即缓存&＃xff0c;而是触发后面的action时&＃xff0c;该RDD将会被缓存在计算节点的内存中&＃xff0c;并供后面重用。
通过查看RDD的源码发现cache最终也是调用了persist无参方法(默认存储只存在内存中)

5.2、持久化存储级别

5.3、总结

1.RDD持久化/缓存的目的是为了提高后续操作的速度
2.缓存的级别有很多&＃xff0c;默认只存在内存中,开发中使用memory_and_disk
3.只有执行action操作的时候才会真正将RDD数据进行持久化/缓存
4.实际开发中如果某一个RDD后续会被频繁的使用&＃xff0c;可以将该RDD进行持久化/缓存

6、第四章 RDD容错机制Checkpoint

●持久化的局限
持久化/缓存可以把数据放在内存中&＃xff0c;虽然是快速的&＃xff0c;但是也是最不可靠的&＃xff1b;也可以把数据放在磁盘上&＃xff0c;也不是完全可靠的&＃xff01;例如磁盘会损坏等。

●问题解决
Checkpoint的产生就是为了更加可靠的数据持久化&＃xff0c;在Checkpoint的时候一般把数据放在在HDFS上&＃xff0c;这就天然的借助了HDFS天生的高容错、高可靠来实现数据最大程度上的安全&＃xff0c;实现了RDD的容错和高可用

6.1、总结

●开发中如何保证数据的安全性性及读取效率
可以对频繁使用且重要的数据&＃xff0c;先做缓存/持久化&＃xff0c;再做checkpint操作

●持久化和Checkpoint的区别
1.位置
Persist 和 Cache 只能保存在本地的磁盘和内存中(或者堆外内存–实验中)
Checkpoint 可以保存数据到 HDFS 这类可靠的存储上

2.生命周期
Cache和Persist的RDD会在程序结束后会被清除或者手动调用unpersist方法
Checkpoint的RDD在程序结束后依然存在&＃xff0c;不会被删除

3.Lineage(血统、依赖链–其实就是依赖关系)
Persist和Cache&＃xff0c;不会丢掉RDD间的依赖链/依赖关系&＃xff0c;因为这种缓存是不可靠的&＃xff0c;如果出现了一些错误(例如 Executor 宕机)&＃xff0c;需要通过回溯依赖链重新计算出来
Checkpoint会斩断依赖链&＃xff0c;因为Checkpoint会把结果保存在HDFS这类存储中&＃xff0c;更加的安全可靠&＃xff0c;一般不需要回溯依赖链

●补充&＃xff1a;Lineage
RDD的Lineage(血统、依赖链)会记录RDD的元数据信息和转换行为&＃xff0c;当该RDD的部分分区数据丢失时&＃xff0c;它可以根据这些信息来重新运算和恢复丢失的数据分区。
在进行故障恢复时&＃xff0c;Spark会对读取Checkpoint的开销和重新计算RDD分区的开销进行比较&＃xff0c;从而自动选择最优的恢复策略。

7、RDD的依赖关系

窄依赖&＃xff1a;
一个RDD的分区至多被子RDD的某个分区使用一次
容错&＃xff1a;某个分区出故障了&＃xff0c;可以快速将丢失的分区并行计算出来&＃xff0c;窄依赖可以在单节点上完成运算。
容错和计算速度都比宽依赖好。
宽依赖&＃xff1a;
一个父RDD的分区会被子RDD的分区使用多次&＃xff0c;即只能前面的算好后才进能进行后续的计算&＃xff0c;只有遇到RDD数据是key-value才会有Partitioner
等到父类的所有数据都被传输到各个节点后才能计算&＃xff08;MR-shuffle)
容错&＃xff1a;
某个分区出故障了&＃xff0c;要计算前面所有父分区&＃xff0c;代价会很大。解决办法&＃xff1a;把之前的结果存在磁盘&＃xff0c;当分区
出现故障&＃xff0c;直接读磁盘文件。

好了&＃xff0c;以上内容就到这里了。不知道小编本篇内容有没有帮助到你呢。欢迎路过的朋友关注小编哦。各位朋友关注点赞是小编坚持下去的动力。小编会继续为大家分享更多的知识哦~~~。

我是小哪吒&＃xff0c;一名互联网行业的工具人。小编的座右铭&＃xff1a;“我不生产代码&＃xff0c;我只做代码的搬运工”…哈哈哈&＃xff0c;我们下期见哦&＃xff0c;Bye~