Spark提交应用程序之Spark-Submit分析

Posted 2020-11-19 itboys

1.提交应用程序 
在提交应用程序的时候，用到 spark-submit 脚本。我们来看下这个脚本： 

if [ -z "${SPARK_HOME}" ]; then
  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
fi
 
# disable randomized hash for string in Python 3.3+
export PYTHONHASHSEED=0
 
exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "[email protected]"

可以看出来 spark-class 脚本才是真正的提交程序的，spark-submit 脚本只是在其上封装一层，并传参 org.apache.spark.deploy.SparkSubmit 给它。这样做的目的是分层管理和方便维护的作用。符合计算机中“遇到问题，往上加多一层解决问题的思想(通过加层解决不了问题，唯一的原因就是层次太多，无法再加了)” 
 
下面来看下真正的提交程序脚本 spark-class ： 

if [ -z "${SPARK_HOME}" ]; then
  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
fi
 
. "${SPARK_HOME}"/bin/load-spark-env.sh     #导入Spark运行的环境变量，获取和验证Spark Scala的版本，
 
# Find the java binary    #获取java的路径，用于后面的用户程序运行。因为Saprk虽然是用Scala写的，但是其实跑在JVM上的。
if [ -n "${JAVA_HOME}" ]; then   
  RUNNER="${JAVA_HOME}/bin/java"    #如果 JAVA_HOME 环境变量已设置，直接获取java脚本路径
else
  if [ `command -v java` ]; then
    RUNNER="java"     #如果 JAVA_HOME 环境变量没设置，通过 command -v java 命令获得
  else
    echo "JAVA_HOME is not set" >&2        #如果没有找到java命令，那么直接退出，把信息重定向到标准错误输出
    exit 1
  fi
fi
 
# Find assembly jar     #获取Spark运行相关jar包路径
SPARK_ASSEMBLY_JAR=
if [ -f "${SPARK_HOME}/RELEASE" ]; then        
  ASSEMBLY_DIR="${SPARK_HOME}/lib"    #如果RELEASE文件存在，表明Spark已结合hadoop编译生成的jar包在lib。建议使用cdh5的生态圈，不用自己编译，解决版本冲突的问题
else
  ASSEMBLY_DIR="${SPARK_HOME}/assembly/target/scala-$SPARK_SCALA_VERSION"     #注意，在Spark2.0版本后，这个目录已经没有了，把一个大的jar包分为若干个jar包
fi
#ls -l lib目录，并通过正则匹配 spark-assembly-1.6.0-cdh5.7.0-hadoop2.6.0-cdh5.7.0.jar(我这里以spark-1.6.0-cdh5.7.0为例子，并统计文件行数）
GREP_OPTIONS= num_jars="$(ls -1 "$ASSEMBLY_DIR" | grep "^spark-assembly.*hadoop.*.jar$" | wc -l)" 

#如果上面的num_jars统计文件行数为0，或上面的$SPARK_ASSEMBLY_JAR长度为0，或$SPARK_PREPEND_CLASSES不存在，打印log并退出
if [ "$num_jars" -eq "0" -a -z "$SPARK_ASSEMBLY_JAR" -a "$SPARK_PREPEND_CLASSES" != "1" ]; thenecho "Failed to find Spark assembly in $ASSEMBLY_DIR." 1>&2
  echo "You need to build Spark before running this program." 1>&2
  exit 1
fi
if [ -d "$ASSEMBLY_DIR" ]; then        #如果上面的$ASSEMBLY_DIR找到
  ASSEMBLY_JARS="$(ls -1 "$ASSEMBLY_DIR" | grep "^spark-assembly.*hadoop.*.jar$" || true)"
  if [ "$num_jars" -gt "1" ]; then    #如果有多个num_jars，则打印log并退出，防止运行时因为Spark版本出错
    echo "Found multiple Spark assembly jars in $ASSEMBLY_DIR:" 1>&2
    echo "$ASSEMBLY_JARS" 1>&2
    echo "Please remove all but one jar." 1>&2
    exit 1
  fi
fi
 
SPARK_ASSEMBLY_JAR="${ASSEMBLY_DIR}/${ASSEMBLY_JARS}"    #获取spark-assembly-1.6.0-cdh5.7.0-hadoop2.6.0-cdh5.7.0.jar(我这里以spark-1.6.0-cdh5.7.0为例子)
 
LAUNCH_CLASSPATH="$SPARK_ASSEMBLY_JAR"    #简单赋值，更换变量是为了方便维护
 
# Add the launcher build dir to the classpath if requested.    #若有需要，把启动程序构建目录添加到classpath
if [ -n "$SPARK_PREPEND_CLASSES" ]; then
  LAUNCH_CLASSPATH="${SPARK_HOME}/launcher/target/scala-$SPARK_SCALA_VERSION/classes:$LAUNCH_CLASSPATH"
fi
 
export _SPARK_ASSEMBLY="$SPARK_ASSEMBLY_JAR"    #导出$SPARK_ASSEMBLY_JAR，相当于作为全局变量，供其他程序访问
 
# For tests
if [[ -n "$SPARK_TESTING" ]]; then
  unset YARN_CONF_DIR
  unset HADOOP_CONF_DIR
fi
 
#获取命令行传进来的全部参数，赋给启动程序
CMD=()
while IFS= read -d ‘‘ -r ARG; do
  CMD+=("$ARG")
done < <("$RUNNER" -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "[email protected]")
exec "${CMD[@]}" 

通过对Spark-class的解读，可以知道应用程序提交要做的事情，总的来说，就是导入Spark运行的环境变量，获取java命令路径，相关与Spark运行相关的jar包，获取命令行的参数等。最终通过 org.apache.spark.launcher.Main  Main.class这个类来启动应用程序。 
 
2.添加应用程序的依赖项 
 --jar     添加自定义的jar包  
 --py-files    添加.py .zip .egg等文件。如果是多个.py文件，需要压缩成.zip文件 
 
3.spark-submit启动应用程序 

./bin/spark-submit  
  --class <main-class>  
  --master <master-url>  
  --deploy-mode <deploy-mode>  
  --conf <key>=<value>  
  ... # other options 
  <application-jar>  
  [application-arguments] 
 

——class:应用程序的入口点(如org.apache.spark.examples.SparkPi) 
——master:主集群的URL(如spark://192.168.1.100:7077) 
——deploy-mode:是否将驱动程序部署到工作节点(集群)或本地作为外部客户端(客户端)(默认:客户端) 
——conf:键=值格式的任意Spark配置属性。对于包含空格的值，用引号括起“key = value” 
application-jar:绑定jar的路径，包括应用程序和所有依赖项。URL必须在集群内部全局可见，例如，一个hdfs://路径或文件://在所有节点上存在的路径 
[application-arguments]:传递给主类主方法的参数 
【对于Python应用程序，比如test.py 只需在<application-JAR>不提交jar，而是通过 --py-files 添加Python文件, 比如.zip, .egg或.py文件到此参数】 
 
3.Master URLs 
local                     　　 　　　　　　　　　  本地单线程方式运行 
local[K]                 　　　　　　　　　　      本地多线程方式运行 
local[K,F]                 　 　　　　　　　　      本地多线程方式运行，并指定最大失败次数 
local[*]                 　　  　　                              本地有多少core就有多少线程运行 
local[*,F]                 　   　　　　　　　　　  本地有多少core就有多少线程运行，并指定最大失败次数 
spark://HOST:PORT         　　　　　　　　 standalone模式运行，PORT默认值为7077 
spark://HOST1:PORT1,HOST2:PORT2         standalone高可用模式运行，指定多个master的host和port，port默认值为7077 
mesos://HOST:PORT         　　　　　　       mesos集群方式运行，分粗粒度模式和细粒度模式 
yarn                    　　　　　　　　　　　　   yarn模式运行。 
 
4.从文件加载配置(spark-defaults.conf) 
除了在Spark-Submit提交程序时通过-conf "key=value"方式传入Spark参数之外，在这里可以设定大量的参数来调优化，主要涉及到环境参数，应用运行参数，内存管理，网络IO，任务调度，动态分配，安全，身份验证等方面。由于篇幅问题，把官网翻译的全部参数弄成表格放到github上。从此设置参数，优化Spark，妈妈再也不用担心我了。由于水平问题，难免翻译不准确，欢迎大家补充。

5.依赖管理 
在使用spark-submit时，应用程序jar和包含的任何jar包都将自动上传到集群。之后提供的url-jar必须用逗号分隔。该列表包含在driver和excutor类路径中。目录扩展不能用-jar。 
file:-绝对路径和 file:/uri 由driver的HTTP文件服务器提供，每个excutor从driver HTTP服务器获取文件。 
hdfs:，http:, https:, ftp: 从URI中提取文件和jar 
本地: local:/ 将作为每个工作节点上的本地文件存在。这意味着不会产生任何网络IO，并且适用于被推到每个worker上的大文件/jar，或者通过NFS、GlusterFS共享。