Spark on Yarn詳解

時間  2020-06-01
標籤 spark yarn 詳解 欄目 Spark 简体版
原文   原文鏈接

Spark 能夠跑在不少集羣上，好比跑在local上，跑在Standalone上，跑在Apache Mesos上，跑在Hadoop YARN上等等。無論你Spark跑在什麼上面，它的代碼都是同樣的，區別只是–master的時候不同。其中Spark on YARN是工做中或生產上用的很是多的一種運行模式。今天主要對Spark on Yarn 這種方式作講解。html

做業提交

Standalone模式的提交

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

以後將測試代碼打包成jar包，上傳到服務器，使用spark-submit提交做業。node

提交命令：web

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須要的自定義參數，若是有多個，空格隔開便可。這裏只須要一個參數，用來指定程序要加載的文件目錄。apache

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

yarn模式兩種提交任務方式

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

yarn-client提交任務方式

配置：服務器

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

image

注意client只須要有Spark的安裝包便可提交任務，不須要其餘配置（好比slaves）jvm

提交命令

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

image

執行流程圖：

image

執行流程：

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端。

小結： 一、Yarn-client模式一樣是適用於測試，由於Driver運行在本地，Driver會與yarn集羣中的Executor進行大量的通訊，會形成客戶機網卡流量的大量增長.

二、 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的日誌裏面：

image

執行流程圖：

image

執行流程：

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 /

leTLy4.png

啓動Yarn

start-yarn

訪問 http://localhost:8088，能夠看到yarn 啓動正常。

lev1wF.md.png

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

建立spark任務

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

lexNNQ.md.png

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

lmldPJ.md.png

查看spark庫zip文件屬性

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

lmlXGj.md.png

提交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的資源分配策略。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程