本文章，原创 若泽数据 ，禁止所有阅读，转载，分享及评论
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
网友评论