在Spark中,每一个RDD是对于数据集在某一状态下的表现形式,比如说:map、filter、group by等都算一次操作,这个状态有可能是从前一状态转换而来的;
因此换句话说一个RDD可能与之前的RDD(s)有依赖关系;RDD之间存在依赖关系;
根据依赖关系的不同,可以将RDD分成两种不同的类型:宽依赖和窄依赖。
窄依赖:一个父RDD的partition至多被子RDD的某个partition使用一次;
宽依赖:一个父RDD的partition会被子RDD的partition使用多次,需要shuffle操作;
图中方框描述:外面的大方框是一个RDD,里面的小方块是RDD中的partition,多个partition组成一个RDD
窄依赖
定义:一个父RDD的partition至多被子RDD的某个partition使用一次;
不需要shuffle,partition范围不会改变,一个partition经过transform后还是一个partition,虽然内容发生了变化;可以进行pipeline计算,快速完成;
在某个节点上可以一次性全部计算完所有的父partition(pipeline流水式的计算方式):
a.map().filter().reduceByKey() 这样多步操作一次性计算完毕,而不需要第一步执行完后保存起来,第二步再去读取再计算再存储。。。。。。
窄依赖可以在单节点上完成运算,非常高效。
容错:某个partition挂了,快速将丢失的partition并行计算出来。
容错和计算速度都比宽依赖强。
窄依赖又分为两种:
OneToOneDependency:一对一的依赖,一父一子,最典型的是map/filter。
RangeDependency:一定范围的RDD直接对应,最典型的是Union。
parent RDD的某个分区的partitions对应到child RDD中某个区间的partitions;
union:多个parent RDD合并到一个chind RDD,故每个parent RDD都对应到child RDD中的一个区间;
注意:union不会把多个partition合并成一个partition,而是简单的把多个RDD的partitions放到一个RDD中,partition不会发生变化。
宽依赖
定义:一个父RDD的partition会被子RDD的partition使用多次;只能前面的算好后才能进行后续的计算;只有等到父partition的所有数据都传输到各个节点后才能计算(经典的mapreduce场景)
容错:某个partition挂了,要计算前面所有的父partition,代价很大。
spark是把map部分的数据计算完成后物化到map端的磁盘上,挂了之后直接从磁盘中读取即可。
class ShuffleDependency[K, V]( @transient rdd: RDD[_ <: Product2[K, V]], val partitioner: Partitioner, val serializerClass: String = null) extends Dependency(rdd.asInstanceOf[RDD[Product2[K, V]]]) { val shuffleId: Int = rdd.context.newShuffleId() }
首先:需要基于PairRDD,因为一般需要依据key进行shuffle,所以数据结构往往是key-value;
其次:由于需要shuffle,所以就需要给出partitioner;
然后:shuffle不像map可以在local运行,往往需要网络传输或存储,所以需要serializerClass;
最后:每个shuffle需要分配一个全局的id,context.newShuffleId()的实现就是把全局id累加;
Spark分析之Dependency