spark transform系列__reduceByKey

reduceByKey 通过PairRDDFunctions进行的实现,reduceByKey的操作是把两个V类型的值进行处理,并最终返回的还是一个V类型的结果(V类型 就是value的类型).针对一个reduceByKey的操作,需要执行shuffle的操作,也就是说如果包含有reduceByKey时,会生成两 个执行的stage,第一个stage会根据shuffle的partition与分区的算子,对数据重新进行分区操作,第二个stage去读取重新分区 的数据.

def reduceByKey(func: (V, V) => V): RDD[(K, V)] = self.withScope {

这里首先先检查要做reduceByKey的操作的rdd是否包含有一个partitioner的算子,如果有,直接使用这个算子,否则生成Hash算子,检查spark.default.parallelism配置是否有配置,如果有配置并行度,重新分区的个数读取这个配置,否则取上个rdd的partition的个数.

Func是需要执行的reduceByKey操作的function,这个function的传入类型与返回类型是相同的类型.
  reduceByKey(defaultPartitioner(self), func)
}

 

def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = self.withScope {

这里调用了combineByKeyWithClassTag的函数,

这个函数需要传入的参数中,前三个参数的说明:

参数1:(v:V)=>v, 这个function表示如果key还没有值时createCombiner: V => C 的函数操作,在这里直接是把传入的key对应的value当成createCombiner的返回值,

参数2:这个地方直接把reduceByKey的function当作为mergeValue: (C, V) => C函数的操作,这是对两个key相同的值进行的合并操作,

参数3:这个地方直接把reduceByKey的function当作为mergeCombiners: (C, C) => C函数的操作,这个是对key相同的两个combiner进行的合并操作.

在combineByKeyWithClassTag中定义的V表示上面三个参数中C的类型直接使用V(原rdd中value的类型)的类型

在执行combineByKeyWithClassTag的操作时,默认mapSideCombine参数的值为true,表示需要执行map端的combine的操作.
  combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner)
}

 

接下来看看combineByKeyWithClassTag的函数的实现:

如果执行这个操作时,rdd的key是一个数组类型时,同时设置有map端执行combine操作,提示错误.

如果rdd的key是一个数组类型时,同时分区算子是默认的Hash算子时,提示错误.

if (keyClass.isArray) {
  if (mapSideCombine) {
    throw new SparkException("Cannot use map-side combining with array keys.")
  }
  if (partitioner.isInstanceOf[HashPartitioner]) {
    throw new SparkException("Default partitioner cannot partition array keys.")
  }
}

 

根据传入的前三个参数,生成Aggregator,设置有mapSideCombine时,Aggregator必须存在.
val aggregator = new Aggregator[K, V, C](
  self.context.clean(createCombiner),
  self.context.clean(mergeValue),
  self.context.clean(mergeCombiners))
if (self.partitioner == Some(partitioner)) {

这个情况需要说明一下,如果执行当前的操作传入的partitioner与执行这个操作对应的RDD是相同的算子时,这个时候不对当前的操作生成新的RDD,也就是这个操作不在执行SHUFFLE的操作,直接使用当前操作的RDD的Iterator.
  self.mapPartitions(iter => {
    val context = TaskContext.get()
    new InterruptibleIterator(context, aggregator.combineValuesByKey(iter, 

       context))
  }, preservesPartitioning = true)
} else {

这种情况下,执行这个操作的partitioner是一个新生成的或者说与当前要执行这个操作的RDD的partitioner不是相同的实例,表示这个操作需要执行shuffle的操作,生成一个ShuffledRDD实例.
  new ShuffledRDD[K, V, C](self, partitioner)
    .setSerializer(serializer)
    .setAggregator(aggregator)
    .setMapSideCombine(mapSideCombine)
}

 

ShuffledRDD的实例生成:

先看看实例的生成部分:

这里传入的prev是生成ShuffleRDD的上层的RDD.

下面在ShuffledRDD的实例定义部分的红色部分,在实例生成时设置对上层的RDD的依赖为Nil,表示对上层RDD的依赖部分不在实例生成部分定义.

class ShuffledRDD[K: ClassTag, V: ClassTag, C: ClassTag](
    @transient var prev: RDD[_ <: Product2[K, V]],
    part: Partitioner)
  extends RDD[(K, C)](prev.context, Nil) {

 

接下来看看ShuffledRDD如何处理上层RDD的依赖部分:

在ShuffledRDD中会重写getDependencies函数:

override def getDependencies: Seq[Dependency[_]] = {

在这里,生成对ShuffledRDD的依赖为ShuffleDependency实例.这个依赖的RDD就是生成这个ShuffledRDD的上层的RDD的实例.
  List(new ShuffleDependency(prev, part, serializer, keyOrdering, aggregator,

       mapSideCombine))
}

 

生成ShuffleDependency实例的关键代码:

每生成一个ShuffleDependency的实例时,会对每个Shuffle的依赖生成一个唯一的shuffleId,用于对此stage中每个task的结果集的跟踪.

val shuffleId: Int = _rdd.context.newShuffleId()

向shuffleManager注册这个这个shuffle的依赖.task的结果集向driver通知时,首先需要这个shuffle是一个注册的shuffle.
val shuffleHandle: ShuffleHandle = _rdd.context.env.shuffleManager.registerShuffle(
  shuffleId, _rdd.partitions.size, this)