1.1.1執行環境Jar包管理及和資料本地性原理

在YARN上執行Spark需要在Spark-env.sh或環境變數中配置HADOOP_CONF_DIR或YARN_CONF_DIR目錄指向Hadoop的配置檔案。

Spark-default.conf中配置Spark.YARN.jars指向hdfs上的Spark需要的jar包。如果不配置該引數，每次啟動Spark程式將會將Driver端的SPARK_HOME打包上傳分發到各個節點。配置如下例所示：

1.          spark.yarn.jarshdfs://clustername/spark/spark210/jars/*

對於分散式系統來說，由於資料可能也是分散式的，所以資料處理往往也是分散式的。要想保證效能，儘量保證資料本地性是很重要的。

分散式計算系統的精粹在於移動計算而非移動資料，但是在實際的計算過程中，總存在著移動資料的情況，除非是在叢集的所有節點上都儲存資料的副本。移動資料，將資料從一個節點移動到另一個節點進行計算，不但消耗了網路IO，也消耗了磁碟IO，降低了整個計算的效率。為了提高資料的本地性，除了優化演算法（也就是修改Spark記憶體，難度有點高），就是合理設定資料的副本。設定資料的副本，這需要通過配置引數並長期觀察執行狀態才能獲取的一個經驗值。

Spark中的資料本地性有以下幾種：

l  PROCESS_LOCAL：程序本地化。程式碼和資料在同一個程序中，也就是在同一個Executor中；計算資料的Task由Executor執行，資料在Executor的BlockManager中；效能最好。

l  NODE_LOCAL：節點本地化：程式碼和資料在同一個節點中，資料作為一個HDFS block塊，就在節點上，而Task在節點上某個Executor中執行；或者是資料和Task在一個節點上的不同Executor中；資料需要在程序間進行傳輸。也就是說，資料雖然在同一Worker 中, 但不是同一JVM中。這隱含著程序間移動資料的開銷。

l  NO_PREF：資料沒有區域性性首選位置, 它能從任何位置同等訪問。對於Task來說，資料從哪裡獲取都一樣，無好壞之分。

l  RACK_LOCAL：機架本地化。資料在不同的伺服器上，但在相同的機架。資料需要通過網路在節點之間進行傳輸。

l  ANY：資料在不同的伺服器及機架上面。這種方式效能最差。

Spark應用程式執行本身包含程式碼和資料兩部分，單機版本一般情況下很少考慮資料本地性的問題，因為資料在本地。單機版本的程式，資料本性有PROCESS_LOCAL和NODE_LOCAL之分，但也應儘量讓資料處於PROCESS_LOCAL級別。

通常，讀取資料要儘量使資料以PROCESS_LOCAL或NODE_LOCAL方式讀取。其中PROCESS_LOCAL還和Cache有關，如果RDD經常用的話，應將該RDD Cache到記憶體中。注意，由於Cache是Lazy級別的，所以必須通過一個Action的觸發，才能真正的將該RDD Cache到記憶體中。

1.1.2執行環境Jar包管理及資料本地性調優實踐

啟動Spark程式時，其他節點會自動去下載這些包並進行快取，下次啟動如果包沒有變化則會直接讀取本地快取的包。快取清理間隔在YARN-site.xml通過以下引數配置：

1.           yarn.nodemanager.localizer.cache.cleanip.interval-ms

啟動Spark-Shell，指定master為YARN，預設為client模式。在cluster模式下可以使用—supervise，如果Driver異常退出將會自行重啟。

1.         /usr/install/Spark210/bin/Spark-Shell

2.         --master YARN

3.         --deploy-mode client

4.         --Executor-memory 1g

5.         --Executor-cores 1

6.         --Driver-memory 1g

7.         --queue test (預設佇列是default)

初次講到引數配置，需要提醒的是在程式碼中的SparkConf中的配置引數具有最高優先順序，其次是Spark-Submit或Spark-Shell的引數，最後是配置檔案(conf/Spark-defaults.conf)中的引數。

本地性級別是以最近的級別開始，以最遠的級別結束。Spark 排程任務執行是要讓任務獲得最近的本地性級別的資料。然而，有時候它不得不放棄最近本地性級別, 因為一些資源上的原因，有時不得不選擇更遠的。在資料與Executor 在同一機器但是資料處理比較忙的情況下, 且存在一些空閒的Executor 遠離資料的情況下, Spark會讓Executor 等特定的時間，看其能否完成正在處理的工作。如果該Executor仍不可用, 一個新的任務將則啟用在更遠的可用節點上。即資料被傳送給那個節點。

可以給Spark中的Executor配置單種資料本地性級別可以等待的空閒時長。如下例所示：

1.         Val SparkConf = new SparkConf()

2.         SparkConf.set("Spark.locality.wait","3s")

3.         SparkConf.set("Spark.locality.wait.node","3s")

4.         SparkConf.set("Spark.locality.wait.process","3s")

5.         SparkConf.set("Spark.locality.wait.rack","3s")

預設情況下，這些等待時長都是3秒。下面講解調節這個引數的方式。推薦大家在測試的時候，先用Client模式，在本地就直接可以看到比較全的日誌。觀察日誌Spark作業的執行日誌。

裡面會顯示如下日誌：

1.          17/02/11 21:59:45 INFOscheduler.TaskSetManager: Starting Task 0.0 in Stage 0.0 (TID 0, sandbox,PROCESS_LOCAL, 1260 bytes)

觀察大部分Task的資料本地化級別：如果大多都是PROCESS_LOCAL，那就不用調節了；如果發現，很多的級別都是NODE_LOCAL、ANY，那麼最好就去調節一下資料本地化的等待時長。調節往往不可能一次到位，應該是要反覆調節，每次調節完以後，再來執行，觀察日誌。看看大部分的Task的本地化級別有沒有提升；也看看整個Spark作業的執行時間有沒有縮短。但別本末倒置，如果本地化級別提升了，但是因為大量的等待時長而導致Spark作業的執行時間反而增加，那也是不好的調節。