2.3 运行Spark应用程序

运行Spark应用程序主要包括Local模式运行、Standalone模式运行、YARN模式运行、Mesos模式运行（参考官方文档）。

2.3.1 Local模式运行Spark应用程序

Local模式运行Spark应用程序是最简单的方式，以计算圆周率的程序为例，进入安装主目录，如spark-1.5.0，执行命令：

            # 提交Spark任务的入口
            ./bin/spark-submit \
            #主程序设置本地，local[*]，其中*是指设置线程数
            --master local[10] \

          #选择主类名称
          --class org.apache.spark.examples.SparkPi \
          #examples Jar包
          /home/hadoop/spark-1.5.0/lib/spark-examples-1.5.0-hadoop2.3.0.jar

执行结果如下:

          *** INFO spark.SparkContext: Starting job: reduce at SparkPi.scala:35
          ……
          *** INFO spark.SparkContext: Job finished: reduce at SparkPi.scala:35, took 0.751093979 s
          Pi is roughly 3.14826 # 计算出Pi值，表明运行成功。

2.3.2 Standalone模式运行Spark应用程序

Spark独立模式下应用程序的运行以及资源调度、监控和日志等内容会在接下来做简单的介绍。

1. spark-shell运行应用程序

在Spark集群上运行应用程序，需要将Master的Spark：//IP:PORT传递给SparkContext构造函数。

在集群上运行交互式的Spark命令spark-shell，该命令将会用spark-env.sh中的SPARK_MASTER_IP和SPARK_MASTER_PORT自动设置Master。

./bin/spark-shell --master spark:// IP:PORT

另外，还可以通过传递一个--cores<numCore>来控制spark-shell在集群上使用的核心的数目。

2. spark-submit启动应用程序

spark-submit脚本提供了一种向集群提交Spark应用程序的最直接的方法，对于一个独立部署模式的集群，Spark目前支持Client部署模式，即在提交应用的客户端进程中部署Driver。

如果应用程序通过spark-submit启动，那么应用程序的Jar包将会自动地分配给所有的Worker节点；对于任何其他运行应用程序时需要依赖的Jar包，可以通过-jar声明，Jar包名之间用逗号隔开。

以SparkPi为例，提交应用程序命令如下：

          ./bin/spark-submit \
              --class org.apache.spark.examples.SparkPi \
              --master spark://$YOUR_MASTER_IP:7077 \
              --executor-memory 2G \
              --total-executor-cores 2 \
          $YOUR_SPARK_HOME/lib/spark-examples-1.5.0-hadoop2.3.0.jar

其中，spark-submit为提交Spark任务的脚本，--class org.apache.spark.examples.SparkPi为Spark项目包中指定运行的类的全路径。--master spark://$YOUR_MASTER_IP:7077为主节点的路径和端口号。$YOUR_SPARK_HOME/lib/spark-examples-1.5.0-hadoop2.3.0.jar是Spark项目程序包的路径。

直接运行即可验证Spark在Standalone模式下的计算圆周率的程序。如果启动上述应用程序成功，则执行结果如下：

            *** INFO client.AppClient$ClientActor: Connecting to master spark://$YOUR_MASTER_
            IP:7077...
            *** INFO spark.SparkContext: Starting job: reduce at SparkPi.scala:35
            ……
            *** INFO spark.SparkContext: Job finished: reduce at SparkPi.scala:35, took
            15.530349566 s
            Pi is roughly 3.14058

当出现Pi is roughly 3.14058，表明计算成功。

3.资源调度

独立部署模式支持FIFO作业调度策略。不过，为了允许多并发执行，你可以控制每一个应用可获得的资源最大值。默认情况下，如果集群中一次只运行一个应用程序，它就会获得所有CPU核。你可以在SparkConf中设置spark.cores.max来配置获得最多内核的数量，示例如下所示：

            val conf = new SparkConf()
                      .setMaster(...)
                      .setAppName(...)
                      .set("spark.cores.max", "10")
            val sc = new SparkContext(conf)

除此之外，还可以通过配置集群上的spark.deploy.defaultCores来改变应用程序使用的默认值，而不必设置spark.cores.max，需要在spark-env.sh中加入以下内容：

export SPARK_MASTER_OPTS="-Dspark.deploy.defaultCores=<value>"

当用户不会在共享集群上独立配置CPU核数最大值的时候，这显得特别重要。

4.监控和日志

Spark独立部署模式提供了一个基于Web的用户接口，监控集群运行状况。Master和每一个Worker都会有一个WebUI来显示集群的统计信息。默认情况下，可以通过8080端口访问Master的WebUI。

另外，每个Slave节点上作业运行的日志也会详细地记录到默认的$SPARK_HOME/work目录下每个作业会对应两个文件：stdout和stderr，包含了控制台上所有的历史输出。

如图2-2所示，本书所用数据平台是通过$YOUR_SPARK_MASTER_IP:8080访问集群统计信息情况。

图2-2 Spark集群统计信息

2.3.3 YARN模式运行Spark

Spark 0.6版本以后，加入了对在Hadoop YARN上运行的支持，并在之后发布版本中不断演进，逐渐发展成Spark应用程序运行的主要模式。

Hadoop与Spark部署完毕后，即可在YARN上运行Spark程序。其任务提交的方式与独立模式类似，只是工作原理有一些不同。

（1）在YARN上启动Spark应用

在YARN上启动Spark应用有两种模式：yarn-cluster模式和yarn-client模式。

在yarn-cluster模式下，框架会在集群中启动的Driver程序；

在yarn-client模式中，框架会在client端启动Driver程序。在YARN中，Resourcemanager的地址是从Hadoop配置中选取的。因此，master参数可以简单配置为yarn-client或yarn-cluster。

要在yarn-cluster模式下调用Spark应用，示例如下：

          ./bin/spark-submit \
          --class path.to.your.Class \
          --master yarn-cluster [options] <app jar> [app options]

以SparkPi为例，提交应用程序命令如下：

          $ ./bin/spark-submit --class org.apache.spark.examples.SparkPi \
              --master yarn-cluster \
              --num-executors 3 \
              --driver-memory 4g \
              --executor-memory 2g \

            --executor-cores 2 \
            $YOUR_SPARK_HOME/lib/spark-examples-1.5.0-hadoop2.3.0.jar \
            10

首先启动一个Application Master的客户程序，之后SparkPi将作为Application Master的一个子进程运行。

Client会周期性地检查Application Master以获得状态更新，当应用程序运行结束时，Client会立刻退出。

要以yarn-client模式启动一个Spark应用，操作相同，用“yarn-client”替换“yarn-cluster”即可。

在YARN模式下，通过spark-submit提交任务应用程序，示例如下：

            ./bin/spark-submit \
                --class org.apache.spark.examples.SparkPi \
                --master yarn-client \
                --executor-memory 2G \
                --num-executors 3 \
            $YOUR_SPARK_HOME/lib/spark-examples-1.5.0-hadoop2.3.0.jar

直接运行上述程序即可测试Spark在YARN模式下计算圆周率的程序。

（2）添加其他的Jar包

在yarn-cluster模式下，Driver和Client运行于不同的机器，SparkContext.addJar不会作用于Client本地文件。要使SparkContext.addJar可使用Client的本地文件，在启动指令中的--jars选项中加入文件。

            $ ./bin/spark-submit
            --class my.main.Class \
            --master yarn-cluster \
            --jars my-other-jar.jar,my-other-other-jar.jar
            my-main-jar.jar app_arg1 app_arg2

（3）监控和日志

程序运行完毕后，如果Hadoop集群配置完毕，那么在Hadoop集群Cluster页面上，也就是在http://$YOUR_MASTER_IP:8088/cluster/apps/FINISHED路径下可以看到执行完成的Spark程序及其输出日志，或者在http://$YOUR_MASTER_IP:8080/路径下可以看到执行的Spark任务。

如图2-3所示，本书所用数据平台是通过$YOUR_MASTER_IP:8088访问集群统计信息情况。

图2-3 Spark YARN模式任务统计信息

2.3.4 应用程序提交和参数传递

提交任务是Spark运行的起步，理解任务提交时不同参数的含义十分重要。

1.应用程序提交过程

Spark应用程序在集群上以独立进程集合的形式运行，接受用户Driver程序中main函数SparkContext对象的协调。

确切来说，当任务提交到集群上，SparkContext对象可以与多种集群管理器（Standalone部署、Mesos、YARN模式）连接，这些集群管理器负责为所有应用分配资源。

一旦连接建立，Spark可以在集群的节点上获得Executor，这些Executor进程负责执行计算和为应用程序存储数据。

接下来，Spark会将应用程序代码（传递给SparkContext的Jar或Python文件）发送给Executor。

最后，SparkContext将任务发送至Executor执行，图2-4所示为Spark任务提交流程。

图2-4 Spark任务提交流程图

执行过程中，需要注意以下情况：

1）每个应用程序都会获得自己拥有的Executor进程，这些进程的生命周期持续在整个应用运行期间，并以多线程的方式执行任务。无论是在Driver驱动程序，还是Executor执行进程，都能很好地隔离应用程序，如果没有将数据写到外部存储，该数据就不能被其他Spark应用共享。

2）Spark对集群管理器是不可知的，只要Spark能够获取Executor进程，并且这些Executor之间可以相互通信，Spark不关注集群管理器是否支持其他的业务，如YARN同时支持MapReduce程序的运行。

3）因为Driver负责与集群上的任务交互，所以Driver应该运行于距离Worker节点近的地方，最好在同一个本地局域网之中。如果想要给集群发送远程请求，那么请为Driver安装RPC协议并从附近提交，而不是在远离Worker节点的地方运行Driver。

2.配置参数传递

在SparkConf中显式配置的参数具有最高优先级，然后是传递给spark-submit的标志位，最后是默认属性配置文件中的值。

如果没有在SparkConf中显示配置，又不想在spark-submit中提交，可以通过默认属性配置文件获取值。spark-submit脚本可以从属性文件中加载Spark配置值，并将它们传递给应用程序。

默认情况下，spark-submit脚本将会从Spark目录下的conf/spark-defaults.conf中读取配置选项。例如，如果Spark的Master属性已经设置，spark-submit提交时可以省略--master标志。

如果应用程序代码依赖于其他的项目，需要将它们与应用程序一起打包，以便应用程序的代码可以分发到Spark集群之上。为此，需要创建一个包含应用程序代码和其依赖的Jar包（当创建Jar包时，需要将Spark和Hadoop作为提供依赖项；这两者不需要被打入包中，在运行时由集群管理器提供加载）。

一旦拥有Jar包，就可以在调用bin/spark-submit脚本时传递Jar包。应用程序的Jar包和所有在--jars选项中的Jar包都会被自动地传递到集群。Spark使用下面的URL支持按不同的策略散播的Jar包。

□ f ile：Driver的HTTP文件服务器提供了绝对路径和f ile:// URI，所有的Executor节点从Driver的HTTP服务器处获得文件。

□ hdfs：从HDFS的URI获取文件和Jar包。

□ local以每个Worker节点相同的local:/开头的URI文件。

注意

在Executor节点上的每个SparkContext对象的Jar包和文件会被复制到工作目录下。这可能会占用大量的空间，并且需要被及时清除。在YARN集群上，清除是自动执行的。在Standalone模式部署集群上，自动清除可以通过spark.worker.cleamup. appDataTtl属性配置。

当在yarn-cluster模式下对本地文件使用--jars选项时，该选项允许你使用SparkContext. addJar函数。若对于HDFS、HTTP、HTTPS或FTP文件，则不必使用。

上一章目录下一章