<dependency>
　　<groupId>org.apache.spark</groupId>
　　<artifactId>spark-yarn_2.12</artifactId>
　　<version>3.0.0</version>
</dependency>
<dependency>
　　<groupId>org.apache.spark</groupId>
　　<artifactId>spark-core_2.12</artifactId>
　　<version>3.0.0</version>
</dependency>

private[spark] trait RpcEndpoint
private[spark] trait IsolatedRpcEndpoint extends RpcEndpointt

//Executor的通信终端
private[spark] class CoarseGrainedExecutorBackend(
    override val rpcEnv: RpcEnv,
    driverUrl: String,
    executorId: String,
    bindAddress: String,
    hostname: String,
    cores: Int,
    userClassPath: Seq[URL],
    env: SparkEnv,
    resourcesFileOpt: Option[String],
    resourceProfile: ResourceProfile)
  extends IsolatedRpcEndpoint with ExecutorBackend with Logging

//Driver的通信终端
class DriverEndpoint extends IsolatedRpcEndpoint with Logging

// 分发处理消息
private val dispatcher: Dispatcher = new Dispatcher(this, numUsableCores)

// 网络传输通信的服务端
private var server: TransportServer

// 当前环境中发向每个 远程设备的 发件箱   RpcAddress（远程设备） 
private val outboxes = new ConcurrentHashMap[RpcAddress, Outbox]()

/*
        调用dispatcher.regist()
         为每个EndPoint创建一个Inbox，在Inbox中放入onStart()
*/
rpcEnv.setupEndpoint

NettyRpcEnv.setUpEndPoint("设备名",设备)

// 分发消息
private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) extends Logging {

    // 每个设备 及其  MessageLoop
  private val endpoints: ConcurrentMap[String, MessageLoop] =
    new ConcurrentHashMap[String, MessageLoop]
    
    // 记录在当前RpcEnv中注册的所有设备及其引用
  private val endpointRefs: ConcurrentMap[RpcEndpoint, RpcEndpointRef] =
    new ConcurrentHashMap[RpcEndpoint, RpcEndpointRef]

// Executor  计算器，计算者，负责运行task，例如启动，杀死
var executor: Executor = null

//数据准备
def main(args: Array[String]): Unit = {
    //创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("TestSparkRDD").setMaster("local[*]")
    //创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd = sc.makeRDD(List(1,2,3,4))
    rdd.collect()

    //关闭连接
    sc.stop()
  }

private[scheduler] abstract class Stage(
   val id: Int,    // stage的id，后续调度时，ID小的stage一般是靠前的，会优先执行
    val rdd: RDD[_], // stage的最后一个RDD
    val numTasks: Int,  //在 ShuffleMapStage 等于 当前Stage的最大分区数，在ResultStage，取决于使用的算子
    val parents: List[Stage],  // 父Stage
    val firstJobId: Int,     // 运行的第一个Job的Id
    val callSite: CallSite)
  extends Logging 

private[spark] class TaskSetManager(
    sched: TaskSchedulerImpl,
    val taskSet: TaskSet,
    val maxTaskFailures: Int,
    blacklistTracker: Option[BlacklistTracker] = None,
    clock: Clock = new SystemClock()) extends Schedulable with Logging

private[spark] trait SchedulableBuilder {
  def rootPool: Pool
  def buildPools(): Unit
  def addTaskSetManager(manager: Schedulable, properties: Properties): Unit
}

def initialize(backend: SchedulerBackend): Unit = {
  this.backend = backend
  schedulableBuilder = {
    schedulingMode match {
        // 由调度模式决定 取决于spark.scheduler.mode，默认为FIFO
      case SchedulingMode.FIFO =>
        new FIFOSchedulableBuilder(rootPool)
      case SchedulingMode.FAIR =>
        new FairSchedulableBuilder(rootPool, conf)
      case _ =>
        throw new IllegalArgumentException(s"Unsupported $SCHEDULER_MODE_PROPERTY: " +
        s"$schedulingMode")
    }
  }
  schedulableBuilder.buildPools()
}

private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {
    // TaskSetManager: Schedulable
  override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
    val priority1 = s1.priority
    val priority2 = s2.priority
      /*
      根据  (priority1 - priority2) 的结果，进行返回
              结果为负数，返回 -1
              结果为0，返回0
              结果为正数，返回 1
      */
    var res = math.signum(priority1 - priority2)
    if (res == 0) {
        // stageId 小的是靠前的阶段
      val stageId1 = s1.stageId
      val stageId2 = s2.stageId
      res = math.signum(stageId1 - stageId2)
    }
    res < 0
  }
}

private[spark] abstract class ShuffleWriter[K, V] {
  /** Write a sequence of records to this task‘s output */
  @throws[IOException]
  def write(records: Iterator[Product2[K, V]]): Unit

  /** Close this writer, passing along whether the map completed */
  def stop(success: Boolean): Option[MapStatus]
}

override def write(records: Iterator[Product2[K, V]]): Unit = {
    
    // SortShuffleWriter 支持map端聚合，可选！
  sorter = if (dep.mapSideCombine) {
    new ExternalSorter[K, V, C](
      context, dep.aggregator, Some(dep.partitioner), dep.keyOrdering, dep.serializer)
  } else {
    // In this case we pass neither an aggregator nor an ordering to the sorter, because we don‘t
    // care whether the keys get sorted in each partition; that will be done on the reduce side
    // if the operation being run is sortByKey.
    new ExternalSorter[K, V, V](
      context, aggregator = None, Some(dep.partitioner), ordering = None, dep.serializer)
  }
    
    // 都创建了一个ExternalSorter
  sorter.insertAll(records)

  // Don‘t bother including the time to open the merged output file in the shuffle write time,
  // because it just opens a single file, so is typically too fast to measure accurately
  // (see SPARK-3570).
  val mapOutputWriter = shuffleExecutorComponents.createMapOutputWriter(
    dep.shuffleId, mapId, dep.partitioner.numPartitions)
  sorter.writePartitionedMapOutput(dep.shuffleId, mapId, mapOutputWriter)
    // 写出所有分区的数据  最终生成索引文件和数据文件
  val partitionLengths = mapOutputWriter.commitAllPartitions()
  mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths, mapId)
}

def insertAll(records: Iterator[Product2[K, V]]): Unit = {
  // TODO: stop combining if we find that the reduction factor isn‘t high
  val shouldCombine = aggregator.isDefined

  if (shouldCombine) {
    // Combine values in-memory first using our AppendOnlyMap
    val mergeValue = aggregator.get.mergeValue
    val createCombiner = aggregator.get.createCombiner
    var kv: Product2[K, V] = null
    val update = (hadValue: Boolean, oldValue: C) => {
      if (hadValue) mergeValue(oldValue, kv._2) else createCombiner(kv._2)
    }
    while (records.hasNext) {
      addElementsRead()
      kv = records.next()
        //  PartitionedAppendOnlyMap  将kv对聚合到map中
      map.changeValue((getPartition(kv._1), kv._1), update)
        // 判断是否该溢写
      maybeSpillCollection(usingMap = true)
    }
  } else {
    // Stick values into our buffer
    while (records.hasNext) {
      addElementsRead()
      val kv = records.next()
        // PartitionedPairBuffer 将kv对，放入buffer不聚合
      buffer.insert(getPartition(kv._1), kv._1, kv._2.asInstanceOf[C])
      maybeSpillCollection(usingMap = false)
    }
  }
}

override def runTask(context: TaskContext): MapStatus = {
 // 获取Shuffle的Writer
  dep.shuffleWriterProcessor.write(rdd, dep, mapId, context, partition)
}

def write(
    var writer: ShuffleWriter[Any, Any] = null
    try {
      val manager = SparkEnv.get.shuffleManager
        //获取writer
      writer = manager.getWriter[Any, Any](
        //写出
      writer.write(
        rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
      writer.stop(success = true).get
    } 
  }
}

//Spark使用G1，可以在SparkEnv中配置
-XX:+UseG1GC