Spark之RDD本质
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,从左至右范围依次减小,呈现一对多的关系。