2.3 運(yùn)行Spark應(yīng)用程序

運(yùn)行Spark應(yīng)用程序主要包括Local模式運(yùn)行、Standalone模式運(yùn)行、YARN模式運(yùn)行、Mesos模式運(yùn)行（參考官方文檔）。

2.3.1 Local模式運(yùn)行Spark應(yīng)用程序

Local模式運(yùn)行Spark應(yīng)用程序是最簡(jiǎn)單的方式，以計(jì)算圓周率的程序?yàn)槔M(jìn)入安裝主目錄，如spark-1.5.0，執(zhí)行命令：

            # 提交Spark任務(wù)的入口
            ./bin/spark-submit \
            #主程序設(shè)置本地，local[*]，其中*是指設(shè)置線程數(shù)
            --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

執(zhí)行結(jié)果如下:

          *** 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 # 計(jì)算出Pi值，表明運(yùn)行成功。

2.3.2 Standalone模式運(yùn)行Spark應(yīng)用程序

Spark獨(dú)立模式下應(yīng)用程序的運(yùn)行以及資源調(diào)度、監(jiān)控和日志等內(nèi)容會(huì)在接下來(lái)做簡(jiǎn)單的介紹。

1. spark-shell運(yùn)行應(yīng)用程序

在Spark集群上運(yùn)行應(yīng)用程序，需要將Master的Spark：//IP:PORT傳遞給SparkContext構(gòu)造函數(shù)。

在集群上運(yùn)行交互式的Spark命令spark-shell，該命令將會(huì)用spark-env.sh中的SPARK_MASTER_IP和SPARK_MASTER_PORT自動(dòng)設(shè)置Master。

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

另外，還可以通過(guò)傳遞一個(gè)--cores<numCore>來(lái)控制spark-shell在集群上使用的核心的數(shù)目。

2. spark-submit啟動(dòng)應(yīng)用程序

spark-submit腳本提供了一種向集群提交Spark應(yīng)用程序的最直接的方法，對(duì)于一個(gè)獨(dú)立部署模式的集群，Spark目前支持Client部署模式，即在提交應(yīng)用的客戶端進(jìn)程中部署Driver。

如果應(yīng)用程序通過(guò)spark-submit啟動(dòng)，那么應(yīng)用程序的Jar包將會(huì)自動(dòng)地分配給所有的Worker節(jié)點(diǎn)；對(duì)于任何其他運(yùn)行應(yīng)用程序時(shí)需要依賴的Jar包，可以通過(guò)-jar聲明，Jar包名之間用逗號(hào)隔開(kāi)。

以SparkPi為例，提交應(yīng)用程序命令如下：

          ./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任務(wù)的腳本，--class org.apache.spark.examples.SparkPi為Spark項(xiàng)目包中指定運(yùn)行的類的全路徑。--master spark://$YOUR_MASTER_IP:7077為主節(jié)點(diǎn)的路徑和端口號(hào)。$YOUR_SPARK_HOME/lib/spark-examples-1.5.0-hadoop2.3.0.jar是Spark項(xiàng)目程序包的路徑。

直接運(yùn)行即可驗(yàn)證Spark在Standalone模式下的計(jì)算圓周率的程序。如果啟動(dòng)上述應(yīng)用程序成功，則執(zhí)行結(jié)果如下：

            *** 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

當(dāng)出現(xiàn)Pi is roughly 3.14058，表明計(jì)算成功。

3.資源調(diào)度

獨(dú)立部署模式支持FIFO作業(yè)調(diào)度策略。不過(guò)，為了允許多并發(fā)執(zhí)行，你可以控制每一個(gè)應(yīng)用可獲得的資源最大值。默認(rèn)情況下，如果集群中一次只運(yùn)行一個(gè)應(yīng)用程序，它就會(huì)獲得所有CPU核。你可以在SparkConf中設(shè)置spark.cores.max來(lái)配置獲得最多內(nèi)核的數(shù)量，示例如下所示：

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

除此之外，還可以通過(guò)配置集群上的spark.deploy.defaultCores來(lái)改變應(yīng)用程序使用的默認(rèn)值，而不必設(shè)置spark.cores.max，需要在spark-env.sh中加入以下內(nèi)容：

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

當(dāng)用戶不會(huì)在共享集群上獨(dú)立配置CPU核數(shù)最大值的時(shí)候，這顯得特別重要。

4.監(jiān)控和日志

Spark獨(dú)立部署模式提供了一個(gè)基于Web的用戶接口，監(jiān)控集群運(yùn)行狀況。Master和每一個(gè)Worker都會(huì)有一個(gè)WebUI來(lái)顯示集群的統(tǒng)計(jì)信息。默認(rèn)情況下，可以通過(guò)8080端口訪問(wèn)Master的WebUI。

另外，每個(gè)Slave節(jié)點(diǎn)上作業(yè)運(yùn)行的日志也會(huì)詳細(xì)地記錄到默認(rèn)的$SPARK_HOME/work目錄下每個(gè)作業(yè)會(huì)對(duì)應(yīng)兩個(gè)文件：stdout和stderr，包含了控制臺(tái)上所有的歷史輸出。

如圖2-2所示，本書(shū)所用數(shù)據(jù)平臺(tái)是通過(guò)$YOUR_SPARK_MASTER_IP:8080訪問(wèn)集群統(tǒng)計(jì)信息情況。

2.3.3 YARN模式運(yùn)行Spark

Spark 0.6版本以后，加入了對(duì)在Hadoop YARN上運(yùn)行的支持，并在之后發(fā)布版本中不斷演進(jìn)，逐漸發(fā)展成Spark應(yīng)用程序運(yùn)行的主要模式。

Hadoop與Spark部署完畢后，即可在YARN上運(yùn)行Spark程序。其任務(wù)提交的方式與獨(dú)立模式類似，只是工作原理有一些不同。

（1）在YARN上啟動(dòng)Spark應(yīng)用

在YARN上啟動(dòng)Spark應(yīng)用有兩種模式：yarn-cluster模式和yarn-client模式。

在yarn-cluster模式下，框架會(huì)在集群中啟動(dòng)的Driver程序；

在yarn-client模式中，框架會(huì)在client端啟動(dòng)Driver程序。在YARN中，Resourcemanager的地址是從Hadoop配置中選取的。因此，master參數(shù)可以簡(jiǎn)單配置為yarn-client或yarn-cluster。

要在yarn-cluster模式下調(diào)用Spark應(yīng)用，示例如下：

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

以SparkPi為例，提交應(yīng)用程序命令如下：

          $ ./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

首先啟動(dòng)一個(gè)Application Master的客戶程序，之后SparkPi將作為Application Master的一個(gè)子進(jìn)程運(yùn)行。

Client會(huì)周期性地檢查Application Master以獲得狀態(tài)更新，當(dāng)應(yīng)用程序運(yùn)行結(jié)束時(shí)，Client會(huì)立刻退出。

要以yarn-client模式啟動(dòng)一個(gè)Spark應(yīng)用，操作相同，用“yarn-client”替換“yarn-cluster”即可。

在YARN模式下，通過(guò)spark-submit提交任務(wù)應(yīng)用程序，示例如下：

            ./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

直接運(yùn)行上述程序即可測(cè)試Spark在YARN模式下計(jì)算圓周率的程序。

（2）添加其他的Jar包

在yarn-cluster模式下，Driver和Client運(yùn)行于不同的機(jī)器，SparkContext.addJar不會(huì)作用于Client本地文件。要使SparkContext.addJar可使用Client的本地文件，在啟動(dòng)指令中的--jars選項(xiàng)中加入文件。

            $ ./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）監(jiān)控和日志

程序運(yùn)行完畢后，如果Hadoop集群配置完畢，那么在Hadoop集群Cluster頁(yè)面上，也就是在http://$YOUR_MASTER_IP:8088/cluster/apps/FINISHED路徑下可以看到執(zhí)行完成的Spark程序及其輸出日志，或者在http://$YOUR_MASTER_IP:8080/路徑下可以看到執(zhí)行的Spark任務(wù)。

如圖2-3所示，本書(shū)所用數(shù)據(jù)平臺(tái)是通過(guò)$YOUR_MASTER_IP:8088訪問(wèn)集群統(tǒng)計(jì)信息情況。

2.3.4 應(yīng)用程序提交和參數(shù)傳遞

提交任務(wù)是Spark運(yùn)行的起步，理解任務(wù)提交時(shí)不同參數(shù)的含義十分重要。

1.應(yīng)用程序提交過(guò)程

Spark應(yīng)用程序在集群上以獨(dú)立進(jìn)程集合的形式運(yùn)行，接受用戶Driver程序中main函數(shù)SparkContext對(duì)象的協(xié)調(diào)。

確切來(lái)說(shuō)，當(dāng)任務(wù)提交到集群上，SparkContext對(duì)象可以與多種集群管理器（Standalone部署、Mesos、YARN模式）連接，這些集群管理器負(fù)責(zé)為所有應(yīng)用分配資源。

一旦連接建立，Spark可以在集群的節(jié)點(diǎn)上獲得Executor，這些Executor進(jìn)程負(fù)責(zé)執(zhí)行計(jì)算和為應(yīng)用程序存儲(chǔ)數(shù)據(jù)。

接下來(lái)，Spark會(huì)將應(yīng)用程序代碼（傳遞給SparkContext的Jar或Python文件）發(fā)送給Executor。

最后，SparkContext將任務(wù)發(fā)送至Executor執(zhí)行，圖2-4所示為Spark任務(wù)提交流程。

執(zhí)行過(guò)程中，需要注意以下情況：

1）每個(gè)應(yīng)用程序都會(huì)獲得自己擁有的Executor進(jìn)程，這些進(jìn)程的生命周期持續(xù)在整個(gè)應(yīng)用運(yùn)行期間，并以多線程的方式執(zhí)行任務(wù)。無(wú)論是在Driver驅(qū)動(dòng)程序，還是Executor執(zhí)行進(jìn)程，都能很好地隔離應(yīng)用程序，如果沒(méi)有將數(shù)據(jù)寫(xiě)到外部存儲(chǔ)，該數(shù)據(jù)就不能被其他Spark應(yīng)用共享。

2）Spark對(duì)集群管理器是不可知的，只要Spark能夠獲取Executor進(jìn)程，并且這些Executor之間可以相互通信，Spark不關(guān)注集群管理器是否支持其他的業(yè)務(wù)，如YARN同時(shí)支持MapReduce程序的運(yùn)行。

3）因?yàn)镈river負(fù)責(zé)與集群上的任務(wù)交互，所以Driver應(yīng)該運(yùn)行于距離Worker節(jié)點(diǎn)近的地方，最好在同一個(gè)本地局域網(wǎng)之中。如果想要給集群發(fā)送遠(yuǎn)程請(qǐng)求，那么請(qǐng)為Driver安裝RPC協(xié)議并從附近提交，而不是在遠(yuǎn)離Worker節(jié)點(diǎn)的地方運(yùn)行Driver。

2.配置參數(shù)傳遞

在SparkConf中顯式配置的參數(shù)具有最高優(yōu)先級(jí)，然后是傳遞給spark-submit的標(biāo)志位，最后是默認(rèn)屬性配置文件中的值。

如果沒(méi)有在SparkConf中顯示配置，又不想在spark-submit中提交，可以通過(guò)默認(rèn)屬性配置文件獲取值。spark-submit腳本可以從屬性文件中加載Spark配置值，并將它們傳遞給應(yīng)用程序。

默認(rèn)情況下，spark-submit腳本將會(huì)從Spark目錄下的conf/spark-defaults.conf中讀取配置選項(xiàng)。例如，如果Spark的Master屬性已經(jīng)設(shè)置，spark-submit提交時(shí)可以省略--master標(biāo)志。

如果應(yīng)用程序代碼依賴于其他的項(xiàng)目，需要將它們與應(yīng)用程序一起打包，以便應(yīng)用程序的代碼可以分發(fā)到Spark集群之上。為此，需要?jiǎng)?chuàng)建一個(gè)包含應(yīng)用程序代碼和其依賴的Jar包（當(dāng)創(chuàng)建Jar包時(shí)，需要將Spark和Hadoop作為提供依賴項(xiàng)；這兩者不需要被打入包中，在運(yùn)行時(shí)由集群管理器提供加載）。

一旦擁有Jar包，就可以在調(diào)用bin/spark-submit腳本時(shí)傳遞Jar包。應(yīng)用程序的Jar包和所有在--jars選項(xiàng)中的Jar包都會(huì)被自動(dòng)地傳遞到集群。Spark使用下面的URL支持按不同的策略散播的Jar包。

□ f ile：Driver的HTTP文件服務(wù)器提供了絕對(duì)路徑和f ile:// URI，所有的Executor節(jié)點(diǎn)從Driver的HTTP服務(wù)器處獲得文件。

□ hdfs：從HDFS的URI獲取文件和Jar包。

□ local以每個(gè)Worker節(jié)點(diǎn)相同的local:/開(kāi)頭的URI文件。

當(dāng)在yarn-cluster模式下對(duì)本地文件使用--jars選項(xiàng)時(shí)，該選項(xiàng)允許你使用SparkContext. addJar函數(shù)。若對(duì)于HDFS、HTTP、HTTPS或FTP文件，則不必使用。