2023学年完整公开课版Range分区

Range分区目录RangePartitionerRange分区案例代码21222RangePartitioner1RangePartitioner

SparkCore中除了HashPartitioner分区器外，另外一个比较重要的已经实现的分区器，主要用于RDD的数据排序相关API中，比如sortByKey底层使用的数据分区器就是RangePartitioner分区器。RangePartitioner分区器的主要作用RDD的数据排序RangePartitioner

RangePartitioner分区器的实现方式主要是通过两个步骤来实现的。

第一步：先重整个RDD中抽取出样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array[KEY]类型的数组变量rangeBounds。RangePartitioner

第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；RDD中的KEY类型必须是可以排序RangePartitioner分区器要求Range分区案例代码2Range分区案例代码

//Range分区案例代码见下文。defdetermineBounds[K:Ordering:ClassTag](candidates:ArrayBuffer[(K,Float)],partitions:Int):Array[K]={valordering=implicitly[Ordering[K]]//按照数据进行数据排序，默认升序排列valordered=candidates.sortBy(_._1)//获取总的样本数量大小valnumCandidates=ordered.size

//代码部分A//代码部分B//代码部分C}

Range分区案例代码

//代码部分A。//计算总的权重大小valsumWeights=ordered.map(_._2.toDouble).sum//计算步长valstep=sumWeights/partitionsvarcumWeight=0.0vartarget=stepvalbounds=ArrayBuffer.empty[K]vari=0varj=0varpreviousBound=Option.empty[K]Range分区案例代码

//代码部分B。while((i<numCandidates)&&(j<partitions-1)){//获取排序后的第i个数据及权重val(key,weight)=ordered(i)//累计权重cumWeight+=weightif(cumWeight>=target){//Skipduplicatevalues.//权重已经达到一个步长的范围，计算出一个分区id的值if(previousBound.isEmpty||ordering.gt(key,previousBound.get)){

Range分区案例代码

//代码部分C。//上一个边界值为空，或者当前边界key数据大于上一个边界的//那么当前key有效，进行计算。bounds+=key//添加当前key到边界集合中target+=step//累计target步长界限//分区数量加1j+=1previousBound=Some(key)//上一个