SparkSql中多个Stage的并发执行

set hive.exec.parallel=true;              //打开任务并行执行
set hive.exec.parallel.thread.number=16;  //同一个sql允许最大并行度，默认为8。

set spark.dynamicAllocation.enabled=true; --动态资源开关
set spark.executor.cores=4; --每个Container分配的core数量
set spark.dynamicAllocation.maxExecutors=1200;  --executor最大申请个数
set spark.dynamicAllocation.initialExecutors=5;  -- 初始申请executor个数,默认等于minExecutors
set spark.dynamicAllocation.minExecutors =5; --executor最小申请个数

大概就是这几个参数，代表能并行4*1200 ，也就是4800个task。
把maxExecutors调大点，就能并行的更多

/** Submits stage, but first recursively submits any missing parents. */
  private def submitStage(stage: Stage): Unit = {
   //获取stage所属的active的JobId
    val jobId = activeJobForStage(stage)
    if (jobId.isDefined) {
      logDebug(s"submitStage($stage (name=${stage.name};" +
        s"jobs=${stage.jobIds.toSeq.sorted.mkString(",")}))")
      //非waiting、running、failed的stage
      if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
        //获取该stage未提交的父stages，并按stage id从小到大排序，也就是stage是从后往前提交的
        val missing = getMissingParentStages(stage).sortBy(_.id)
        logDebug("missing: " + missing)
        if (missing.isEmpty) {
          logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
          //若无未提交的父stage, 则提交该stage对应的tasks
          submitMissingTasks(stage, jobId.get)
        } else {
          //若存在未提交的父stage, 依次提交所有父stage (若父stage也存在未提交的父stage, 则提交, 依次类推)
          for (parent <- missing) {
            submitStage(parent)
          }
          //并把该stage添加到等待stage队列中
          waitingStages += stage
        }
      }
    } else {
      abortStage(stage, "No active job for stage " + stage.id, None)
    }
  }

//以参数stage为起点，向前遍历所有stage，判断stage是否为未提交，若使则加入missing中
private def getMissingParentStages(stage: Stage): List[Stage] = {
  val missing = new HashSet[Stage] //未提交的stage
  val visited = new HashSet[RDD[_]]  //存储已经被访问到得RDD
  // We are manually maintaining a stack here to prevent StackOverflowError
  // caused by recursively visiting
  val waitingForVisit = new ListBuffer[RDD[_]]
  waitingForVisit += stage.rdd
  def visit(rdd: RDD[_]): Unit = {
    if (!visited(rdd)) {
      visited += rdd
      val rddHasUncachedPartitions = getCacheLocs(rdd).contains(Nil)
      if (rddHasUncachedPartitions) {
        for (dep <- rdd.dependencies) {
          dep match {
             //若为宽依赖，生成新的stage
            case shufDep: ShuffleDependency[_, _, _] =>
            //根据shufDep.shuffleId获取对应的ShuffleMapStage
              val mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId)
              if (!mapStage.isAvailable || !mapStage.shuffleDep.shuffleMergeFinalized) {
                  //若stage得状态为available或者 基于push based shuffle合并没有完成，则为未提交stage
                missing += mapStage
              } else {
                // Forward the nextAttemptId if skipped and get visited for the first time.
                // Otherwise, once it gets retried,
                // 1) the stuffs in stage info become distorting, e.g. task num, input byte, e.t.c
                // 2) the first attempt starts from 0-idx, it will not be marked as a retry
                mapStage.increaseAttemptIdOnFirstSkip()
              }
            case narrowDep: NarrowDependency[_] =>
             //若为窄依赖，那就属于同一个stage。并将依赖的RDD放入waitingForVisit中，以能够在下面的while中继续向上visit，直至遍历了整个DAG图
              waitingForVisit.prepend(narrowDep.rdd)
          }
        }
      }
    }
  }
  while (waitingForVisit.nonEmpty) {
    visit(waitingForVisit.remove(0))
  }
  missing.toList
}