Spark on Yarn詳解

Spark 可以跑在很多集群上，比如跑在local上，跑在Standalone上，跑在Apache Mesos上，跑在Hadoop YARN上等等。不管你Spark跑在什么上面，它的代碼都是一樣的，區別只是–master的時候不一樣。其中Spark on YARN是工作中或生產上用的非常多的一種運行模式。今天主要對Spark on Yarn 這種方式做講解。

作業提交

Standalone模式的提交

此處master設置為管理節點的集群地址（spark webui上顯示的地址）

之后將測試代碼打包成jar包，上傳到服務器，使用spark-submit提交作業。

提交命令：

spark-submit --master spark://s44:7077 --name MyWordCount --class com.demo.spark.scala.WordCountScala SparkDemo-1.0-SNAPSHOT.jar hdfs://s44:8020/usr/hadoop/test.txt

把master設置為集群master的url，名字可以隨意定義；class參數要帶包名，一直指定到要運行的類入口，后面緊跟所在的jar,jar包參數后 跟main需要的自定義參數，如果有多個，空格隔開即可。這里只需要一個參數，用來指定程序要加載的文件目錄。

提交后等着打印結果就好,另外可以根據屏幕打印出來的提示查看日志，還可以在webui上查看具體的執行過程。

yarn模式兩種提交任務方式

Spark可以和Yarn整合，將Application提交到Yarn上運行，Yarn有兩種提交任務的方式。

yarn-client提交任務方式

配置：

在client節點配置中spark-env.sh添加Hadoop_HOME的配置目錄即可提交yarn 任務，具體步驟如下：

注意client只需要有Spark的安裝包即可提交任務，不需要其他配置（比如slaves）

提交命令

./spark-submit --master yarn  --class org.apache.spark.examples.SparkPi  ../lib/spark-examples-xx.jar 100

./spark-submit   --master yarn-lient   --class org.apache.spark.examples.SparkPi ../lib/spark-examples-xx.jar 100

./spark-submit  --master yarn --deploy-mode  client --class org.apache.spark.examples.SparkPi ../lib/spark-examples-xx.jar 100

執行流程圖：

執行流程：

1.客戶端提交一個Application，在客戶端啟動一個Driver進程。

2.Driver進程會向RS(ResourceManager)發送請求，啟動AM(ApplicationMaster)。

3.RS收到請求，隨機選擇一台NM(NodeManager)啟動AM。這里的NM相當於Standalone中的Worker節點。

4.AM啟動后，會向RS請求一批container資源，用於啟動Executor。

5.RS會找到一批NM返回給AM,用於啟動Executor。
AM會向NM發送命令啟動Executor。

6.Executor啟動后，會反向注冊給Driver，Driver發送task到Executor,執行情況和結果返回給Driver端。

小結：
1、Yarn-client模式同樣是適用於測試，因為Driver運行在本地，Driver會與yarn集群中的Executor進行大量的通信，會造成客戶機網卡流量的大量增加.

2、 ApplicationMaster的作用：

為當前的Application申請資源

給NodeManager發送消息啟動Executor。

注意：ApplicationMaster有launchExecutor和申請資源的功能，並沒有作業調度的功能。

yarn-cluster提交任務方式

提交命令

./spark-submit --master yarn --deploy-mode cluster  --class org.apache.spark.examples.SparkPi ../lib/spark-examples-xx.jar 100

./spark-submit   --master yarn-cluster  --class org.apache.spark.examples.SparkPi ../lib/spark-examples-xx.jar 100

結果在yarn的日志里面：

執行流程圖：

執行流程：

1.客戶機提交Application應用程序，發送請求到RS(ResourceManager),請求啟動AM(ApplicationMaster)。

2.RS收到請求后隨機在一台NM(NodeManager)上啟動AM（相當於Driver端）。

3.AM啟動，AM發送請求到RS，請求一批container用於啟動Executor。

3.RS返回一批NM節點給AM。

4.AM連接到NM,發送請求到NM啟動Executor。

5.Executor反向注冊到AM所在的節點的Driver。Driver發送task到Executor。

小結

1.Yarn-Cluster主要用於生產環境中，因為Driver運行在Yarn集群中某一台nodeManager中，每次提交任務的Driver所在的機器都是隨機的，不會產生某一台機器網卡流量激增的現象，缺點是任務提交后不能看到日志。只能通過yarn查看日志。

2.ApplicationMaster的作用：

為當前的Application申請資源

給nodemanager發送消息 啟動Excutor。

任務調度。(這里和client模式的區別是AM具有調度能力，因為其就是Driver端，包含Driver進程)

yarn停止任務命令：

yarn application -kill applicationID

Yarn API

hadoop官方文檔里面有很多API接口, http://hadoop.apache.org/docs/r2.8.3/hadoop-yarn/hadoop-yarn-site/ResourceManagerRest.html

接下來使用postman來調用yarn api，你也可以使用編程語言來實現http請求。

首先啟動hdfs：

start-hdfs.sh

准備一個spark jar包，如果沒有可以使用spark樣例spark-examples_xx.jar，並上傳到hdfs上

hdfs dfs -put $SPARK_HOME/examples/jars/spark-examples_xx.jar /

准備需要引入的spark庫，打包成zip文件，並上傳到hdfs上

cd $SPARK_HOME/jars
打包所有jar
hdfs dfs -put __spark_libs__.zip /

cd $SPARK_HOME/jars
打包配置文件
hdfs dfs -put __spark_conf__.zip /

啟動Yarn

start-yarn

訪問 http://localhost:8088，可以看到yarn 啟動正常。

接下來我使用postman 來演示yarn api的調用：

創建spark任務

http://localhost:8088/ws/v1/cluster/apps/new-application

WebHDFS REST API： http://hadoop.apache.org/docs/r2.8.3/hadoop-project-dist/hadoop-hdfs/WebHDFS.html

查看jar包的屬性

http://localhost:50070/webhdfs/v1/spark-examples_xx.jar?op=GETFILESTATUS

查看spark庫zip文件屬性

http://localhost:50070/webhdfs/v1/spark_libs.zip?op=GETFILESTATUS

提交spark任務

每執行一次任務，需要一個app Id，同一個appid 只能提交一次任務

http://localhost:8088/ws/v1/cluster/apps

 {
    "application-id":"application_1577522109894_0001",
    "application-name":"SparkPi",
    "am-container-spec":
    {
      "local-resources":
      {
        "entry":
        [
          {
            "key":"__app__.jar",
            "value":
            {
              "resource":"hdfs://localhost:9000/spark-examples_xx.jar",
              "type":"FILE",
              "visibility":"APPLICATION",
              "size": 2017366,
              "timestamp": 1577517305828
            }
          },
          {
          "key": "__spark_libs__", 
          "value": {
            "resource": "hdfs://localhost:9000/__spark_libs__.zip", 
            "size": 214677962, 
            "timestamp": 1577518946899, 
            "type": "ARCHIVE", 
            "visibility": "APPLICATION"
           }
         }
        ]
      },
      "commands":
      {
        "command":"java -server -Xmx1024m -Dspark.yarn.app.container.log.dir=<LOG_DIR>  -Dspark.master=yarn  -Dspark.submit.deployMode=cluster -Dspark.executor.cores=1 -Dspark.executor.memory=1g -Dspark.app.name=SparkPi  org.apache.spark.deploy.yarn.ApplicationMaster --class org.apache.spark.examples.SparkPi --jar __app__.jar 1><LOG_DIR>/stdout 2><LOG_DIR>/stderr"
      },
      "environment":
      {
        "entry":
        [
          {
            "key": "SPARK_USER",
            "value": "xx"
          },
          {
            "key": "SPARK_YARN_MODE",
            "value": true
          },
          {
            "key": "SPARK_YARN_STAGING_DIR",
            "value": "hdfs://localhost:9000/user/xx/.sparkStaging/application_1577522109894_0001"
          },
          {
            "key": "CLASSPATH",
            "value": "{{PWD}}<CPS>{{PWD}}/__app__.jar<CPS>{{PWD}}/__spark_libs__/*<CPS>$HADOOP_CONF_DIR<CPS>$HADOOP_COMMON_HOME/share/hadoop/common/*<CPS>$HADOOP_COMMON_HOME/share/hadoop/common/lib/*<CPS>$HADOOP_HDFS_HOME/share/hadoop/hdfs/*<CPS>$HADOOP_HDFS_HOME/share/hadoop/hdfs/lib/*<CPS>$HADOOP_YARN_HOME/share/hadoop/yarn/*<CPS>$HADOOP_YARN_HOME/share/hadoop/yarn/lib/*<CPS>$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/*<CPS>$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/lib/*<CPS>{{PWD}}/__spark_conf__/__hadoop_conf__"
          },
          {
            "key": "SPARK_DIST_CLASSPATH",
            "value": "{{PWD}}<CPS>{{PWD}}/__app__.jar<CPS>{{PWD}}/__spark_libs__/*<CPS>$HADOOP_CONF_DIR<CPS>$HADOOP_COMMON_HOME/share/hadoop/common/*<CPS>$HADOOP_COMMON_HOME/share/hadoop/common/lib/*<CPS>$HADOOP_HDFS_HOME/share/hadoop/hdfs/*<CPS>$HADOOP_HDFS_HOME/share/hadoop/hdfs/lib/*<CPS>$HADOOP_YARN_HOME/share/hadoop/yarn/*<CPS>$HADOOP_YARN_HOME/share/hadoop/yarn/lib/*<CPS>$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/*<CPS>$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/lib/*<CPS>{{PWD}}/__spark_conf__/__hadoop_conf__"
          }
        ]
      }
    },
    "unmanaged-AM":false,
    "max-app-attempts":1,
    "resource":
    {
      "memory":1024,
      "vCores":1
    },
    "application-type":"YARN",
    "keep-containers-across-application-attempts":false
  }

查詢spark任務

http://localhost:8088/ws/v1/cluster/apps/application_1577522109894_0001

資源分配

YARN的RM負責管理整個集群，NM則負責管理該工作節點。

YARN的NM可分配core數（即可以分給Container的最大CPU核數）由參數yarn.nodemanager.resource.cpu-vcores指定，一般要小於本節點的物理CPU核數，因為要預留一些資源給其他任務。Hadoop集群工作節點一般都是同構的，即配置相同。NM可分配給Container的最大內存則由參數yarn.nodemanager.resource.memory-mb指定，默認情況下，可分配內存會小於本機內存*0.8。

注意，分配給作業的資源不要超過YARN可分配的集群資源總數。注意：分配給單個Container的核數和內存不能超過閾值，即為Executor設置的核數和內存不能超過閾值。若分配給作業的資源超過上限，將不會啟動指定數目的Executor（也就是說，不會起足夠數目的Container）。

參數設置

yarn 參數設置

在YARN中，資源管理由ResourceManager和NodeManager共同完成，其中，ResourceManager中的調度器負責資源的分配，而NodeManager則負責資源的供給和隔離，將cpu、內存等包裝稱container，一個container代表最小計算資源。

ResourceManager將某個NodeManager上資源分配給任務（這就是所謂的“資源調度”）后，NodeManager需按照要求為任務提供相應的資源，甚至保證這些資源應具有獨占性，為任務運行提供基礎的保證，這就是所謂的資源隔離。

yarn中可以通過yarn-site.xml中設置如下幾個參數達到管理內存的目的：

• yarn.nodemanager.resource.memory-mb 默認值：8192M NM總的可用物理內存，以MB為單位。一旦設置，不可動態修改
• yarn.nodemanager.resource.cpu-vcores 默認值：8 可分配的CPU個數
• yarn.scheduler.minimum-allocation-mb 默認值：1024 可申請的最少內存資源，以MB為單位
• yarn.scheduler.maximum-allocation-mb 默認值：8192 可申請的最大內存資源，以MB為單位
• yarn.scheduler.minimum-allocation-vcores 默認值：1 可申請的最小虛擬CPU個數
• yarn.scheduler.maximum-allocation-vcores 默認值：32 可申請的最 大虛擬CPU個數

yarn.nodemanager.resource.memory-mb與yarn.nodemanager.resource.cpu-vcores的值不會根據系統資源自動設置，需要手動設置，如果系統內存小於8G 、cpu小於8個，最好手動設置

spark 參數設置

spark 執行任務是executor，一個executor可以運行多個task。一個Executor對應一個JVM進程。從Spark的角度看，Executor占用的內存分為兩部分：ExecutorMemory和MemoryOverhead。

• spark.driver.memory 默認值：1g ； 分配給driver process的jvm堆內存大小，SparkContext將會在這里初始化，命令行中可通過 --driver-memory指定，也可通過配置文件指定一個固定值
• spark.driver.cores 默認值：1 ； 分配給driver process的核心數量，只在cluster模式下
• spark.driver.memoryOverhead 默認值：driverMemory * 0.10, with minimum of 384； 用於driver process的啟停jvm內存大小
• spark.executor.cores 默認值：1 ； 分配給executor process的核心數量，命令行中可通過 executor-cores指定
• spark.executor.memory 默認值：1g ； 分配給每個executor的程序的內存大小，命令行中可通過 --executor-memory指定
• spark.executor.memoryOverhead 默認值：executorMemory * 0.10, with minimum of 384； jvm非堆內存的開銷，一般占max（executorMemory *10%，384M）大小

Spark On YARN資源分配策略

當在YARN上運行Spark作業，每個Spark executor作為一個YARN容器運行。Spark可以使得多個Tasks在同一個容器里面運行。

• 對於集群中每個節點首先需要找出nodemanager管理的資源大小，總的資源-系統需求資源-hbase、HDFS等需求資源=nodemanager管理資源
• 划分內存資源，有上文中jvm資源需求等於executor.memory（JVM堆資源）+executor.memoryOverhead（JVM非堆需要資源），也就是一個executor需要的內存資源=--executor-memory+max（executorMemory *10%，384M）。同時這個值需要通過yarn申請，必須落在minimum-allocation-mb與maximum-allocation-mb之間
• 划分cpu資源，通過executor.cores指定executor可擁有的cpu個數，也就是task可並行運行的個數，一般小於5
• 計算executor個數。設置num-executors

對於client模式：nodemanager管理資源>=executor個數executor資源（內存+cpu） 對於cluster模式：nodemanager管理資源>=executor個數executor（內存+cpu）+driver資源（內存+cpu）

總結

本篇主要講解了Spark on yarn 的提交作業方式， Yarn client 與 Yarn cluster的原理，同時以實例講解了Hadoop Yarn 的Rest api 以及如何調用提交spark的作業，最后講了下Spark參數設置、Yarn的參數設置以及Spark on Yarn的資源分配策略。