zoukankan      html  css  js  c++  java
  • Spark提交应用程序之Spark-Submit分析

    转载必须注明出处:梁杰帆

    在这里要先感谢原作者们!如果各位在这里发现了错误之处,请大家提出

      2017-12-28 15:41:13

    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 "$@" 

    可以看出来 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 "$@") 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共享。

  • 相关阅读:
    畅通工程续 (dijkstra)
    最短路径问题 HDU 3790
    【基础算法-模拟-例题-玩具谜题】-C++
    【基础算法-模拟-例题-金币】-C++
    【动态规划例题-数塔问题】-C++
    【基本数据结构之'图'】
    【最小生成树之Kruskal例题-建设电力系统】-C++
    【最短路算法例题-升降梯上】-C++
    【基本数据结构之栈】
    【栈-例题】网页跳转-C++
  • 原文地址:https://www.cnblogs.com/liangjf/p/8134645.html
Copyright © 2011-2022 走看看