源码--spark2.X的shuffle

一、源码流程

1、首先sparkContext被实例化的时候,会创建sparkEnv,在sparkEnv的create方法中对shuffleManager进行了初始化

val shortShuffleMgrNames = Map(
      "sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName,
      "tungsten-sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName)
val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")
val shuffleMgrClass =
      shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase(Locale.ROOT), shuffleMgrName)
//这句代码其实创建了ShuffleManager的子类SortShuffleManager
val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)

2、创建SortShuffleManager

这个类比较重要的方法

2.1、registerShuffle,作用是注册不同的handle,然后运行不同的shuffle机制;

override def registerShuffle[K, V, C](
      shuffleId: Int,
      numMaps: Int,
      dependency: ShuffleDependency[K, V, C]): ShuffleHandle = {
    if (SortShuffleWriter.shouldBypassMergeSort(conf, dependency)) {
      // If there are fewer than spark.shuffle.sort.bypassMergeThreshold partitions and we don't
      // need map-side aggregation, then write numPartitions files directly and just concatenate
      // them at the end. This avoids doing serialization and deserialization twice to merge
      // together the spilled files, which would happen with the normal code path. The downside is
      // having multiple files open at a time and thus more memory allocated to buffers.
      //如果partitions的数目比spark.shuffle.sort.bypassMergeThreshold(默认200)参数设置的阈值少,
      //并且我们也不需要做map端的聚合操作(跟写代码的时候调用的算子有关系),
      //然后就直接写numPartitions个文件,最后把这些文件合并成一个
      //(归并排序-->归并排序,并不是原地排序算法,最后一次合并会占用大约2倍的空间,这点要注意,在分析Shuffle使用内存的时候要考虑到),
    //可以避免在合并溢写文件的时候进行两次的序列化和反序列化(?),正常的代码路径会反正这个情况(BaseShuffleHandle)
    //确定是一次会打开多个文件,因此需要分配给buffer更多的内存
      new BypassMergeSortShuffleHandle[K, V](
        shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]])
    } else if (SortShuffleManager.canUseSerializedShuffle(dependency)) {
      // Otherwise, try to buffer map outputs in a serialized form, since this is more efficient:
      new SerializedShuffleHandle[K, V](
        shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]])
    } else {
      // Otherwise, buffer map outputs in a deserialized form:
      new BaseShuffleHandle(shuffleId, numMaps, dependency)
    }
  }

代码解读:
1、判断条件是什么?
shouldBypassMergeSort
此方面里面2个判断,第一不能是在map端进行aggregation的算子,第二partition要比spark.shuffle.sort.bypassMergeThreshold设置的阈值小
也就是说,如果想要使用BypassMergeSortShuffle机制,那么这个算子不能是支持map端预聚合的算子并且Task的分区数要小于spark.shuffle.sort.bypassMergeThreshold设置的值(默认200)
canUseSerializedShuffle
在不需要map端聚合、partition数量小于16777216，Serializer支持relocation的情况下
2、三种shuffle机制,区别是什么?
BypassMergeSortShuffleWriter:如果reduce的partitions过多,使用这种shuffle机制是比较低效的,因为它同时为所有分区打开了单独的序列化器和文件流,所以,当符合下面3个条件的时候SortShuffleManager才会选择这种机制
i: 不需要排序
ii:不需要做聚合操作
iii:partition的数量比spark.shuffle.sort.bypassMergeThreshold设置的阈值小(默认200)
3、生成多少个文件
每个 task 最终会生成一份磁盘文件和一份索引文件，索引文件中标示了下游每个 task 的数据在文件中的 start offset 和 end offset。

private[spark] object SortShuffleWriter {
  def shouldBypassMergeSort(conf: SparkConf, dep: ShuffleDependency[_, _, _]): Boolean = {
    // We cannot bypass sorting if we need to do map-side aggregation.
    if (dep.mapSideCombine) {
      false
    } else {
      val bypassMergeThreshold: Int = conf.getInt("spark.shuffle.sort.bypassMergeThreshold", 200)
      dep.partitioner.numPartitions <= bypassMergeThreshold
    }
  }
}

def canUseSerializedShuffle(dependency: ShuffleDependency[_, _, _]): Boolean = {
    val shufId = dependency.shuffleId
    val numPartitions = dependency.partitioner.numPartitions
    if (!dependency.serializer.supportsRelocationOfSerializedObjects) {
      log.debug(s"Can't use serialized shuffle for shuffle $shufId because the serializer, " +
        s"${dependency.serializer.getClass.getName}, does not support object relocation")
      false
    } else if (dependency.mapSideCombine) {
      log.debug(s"Can't use serialized shuffle for shuffle $shufId because we need to do " +
        s"map-side aggregation")
      false
    } else if (numPartitions > MAX_SHUFFLE_OUTPUT_PARTITIONS_FOR_SERIALIZED_MODE) {
      log.debug(s"Can't use serialized shuffle for shuffle $shufId because it has more than " +
        s"$MAX_SHUFFLE_OUTPUT_PARTITIONS_FOR_SERIALIZED_MODE partitions")
      false
    } else {
      log.debug(s"Can use serialized shuffle for shuffle $shufId")
      true
    }
  }

2.2、getReader

当我们写的shuffle类的算子,在reduce阶段,读shuffle数据的时候调用这个方法来读

override def getReader[K, C](handle: ShuffleHandle, startPartition: Int, endPartition: Int, context: TaskContext): ShuffleReader[K, C] = {
        new BlockStoreShuffleReader(handle.asInstanceOf[BaseShuffleHandle[K, _, C]], startPartition, endPartition, context)
    }

2.3、getWriter

当我们写的shuffle类的算子,在map阶段,写shuffle数据的时候调用这个方法来写,其实是调用的Writer方法

override def getWriter[K, V](handle: ShuffleHandle, mapId: Int, context: TaskContext): ShuffleWriter[K, V] = {
        numMapsForShuffle.putIfAbsent(handle.shuffleId, handle.asInstanceOf[BaseShuffleHandle[_, _, _]].numMaps)
        val env = SparkEnv.get
        handle match {
            //在不需要map端聚合、partition数量小于16777216，Serializer支持relocation的情况下
            case unsafeShuffleHandle: SerializedShuffleHandle[K@unchecked, V@unchecked] => new UnsafeShuffleWriter(env.blockManager,
                shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver], context.taskMemoryManager(), unsafeShuffleHandle, mapId, context,
                env.conf)
            
            // 在不需要map端聚合、partition数量小于200的情况返回BypassMergeSortShuffleHandle对象
            case bypassMergeSortHandle: BypassMergeSortShuffleHandle[K@unchecked, V@unchecked] => new BypassMergeSortShuffleWriter(env.blockManager,
                shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver], bypassMergeSortHandle, mapId, context, env.conf)
            
            //其他情况，通用情况
            case other: BaseShuffleHandle[K@unchecked, V@unchecked, _] => new SortShuffleWriter(shuffleBlockResolver, other, mapId, context)
        }
    }

最后附图一张