本文章，原创 若泽数据 ，禁止所有阅读，转载，分享及评论

spark on yarn 执行流程前置

构建Spark Application的运行环境（启动SparkContext），SparkContext向资源管理器（可以是Standalone、Mesos或YARN）注册并申请运行Executor资源；
资源管理器分配Executor资源并启动StandaloneExecutorBackend，Executor运行情况将随着心跳发送到资源管理器上；
SparkContext构建成DAG图，将DAG图分解成Stage，并把Taskset发送给Task Scheduler。Executor向SparkContext申请Task，Task Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor。
Task在Executor上运行，运行完毕释放所有资源。

流程大概我们都清楚，但详细的执行流程要从源码里分析
当我们在开发完成一个appliction后，要用spark-submit提交到集群，通常都在提交到yarn上执行，执行的命令如下：

./bin/spark-submit \
  --master yarn-cluster \
  --num-executors 100 \
  --executor-memory 6G \
  --executor-cores 4 \
  --driver-memory 1G \
  --conf spark.default.parallelism=1000 \
  --conf spark.storage.memoryFraction=0.5 \
  --conf spark.shuffle.memoryFraction=0.3 \

spark提供了一个命令spark-submit，那么我们第一步就去看一下这个文件,发现里面只有这一行，说明实际上是去找另一个spark-class的文件，参数是SparkSubmit

exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

在spark-class里，前面是一些Jdk,Spark assembly,还有环境变量的检查与设置，
首先我们看这个文件:
首先检查SPARK_HOME目录若为空，刚设置SPARK_HOME目录，接着加载load.spark-env.sh文件

if [ -z "${SPARK_HOME}" ]; then
  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
fi

. "${SPARK_HOME}"/bin/load-spark-env.sh

那么我们进入load-spark-env.sh后，我们可以看出这个文件做了下面几件事：

检查SPARK_HOME是否设定
检查SPARK_ENV_LOADED，得到SPARK_HOME/conf,加载sparn-env.sh
检查并设置SPARK_SCALA_VERSION

if [ -z "$SPARK_ENV_LOADED" ]; then
  export SPARK_ENV_LOADED=1

  # Returns the parent of the directory this script lives in.
  parent_dir="${SPARK_HOME}"

  user_conf_dir="${SPARK_CONF_DIR:-"$parent_dir"/conf}"

  if [ -f "${user_conf_dir}/spark-env.sh" ]; then
    # Promote all variable declarations to environment (exported) variables
    set -a
    . "${user_conf_dir}/spark-env.sh"
    set +a
  fi
fi
if [ -z "$SPARK_SCALA_VERSION" ]; then

  ASSEMBLY_DIR2="${SPARK_HOME}/assembly/target/scala-2.11"
  ASSEMBLY_DIR1="${SPARK_HOME}/assembly/target/scala-2.10"

  if [[ -d "$ASSEMBLY_DIR2" && -d "$ASSEMBLY_DIR1" ]]; then
    echo -e "Presence of build for both scala versions(SCALA 2.10 and SCALA 2.11) detected." 1>&2
    echo -e 'Either clean one of them or, export SPARK_SCALA_VERSION=2.11 in spark-env.sh.' 1>&2
    exit 1
  fi

  if [ -d "$ASSEMBLY_DIR2" ]; then
    export SPARK_SCALA_VERSION="2.11"
  else
    export SPARK_SCALA_VERSION="2.10"
  fi
fi

返回到spark class文件，下面shell里可以看到，通过上面加载的spark-evn.sh，就可以检查并找到JAVA_HOME,
接着，找到SPARK_ASSEMBLY_JAR及所在目录，如下图：

# Find the java binary
if [ -n "${JAVA_HOME}" ]; then
  RUNNER="${JAVA_HOME}/bin/java"
else
  if [ `command -v java` ]; then
    RUNNER="java"
  else
    echo "JAVA_HOME is not set" >&2
    exit 1
  fi
fi
# Find assembly jar
SPARK_ASSEMBLY_JAR=
if [ -f "${SPARK_HOME}/RELEASE" ]; then
  ASSEMBLY_DIR="${SPARK_HOME}/lib"
else
  ASSEMBLY_DIR="${SPARK_HOME}/assembly/target/scala-$SPARK_SCALA_VERSION"
fi

其实核心就是下面这句了：

CMD=()
while IFS= read -d '' -r ARG; do
  CMD+=("$ARG")
done < <("$RUNNER" -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "$@")
exec "${CMD[@]}"

循环读取ARG参数，加入到CMD中。然后执行了"$RUNNER" -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "$@ 这个是真正执行的第一个spark的类
org.apache.spark.launcher.Main,下面是对这个类的理解，其实说白了，就是把我们上面submit时的参数拼装，成为一个可命令的commend

 * Usage: Main [class] [class args]
   * <p>
   * This CLI works in two different modes:
   * <ul>
   *   <li>"spark-submit": if <i>class</i> is "org.apache.spark.deploy.SparkSubmit", the
   *   {@link SparkLauncher} class is used to launch a Spark application.</li>
   *   <li>"spark-class": if another class is provided, an internal Spark class is run.</li>
   * </ul>

那么下面的exec "${CMD[@]}"就是去执行这个命令
写到这里，我们的写的application jar才开始执行了，上面的全是一些在执行前的准备工作。

spark on yarn执行流程

当上面的准备工作全部完成后，那么执行命令，就开始我们真正的spark执行流程了
首先在通过命令的参数，找到我们打包里开发好的主类(java反射)，
回想一下，之前我们写的spark应用的第一行是不是先构造一个sparkConf,接着通过sparkConfs构造一个非常重要的对象：SparkContext

val conf = new SparkConf()
val sc = new SparkContext(conf)

现在我们进入SparkContext这个类中，发现在在初始化这个类的时候，实际上做了好多事，比如检查--master是哪种运行模式等,如下面的源码,是不是在new SparkContext时，去调用createTaskScheduler方法

 // Create and start the scheduler
    val (sched, ts) = SparkContext.createTaskScheduler(this, master)
    _schedulerBackend = sched
    _taskScheduler = ts
    _dagScheduler = new DAGScheduler(this)
    _heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)

其中在createTaskScheduler做的最重要的两件事：

1：构造出DAGScheduler
2：构造出TaskScheduler

如下面的源码里分析：

 //当发现--master是Yarn-client模式后，那么是不是根据java的反射类加载，得到TaskSchedulerImpl对象
 //也就是上面的构造出TaskScheduler
 case "yarn-client" =>
        val scheduler = try {
          val clazz = Utils.classForName("org.apache.spark.scheduler.cluster.YarnScheduler")
          val cons = clazz.getConstructor(classOf[SparkContext])
          cons.newInstance(sc).asInstanceOf[TaskSchedulerImpl]
                } catch {
          case e: Exception => {
            throw new SparkException("YARN mode not available ?", e)
          }
        }
     //同样，代码真的很相似，YarnClientSchedulerBackend对象也创建出来了，也就是上面的构造出DAGScheduler
        val backend = try {
          val clazz =
            Utils.classForName("org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend")
          val cons = clazz.getConstructor(classOf[TaskSchedulerImpl], classOf[SparkContext])
          cons.newInstance(scheduler, sc).asInstanceOf[CoarseGrainedSchedulerBackend]
        } catch {
          case e: Exception => {
            throw new SparkException("YARN mode not available ?", e)
          }
        }

        scheduler.initialize(backend)
        (backend, scheduler)

其实我们同相看下yarn-cluster的代码，也非常的相似，只不过生成的
TaskScheduler的对象是YarnClusterScheduler
DAGScheduler的对象是YarnClusterSchedulerBackend
如图：

case "yarn-standalone" | "yarn-cluster" =>
        if (master == "yarn-standalone") {
          logWarning(
            "\"yarn-standalone\" is deprecated as of Spark 1.0. Use \"yarn-cluster\" instead.")
        }
        val scheduler = try {
          val clazz = Utils.classForName("org.apache.spark.scheduler.cluster.YarnClusterScheduler")
          val cons = clazz.getConstructor(classOf[SparkContext])
          cons.newInstance(sc).asInstanceOf[TaskSchedulerImpl]

see tomorrow