Spark之RDD本质

Posted 2021-03-08 jaxyoun

1、在一个完整的数据转换流程里往往涉及到多个具有衍生关系RDD，这些RDD其实是通过逻辑串联来利用装饰器模式层层包装扩展的的一堆对象，这些相邻RDD间必须有继承关系。并且比Java中的装饰器来的更彻底，借助Scala的抽象控制特性，这一系列RDD不代表任何实际数据，也不负责装载数据，描述的是纯粹的逻辑抽象DAG，只有调用了尾函数后才会触发逻辑DAG的执行。

2、弹性：分别表现在如下几个方面
血缘：可以使用cache、checkpoint等机制灵活改变血缘继承关系。
计算：计算时可以灵活的使用内存和磁盘（早期shuffle无磁盘参与，存在性能瓶颈和OOM风险，后期加入）。
分区：在创建RDD和转换过程中可以灵活的调整数据分区。
容错：利用checkpoint持久化机制，可以在流程不同点位实现容错。

3、分布式，可分区，每个分区发可以分配不同的Executor，实现对数据的并行计算。对于集合数据，用户可以利用并行度来指定分区大小，对于内容不可预知的文件数据，用户只能指定最小并行度，具体分多少区，由分区算法决定，Spark-RDD底层使用的是Hadoop的分区算法。

4、不可变，RDD类底层实现中没有向用户暴露修改的API，所以用户无法直接修改源RDD的内容，要实现数据在逻辑上的改变，就只能通过转换操作来实现，且转换结果必须是新的RDD对象。

5、迁移数据不如移动计算的位置，移动数据涉及网络IO，其成本往往较高，所以尽量让计算逻辑从本地取数据。所以Spark在分配任务时，优先将任务分配到源数据分片所在的节点上。
为了实现负载均衡，当某些Executor进程空闲下来后，首先会从进程本地获取数据，如果不成功就会从本机上其他Executor进程获取数据，如果仍拿不到，就会从本机架上的其他机器的Executor获取数据，最后才会从其他机架的机器上获取数据，总之就是一最小的IO代价为原则，从最近的地方获取数据。

6、算子：人类解决问题的思维模式一般是渐进式的，也就是从问题的初始局面到被解决的整个过程通常是被多步分解的，每个步骤的推进伴随着问题状态的改变，要改变问题的状态就需要做相应的操作，这些操作就统称为Operator即算子。一个算子可以由一个命令、命令组合、函数、函数组合构成。所有的RDD方法都是算子，它们又可以分为做转换的TransformationOperator和促发计算的ActionOperator。

7、在一个Spark用户作业程序中，创建SparkContext对象所在的类的实例化对象，被称之为Driver对象。程序里只有算子中直接参与计算的代码才会被分配到Executor上去执行，负责真实数据的计算，其他部分【包括flatMap算子】均在Driver中执行。如rdd.flatMap(it => it * 2);这段代码，只有it * 2这段代码才承载着真正的计算逻辑，所以只有它会被分配到Executor上去执行。

8、创建：
1.从集合创建：sc.makeRDD(0 to 10, 4)
2.从别的RDD创建：RDD2 = RDD1.map()
3.从外部文件读取：sc.textFile("/opt/data/text.txt")

9、RDD.map、RDD.mapPartitions、RDD.mapPartitionsWithIndex三个算子的区别：
1.RDD.map(it => it * 2)：会根据集合中元素的数量N，map算子在Driver中执行N次，也会分发真实计算逻辑task【it * 2】到Executor N次，会额外增加分发开销。
2.RDD.mapPartitions(dataList => {dataList.foreach( { _ * 2 })：dataList代表的是一个分区上的所有数据元素的集合，会根据数据分区数量P，mapPartitions算子在Driver中执行P次，计算逻辑【 {dataList.foreach( { _ * 2 }】会分发到Executor P次，各个Executor上的真实计算逻辑执行次数与对应分片上的数据集元素个数相同，最终计算逻辑执行的总次数还是等于整个RDD元素的数量N。这样可以大大减少了计算逻辑分发的次数，由于分发过程涉及进程间通信，而进程间通信比内存中通信至少慢三个数量级，所以更高效。
但是也存在一个严重风险，在一个Executor上，一个算子的计算逻辑对一个分区数据的处理，执行持续的时间是以整个分区处理时间为生命周期，如果分区中数据源源不断地增加，整个处理过程就可能持续很久，程序对数据的引用一直没有断开，GC无法回收已处理数据占用的内存，对内存的占用势必不断增加，可能引发OOM。
3.RDD.mapPartitionsWithIndex{
case(pId, dataList) => {
dataList.map((_, "pId=" + pId))
}
在计算和性能上与mapPartitions算子是一样的，但它可以拿到分区的序号。

10、RDD分区与任务的分配：一个partition上的数只能由一个task来消费，通常一个分区就对应一个task，适当增加RDD分区数量能提高App的并行度，每个task可以灵活的分配到Executor上去执行。【与Flink类似，operator是DAG层面的东西，构成DAG的节点，属于逻辑执行计划。而task则属于物理执行计划，是operator在具体计算环境下的真实执行体，一个operator可能拥有代表其逻辑的多个task作为分身在集群中执行】

11、当上游RDD的一个分区中的元素被分散到多个下游RDD分区(宽依赖)，必然经过Shuffle操作。如果上游分区中的元素以汇聚的形式流向下游分区，亦即保证了来自同一个上游分区的所有元素，到达下游RDD时仍然能团结在一个下游分区中就不会发生shuffle，此时下游有效分区数不会超过上游分区数。
***
12、一个SparkContext代表着一个App，执行流程每遇到一个Action算子形成一个新Job，在每个Job内每遇到一次Shuffle操作形成一个新Stage，在一个Stage中可以包含流水线的一或多个算子，这些算子形成OperatorChain，系统根据Stage中最后一个RDD的分区数N，并依照OperatorChain的计算逻辑建立N个Task。
在范围上大致关系：App > Job > Stage > Task，从左至右范围依次减小，呈现一对多的关系。