学习目标：
①从宏观上了解Job（Spark on YARN）提交的流程。（画图）
②了解Job在提交之后，进行任务的划分，Stage的划分，任务的调度的过程！
结合： 宽依赖，窄依赖，Stage，task , job
③了解整个Job在执行期间Driver和Executor之间的通信方式
④Shuffle （区别不同的shuffle）
Spark是如何实现Shuffle!1
不同的Shuffle的效率影响！
⑤Spark的内存管理 （只有统一内存管理，用什么GC回收器）

spark的通信框架

spark每个终端（endpoint）都会有以后收信箱（inbox），还可以选择启动多个发信信箱（outbox）,具体由需要发送给什么终端决定。

image.png

每个终端通过dispatcher，将信息发送给本地或者其他终端，
如果发送个其他终端则先将信息发送到outbox，之后通过transportClient将数据发送出去，由别的终端的transportServer接收后传输到inbox，被其他终端消费，依据是指向inbox的标识nettyRpcEndpointRef
image.png

spark的app部署模式

SparkOnYarnClient提交流程：

image.png

client模式的源码解析：

  1. <dependency>
  2.             <groupId>org.apache.spark</groupId>
  3.             <artifactId>spark-yarn_2.12</artifactId>
  4.             <version>3.0.0</version>
  5.         </dependency>
  7.  spark-submit  --master yarn    --xxx xx  com.atguigu.spark.WordCount  /input
  8. -----------------------------------
  9. 第一部分： 客户端和YARN通信
  11. org.apache.spark.deploy.SparkSubmit.main
  12.     -- submit.doSubmit(args)
  13.         --super.doSubmit(args)
  14.             // 解析spark-submit后传入的参数，加载spark默认的参数
  15.             -- val appArgs = parseArguments(args)  
  16.             --submit(appArgs, uninitLog)
  17.                 --doRunMain()
  18.                 //运行spark-submit 提交的主类 中的main方法
  19.                     --runMain(args, uninitLog)
  21.                     // 如果deployMode == CLIENT，此时childMainClass=自己提交的全类名
  22.                     // 如果deployMode == CLUSTER,是YARN集群  childMainClass=org.apache.spark.deploy.yarn.YarnClusterApplication
  23.                         --val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)
  25.                         //创建childMainClass的Class类型
  26.                         --mainClass = Utils.classForName(childMainClass)
  29.                          --val app: SparkApplication = if (classOf[SparkApplication].isAssignableFrom(mainClass)) {
  31.                           mainClass.getConstructor().newInstance().asInstanceOf[SparkApplication]
  32.                         } else {
  33.                          //如果是client 模式，此时运行下一行
  34.                           new JavaMainApplication(mainClass)
  35.                             }
  38.                             //JavaMainApplication.start
  39.                          -- app.start(childArgs.toArray, sparkConf)
  41.                             //获取用户编写的 app类的main
  42.                             --  val mainMethod = klass.getMethod("main", new Array[String](0).getClass)
  44.                             //执行main方法，Driver启动
  45.                             --mainMethod.invoke(null, args)
  47. ------------------------------------------------------------------------
  48. 第二部分： WordCount.main
  50.   SparkContext sc=new SparkContext
  52.   SparkContext的重要组件：
  53.           private var _taskScheduler: TaskScheduler = _   //负责Task的调度
  55.           private var _dagScheduler: DAGScheduler = _   // Job中DAG Stage的切分
  57.           private var _env: SparkEnv = _ //  RpcEnv(通信环境) BlockManager(存数据)
  60. --------------
  61. _taskScheduler = ts
  62.     _taskScheduler.start()
  63.         //YarnClientSchedulerBackend.start()
  64.         --backend.start()
  65.             --   client = new Client(args, conf, sc.env.rpcEnv)
  67.             //提交应用程序到YARN
  68.             --   bindToYarn(client.submitApplication(), None)
  70.             //参考cluster模式
  71.             --......
  73.             --在容器中运行的AM的全类名： org.apache.spark.deploy.yarn.ExecutorLauncher
  76. -----------------------------
  77. 第二部分：启动AM 
  78.  org.apache.spark.deploy.yarn.ExecutorLauncher.main
  79.           //ExecutorLauncher 是对AM的封装，也是AM的实现
  80.          ApplicationMaster.main(args)
  81.             -- --master.run()
  82.                if (isClusterMode) {
  84.                     --runDriver()
  85.                   } else {
  86.                     //client模式运行
  87.                     runExecutorLauncher()   //申请Container启动Executor
  88.                   }
  97. ----------------------------------------------------------------------------
  99.                          // YarnClusterApplication.start()  Client是可以和yarn进行通信的一个客户端
  100.                             --new Client(new ClientArguments(args), conf, null).run()
  101.                                 --this.appId = submitApplication()
  103.                                     // 提交应用程序到YARN上，获取YARN的返回值
  104.                                     -- val newApp = yarnClient.createApplication()
  106.                                     //确定app保存临时数据的作业目录，通常是在hdfs中的/tmp目录中生成一个子目录
  107.                                     -- val appStagingBaseDir = sparkConf.get(STAGING_DIR)
  110.                                     //确保YARN有足够的资源运行当前app的 AM
  111.                                     // 如果集群资源不足，此时会阻塞，一直阻塞到超时，会FAILD
  112.                                     -- verifyClusterResources(newAppResponse)
  115.                                     --  val containerContext = createContainerLaunchContext(newAppResponse)
  117.                                         // 确定AM的主类名
  118.                                         // AM是YARN提供的一个接口
  119.                                         // 任何的应用程序如果希望提交APP到YARN，实现AM的接口
  120.                                         //  MR写的app，MR实现AM
  121.                                         //  spark写的App，spark实现AM
  122.                                         --val amClass =
  123.                                           if (isClusterMode) {
  124.                                                 //集群模式AM的全类名是org.apache.spark.deploy.yarn.ApplicationMaster
  125.                                             Utils.classForName("org.apache.spark.deploy.yarn.ApplicationMaster").getName
  126.                                           } else {
  127.                                             Utils.classForName("org.apache.spark.deploy.yarn.ExecutorLauncher").getName
  128.                                           }
  129.                                     --  val appContext = createApplicationSubmissionContext(newApp, containerContext)
  132.                                     //提交运行AM
  133.                                     --  yarnClient.submitApplication(appContext)

SparkOnYarnCluster提交流程架构图示：

image.png

image.png

流程源码概述：

1、客户端与yarn通信：
1）通过：spark-submit --master yarn --deploy-mode cluster --xxx xx com.atguigu.spark.WordCount /input指令
2）启动执行org.apache.spark.deploy.SparkSubmit.main进程，启動client线程，在客户端中提交申请应用给resouremanage
3）之后resourceManager找到一台nodemanager
2、AM进程启动
image.png
1）創建amContainer并在其中执行ApplicationMaster.main.run 方法，启动一个线程来运行用户自己编写的app应用程序，并起名为“Driver”
2）等用户的app运行完毕，获取用户自己创建的sparkContext，通過獲取sparkContext的參數向RM汇报AM启动成功（registerAM，其实是注册）并申请Executor的资源：container
3）之后RM返回其资源列表给appmaster，master获取到已经分配的container，在其中运行进程，每个container都会创建一个 ExecutorRunnable，交给线程池运行
4）org.apache.spark.executor.YarnCoarseGrainedExecutorBackend就是那个运行的进程，這個進程就是我們的executor
3、启动YarnCoarseGrainedExecutorBackend进程（executor,负责通信）：
1）进程中向driver请求spark配置信息：new SparkConf，与driver产生通信
2）并且给当前进程在网络中起了一个“Executor”别名，这里的负责通信的executor和下面负责计算的executor都是图中所说的executor，只是负责的功能不一样
其中：进程YarnCoarseGrainedExecutorBackend别名就是executor，负责通信
而这个进程类中，就有个属性 val executor =null ；负责计算
下图所框出来的起始就是负责通信的executor
image.pngimage.png
3）之后该进程向Dirver发送RegisterExecutor的反向注册信息，Driver接收到注册信息后返回注册成功信息
4）注册成功过后，YarnCoarseGrainedExecutorBackend.receive方法接收到返回信息，开始调用父类GrainedExecutorBackend的Executor，（spark的计算者，他维护 了一个线程池，用来计算各种Task，但需要等待driver发送task过来才能运行。）之后返回executor启动成功的信息给driver
5）之后driver线程调用DriverEndPoint.receive方法处理过来的信息，并向Executor调度工作任务makeOffers（上图的发offers，就是指派任务的意思）：（Task）
task如何发送：
//从调度池（TaskSet）中，根据TaskSet的优先级，调度其中的Task。
// 按照轮流发送的原则将Task调度到所有的Executor，保证负载均衡。
//调度应该发往次Executor的Task
6）之后反序列化task，然后创建task运行线程new TaskRunner(context, taskDescription)执行task：
ResultTask
ShufflerTask

cluster模式的源码解析：

  1.  <dependency>
  2.             <groupId>org.apache.spark</groupId>
  3.             <artifactId>spark-yarn_2.12</artifactId>
  4.             <version>3.0.0</version>
  5.         </dependency>
  7.  spark-submit  --master yarn  --deploy-mode cluster  --xxx xx  com.atguigu.spark.WordCount  /input
  8. -----------------------------------
  9. 第一部分： 客户端和YARN通信
  11. org.apache.spark.deploy.SparkSubmit.main
  12.     -- submit.doSubmit(args)
  13.         --super.doSubmit(args)
  14.             // 解析spark-submit后传入的参数，加载spark默认的参数
  15.             -- val appArgs = parseArguments(args)  
  16.             --submit(appArgs, uninitLog)
  17.                 --doRunMain()
  18.                 //运行spark-submit 提交的主类 中的main方法
  19.                     --runMain(args, uninitLog)
  21.                     // 如果deployMode == CLIENT，此时childMainClass=自己提交的全类名
  22.                     // 如果deployMode == CLUSTER,是YARN集群  childMainClass=org.apache.spark.deploy.yarn.YarnClusterApplication
  23.                         --val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)
  25.                         //创建childMainClass的Class类型
  26.                         --mainClass = Utils.classForName(childMainClass)
  29.                          --val app: SparkApplication = if (classOf[SparkApplication].isAssignableFrom(mainClass)) {
  30.                          //如果是cluster 模式，此时运行下一行
  31.                           mainClass.getConstructor().newInstance().asInstanceOf[SparkApplication]
  32.                         } else {
  33.                           new JavaMainApplication(mainClass)
  34.                             }
  36.                          -- app.start(childArgs.toArray, sparkConf)
  38.                          // YarnClusterApplication.start()  Client是可以和yarn进行通信的一个客户端
  39.                             --new Client(new ClientArguments(args), conf, null).run()
  40.                                 --this.appId = submitApplication()
  42.                                     // 提交应用程序到YARN上，获取YARN的返回值
  43.                                     -- val newApp = yarnClient.createApplication()
  45.                                     //确定app保存临时数据的作业目录，通常是在hdfs中的/tmp目录中生成一个子目录
  46.                                     -- val appStagingBaseDir = sparkConf.get(STAGING_DIR)
  49.                                     //确保YARN有足够的资源运行当前app的 AM
  50.                                     // 如果集群资源不足，此时会阻塞，一直阻塞到超时，会FAILD
  51.                                     -- verifyClusterResources(newAppResponse)
  54.                                     --  val containerContext = createContainerLaunchContext(newAppResponse)
  56.                                         // 确定AM的主类名
  57.                                         // AM是YARN提供的一个接口
  58.                                         // 任何的应用程序如果希望提交APP到YARN，实现AM的接口
  59.                                         //  MR写的app，MR实现AM
  60.                                         //  spark写的App，spark实现AM
  61.                                         --val amClass =
  62.                                           if (isClusterMode) {
  63.                                                 //集群模式AM的全类名是org.apache.spark.deploy.yarn.ApplicationMaster
  64.                                             Utils.classForName("org.apache.spark.deploy.yarn.ApplicationMaster").getName
  65.                                           } else {
  66.                                             Utils.classForName("org.apache.spark.deploy.yarn.ExecutorLauncher").getName
  67.                                           }
  68.                                     --  val appContext = createApplicationSubmissionContext(newApp, containerContext)
  71.                                     //提交运行AM
  72.                                     --  yarnClient.submitApplication(appContext)
  75. ----------------------------------------
  76. 第二部分  AM启动
  77.     ApplicationMaster.main    
  78.       --master.run()
  79.         --if (isClusterMode) {
  80.                 //cluster模式运行
  81.             --runDriver()
  82.           } else {
  83.             runExecutorLauncher()
  84.           }
  87.           // 启动一个线程，运行用户自己编写的app应用程序的main方法
  88.           --userClassThread = startUserApplication()
  90.                 //为线程起名Driver
  91.                 --userThread.setName("Driver")
  93.           //等待用户自己编写的app的main方法运行结束后，获取用户自己创建的SparkContext
  94.           --val sc = ThreadUtils.awaitResult(sparkContextPromise.future,
  95.                     Duration(totalWaitTime, TimeUnit.MILLISECONDS))
  97.           // 向RM回报，AM已经启动成功了
  98.            --registerAM(host, port, userConf, sc.ui.map(_.webUrl), appAttemptId)
  100.           // 创建一个可以向RM申请资源的对象，申请资源启动Executor
  101.           -- createAllocator(driverRef, userConf, rpcEnv, appAttemptId, distCacheConf)
  103.             //allocator: YarnAllocator 作用就是向RM申请Container，决定获取到COntainer后，使用Container干什么
  104.             --allocator = client.createAllocator()
  108.             --allocator.allocateResources()
  109.                     //发送申请请求
  110.                 -- val allocateResponse = amClient.allocate(progressIndicator)
  112.                         //从YARN的响应中获取已经分配的Container
  113.                 --val allocatedContainers = allocateResponse.getAllocatedContainers()
  116.                 //处理申请到的Container
  117.                 -- handleAllocatedContainers(allocatedContainers.asScala)
  119.                     //在Container中运行进程
  120.                     -- runAllocatedContainers(containersToUse)
  122.                         //每个Container都会创建一个ExecutorRunnable，交给线程池运行
  123.                         --new ExecutorRunnable.run
  125.                             -- nmClient = NMClient.createNMClient()
  126.                                 nmClient.init(conf)
  127.                                 nmClient.start()
  129.                                 //准备容器中的启动进程
  130.                                 startContainer()
  131.                                   --org.apache.spark.executor.YarnCoarseGrainedExecutorBackend 容器中启动的进程
  132.                                     --val commands = prepareCommand()
  137. org.apache.spark.executor.YarnCoarseGrainedExecutorBackend
  138.     ①本身就是一个通信端点
  139.     ②var driver: Option[RpcEndpointRef] = None
  140.             RpcEndpointRef： 某个通信端点的引用 
  141.                                     类比为网络中的一个通讯设备都有唯一的电话号
  142. ------------------
  143. 第一部分： YarnCoarseGrainedExecutorBackend进程启动
  144. --YarnCoarseGrainedExecutorBackend.main
  145.     -- CoarseGrainedExecutorBackend.run(backendArgs, createFn)
  147.             //向Driver请求spark的配置信息
  148.           --val executorConf = new SparkConf
  150.             // 当前进程在网络中有一个通信的别名称为Executor
  151.          -- env.rpcEnv.setupEndpoint("Executor",
  152.            backendCreateFn(env.rpcEnv, arguments, env, cfg.resourceProfile))
  154.           //阻塞，一直运行，除非Driver发送停止命令
  155.           --env.rpcEnv.awaitTermination()
  163.  constructor -> onStart -> receive* -> onStop
  164. --------------------------------------
  165. 第二部分： YarnCoarseGrainedExecutorBackend进程向Driver发送注册请求
  166. onStart
  167.      // ref:RpcEndpointRef   代表Driver的通信端点引用(电话号)
  168.      // 向Driver发RegisterExecutor消息，请求答复一个Boolean的值
  169.     -- ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores, extractLogUrls,
  170.         extractAttributes, _resources, resourceProfile.id))
  171.     }(ThreadUtils.sameThread).onComplete {
  172.       case Success(_) =>
  173.       //成功，自己给自己发 RegisteredExecutor消息
  174.         self.send(RegisteredExecutor)
  175.         //失败，进程就退出
  176.       case Failure(e) =>
  177.         exitExecutor(1, s"Cannot register with driver: $driverUrl", e, notifyDriver = false)
  187. -------------------------------------
  188. 第三部分： Driver处理注册请求
  189.     DriverEndPoint.receiveAndReply
  190.         --case RegisterExecutor
  191.             -- ①判断是否已经注册过了，如果注册过了，回复失败； 判断是否拉黑，拉黑，注册失败，
  192.             -- 否则注册，注册完成后，回复true
  193.             context.reply(true)
  198. ----------------------------------------
  199. 第四部分： YarnCoarseGrainedExecutorBackend 注册成功后
  201. YarnCoarseGrainedExecutorBackend.receive
  202.     -- case RegisteredExecutor =>
  203.        logInfo("Successfully registered with driver")
  204.       try {
  205.         // Spark的计算者，维护了一个线程池，用来计算各种Task
  206.         executor = new Executor(executorId, hostname, env, userClassPath, isLocal = false,
  207.           resources = _resources)
  209.           // 给Driver发LaunchedExecutor消息
  210.         driver.get.send(LaunchedExecutor(executorId))
  212. -----------------------------------------------------
  213. 第五部分： Driver处理LaunchedExecutor请求
  214.      DriverEndPoint.receive
  215.         -- case LaunchedExecutor(executorId) =>
  216.             executorDataMap.get(executorId).foreach { data =>
  217.           data.freeCores = data.totalCores
  218.         }
  219.         //Driver向Executor发送工作任务（Task）
  220.         makeOffers(executorId)
  221.                 //从调度池中，根据TaskSet的优先级，调度其中的Task。
  222.                 // 按照轮流发送的原则将Task调度到哦所有的Executor，保证负载均衡。
  223.                 //调度应该发往次Executor的Task
  224.             --scheduler.resourceOffers(workOffers)
  226.                 -- launchTasks(taskDescs)
  227.                     -- val serializedTask = TaskDescription.encode(task)
  229.                     //为要发送的Executor打个招呼，发送Task  
  230.                     --executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
  233. --------------------------------------------------
  234. 第六部分: Executor收到Task
  235.     -- case LaunchTask(data) =>
  236.       if (executor == null) {
  237.         exitExecutor(1, "Received LaunchTask command but executor was null")
  238.       } else {
  239.       // 反序列化得到Task的描述(TaskDescription)
  240.         val taskDesc = TaskDescription.decode(data.value)
  241.         logInfo("Got assigned task " + taskDesc.taskId)
  242.         taskResources(taskDesc.taskId) = taskDesc.resources
  244.         //启动Task的运算
  245.         executor.launchTask(this, taskDesc)
  247.             //创建一个Task运行的线程
  248.             -- val tr = new TaskRunner(context, taskDescription)
  249.                 runningTasks.put(taskDescription.taskId, tr)
  251.                 // 启动线程
  252.                 threadPool.execute(tr)
  254.                     --TaskRunner.run()
  255.                         //构造可以运行的Task对象
  256.                         --task = ser.deserialize[Task[Any]]
  258.                         //运行Task获取结果
  259.                         -- val res = task.run()
  261.                             // ShuffleMapTask  或  ResultTask
  262.                             --runTask(context)
  264.       }

job的stage的task的调度

  • Application: 应用
    Job: 任务 .—>一个行动算子一个job
     stage: 阶段
    task: 单个任务
     一个SparkContext对应一个Application
    一个Application对应多个Job,一个action（行动算子）产生一个job
    image.png
     一个job中对应多个stage,stage的个数 = shuffle个数（shufflerMapStage） + 1（ResultStage），也就是说一个action对应一个resultStage
    * 一个stage对应多个task，task的个数 = 当前stage中最后一个rdd的分区数
    一个task也对应多个stage
    只有Executor的代码会产生job，Driver端执行的代码不走job
    stage与task是交叉关系
    下图是一个job：
    image.png

    task源码流程概述

    image.png
    在driver中：
    rdd.collect处理各分区的task汇聚成job
    通过dagScheduler.handleJobSubmitted方法，开始创建stage
    递归依次从后往前找，将ResultStage的所有祖先 ShuffleMapStage都创建完毕，最后才创建最后一个Stage：finalStage = createResultStage
    然后提交ResultStage、ShuffleMapStage。
    之后根据Stage类型创建对应的Task，（ShuffleMapStage =>new ShuffleMapTask，ResultStage =>new ResultTask）
    task个数根据当前Stage要计算的分区数（最后一个stage的分区数）确定

task在hadoop的mr里和在spark的区别

hadoop中：maptask、reducetask：进程级别
spark中：ShuffleMapTask 或ResultTask:线程级别

task调度源码解释

  1. Task的产生
  3. --------------
  4. RDD.collect
  5.     --val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
  7.     //判断是否存在闭包，闭包变量是否可以序列化
  8.     -- val cleanedFunc = clean(func)
  11.     -- dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
  13.     -- val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
  15.         // event处理循环，放入一个事件 JobSubmitted ，提交job
  16.         //  eventProcessLoop = new DAGSchedulerEventProcessLoop
  17.         -- eventProcessLoop.post(JobSubmitted())
  18.             --EventLoop.onReceive(event)
  19.                 -- doOnReceive(event)
  20.                     --case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>
  21.                         dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)
  24. ----------------------
  25.  dagScheduler.handleJobSubmitted
  26.        // Job的最后一个阶段stage
  27.       --var finalStage: ResultStage = null
  29.       -- finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
  31.         -- 递归依次从后往前找，将ResultStage的所有祖先 ShuffleMapStage都创建完毕，最后才创建最后一个Stage
  32.         -- val parents = getOrCreateParentStages(rdd, jobId)
  34.         --val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
  36.         // ResultStage创建完毕后，才会生成Job
  37.         -- val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)
  39.         //提交ResultStage
  40.         --submitStage(finalStage)
  42.             //查看当前阶段有没有未提交的父阶段，按照父阶段的ID排序
  43.             --val missing = getMissingParentStages(stage).sortBy(_.id)
  45.                 //如果父阶段未提交，按照顺序依次提交，最后提交自己
  46.                 --if (missing.isEmpty) {
  47.                   logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
  48.                   submitMissingTasks(stage, jobId.get)
  49.                 } else {
  50.                   for (parent <- missing) {
  51.                     submitStage(parent)
  52.                   }
  53.                   waitingStages += stage
  54.                 }
  57.                     // 根据Stage类型创建对应的Task，创建时，根据当前Stage要计算的分区数确定
  58.                     //  如何知道每一个Stage要计算的partition的数量？   取决于Stage最后一个RDD的分区数
  59.                     --val tasks: Seq[Task[_]] = try {
  60.                           val serializedTaskMetrics = closureSerializer.serialize(stage.latestInfo.taskMetrics).array()
  61.                           stage match {
  62.                             case stage: ShuffleMapStage =>
  63.                               stage.pendingPartitions.clear()
  64.                               partitionsToCompute.map { id =>
  65.                                 val locs = taskIdToLocations(id)
  66.                                 val part = partitions(id)
  67.                                 stage.pendingPartitions += id
  68.                                 new ShuffleMapTask(stage.id, stage.latestInfo.attemptNumber,
  69.                                   taskBinary, part, locs, properties, serializedTaskMetrics, Option(jobId),
  70.                                   Option(sc.applicationId), sc.applicationAttemptId, stage.rdd.isBarrier())
  71.                               }
  73.                             case stage: ResultStage =>
  74.                               partitionsToCompute.map { id =>
  75.                                 val p: Int = stage.partitions(id)
  76.                                 val part = partitions(p)
  77.                                 val locs = taskIdToLocations(id)
  78.                                 new ResultTask(stage.id, stage.latestInfo.attemptNumber,
  79.                                   taskBinary, part, locs, id, properties, serializedTaskMetrics,
  80.                                   Option(jobId), Option(sc.applicationId), sc.applicationAttemptId,
  81.                                   stage.rdd.isBarrier())
  82.                               }
  83.                           }
  85.                           //将每个Stage的task放入调度池，通知Driver，去调度
  86.                           // TaskSet： 包含了一个Stage所有的Tasks, 一个Task的集合
  87.                     --taskScheduler.submitTasks(new TaskSet(
  88.                         tasks.toArray, stage.id, stage.latestInfo.attemptNumber, jobId, properties))
  90.                         //TaskSetManager： 追踪每一个TaskSet中所有Task的运行情况，在它们失败时重启。通过延迟调度
  91.                         // 处理位置敏感的调度，调度每个Task
  92.                            --val manager = createTaskSetManager(taskSet, maxTaskFailures)
  95.                            //将TaskSetManager加入调度池(FIFO | FAIR),不同的调度池，会决定不同的TaskSet被调度的优先级
  96.                            -- schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
  98.                            -- backend.reviveOffers()
  99.                             //给Driver发消息，告诉Driver有新的TaskSetManager入池了，可以开始发送了
  100.                             -- driverEndpoint.send(ReviveOffers)
  101.                                 -- case ReviveOffers =>
  102.                                                 makeOffers()

shuffler

什么是shuffler

在Spark中，很多算子都会引起RDD中数据的重分区！新的分区被创建，旧的分区被合并或数据被重新分配！在重分区的过程中，如果数据发生了跨节点的移动，就称为shuffle！

shuffle的作用就是为了在不同的task中交换数据！

hashshuffler（过时）

对应hadoop中的MR来解释

每个maptask（spark task）的数据会配分配到多个reducetask中，
N个maptask（spark task）分配到M个reducetask中就是生成：M*N个小文件
image.png

优化后的hashshuffler

原理：
core number: M
将所有task的内容写进一个core中
reduceTask个数：N
M*N个小文件

应用：
启合并机制的配置是spark.shuffle.consolidateFiles。该参数默认值为false，将其设置为true即可开启优化机制。

image.png

SortShuffle（sort-based shuffle）

随着Hadoop2.0的发布，Spark借鉴了Hadoop2.0中的shuffle过程（sort-based shuffle）！

本质：
有由索引实现排序

sort-based shuffle核心要义：
1）一个MapTask最终只生成一个数据文件，
2）这一个数据文件中包含2个文件：
一个是包含有若干分区的数据文件，
另一个是包含索引的元数据文件，其中记录了分区的边界！
在产生文件时，默认SortShuffleWriter会进行排序！

3）并不是所有的sort-based shuffle都会对shuffle写出的数据进行排序！：

对比Hadoop2.0的shuffle：
在MapTask上，先分区———>溢写前，进行排序——->溢写为一个片段
所有片段全部溢写完成后———->merge———->合并为一个总的文件！

区别：
hadoop: MapTask: map————->sort————->merge
ReduceTask: sort————->reduce

spark : MapTask: map（各种算子）————->sort————->merge
ReduceTask: merge——->算子（reduce阶段不排序，算子中排序）

spark的shuffle在ReduceTask端不排序！
原因：
1）hadoop的reduce是一个迭代器，且只有一个map和reduce
如果有序：
(a,1),(a,1),(b,1)则输出(a,2),(b,1)
如果无序如：
(a,1),(b,1),(a,1) 则输出(a,1),(b,1),(a,1)

2）而spark会有一些算子实现排序：

  1.      - `mapPartitions` to sort each partition using, for example, `.sorted`
  2.      - `repartitionAndSortWithinPartitions` to efficiently sort partitions while simultaneously repartitioning
  3.      - `sortBy` to make a globally ordered RDD

普通SortShuffle

在该模式下，数据会先写入一个数据结构，reduceByKey写入Map，一边通过Map局部聚合，一遍写入内存。Join算子写入ArrayList直接写入内存中。然后需要判断是否达到阈值，如果达到就会将内存数据结构的数据写入到磁盘，清空内存数据结构。
在溢写磁盘前，先根据key进行排序，排序过后的数据，会分批写入到磁盘文件中。默认批次为10000条，数据会以每批一万条写入到磁盘文件。写入磁盘文件通过缓冲区溢写的方式，每次溢写都会产生一个磁盘文件，也就是说一个Task过程会产生多个临时文件。
最后在每个Task中，将所有的临时文件合并，这就是merge过程，此过程将所有临时文件读取出来，一次写入到最终文件。意味着一个Task的所有数据都在这一个文件中。同时单独写一份索引文件，标识下游各个Task的数据在文件中的索引，start offset和end offset。
image.png

bypass SortShuffle

bypass运行机制的触发条件如下：
1) shuffle reduce task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值，默认为200。
2) 不是聚合类的shuffle算子（比如reduceByKey）。
此时task会为每个reduce端的task都创建一个临时磁盘文件，并将数据按key进行hash然后根据key的hash值，将key写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的。最后，同样会将所有临时磁盘文件都合并成一个磁盘文件，并创建一个单独的索引文件。
该过程的磁盘写机制其实跟优化后的的HashShuffleManager是一模一样的，只多了创建的索引文件，通过索引文件，节省了排序功能，提升性能
而该机制与普通SortShuffleManager运行机制的不同在于：不会进行排序。也就是说，启用该机制的最大好处在于，shuffle write过程中，不需要进行数据的排序操作，也就节省掉了这部分的性能开销。
image.png