Spark項目故障總結

1.OOM問題，reduce端的緩衝大小，太大的話，吃撐了，一下過來不少數據，容易OOM，默認48，能夠改小哦。spark.reducer.maxSizeInFlight，48---》24

2.JVM-GC致使的shuffle文件拉取失敗，shuffle file not found

spark.shuffle.io.maxRetries 3 第一個參數，意思就是說，shuffle文件拉取的時候，若是沒有拉取到（拉取失敗），最多或重試幾回（會從新拉取幾回文件），默認是3次。 spark.shuffle.io.retryWait 5s 第二個參數，意思就是說，每一次重試拉取文件的時間間隔，默認是5s鍾。 默認狀況下，假如說第一個stage的executor正在進行漫長的full gc。第二個stage的executor嘗試去拉取文件，結果沒有拉取到，默認狀況下，會反覆重試拉取3次，每次間隔是五秒鐘。最多隻會等待3 * 5s = 15s。若是15s內，沒有拉取到shuffle file。就會報出shuffle file not found。

spark.shuffle.io.maxRetries 60

spark.shuffle.io.retryWait 60s

3.YARN隊列資源不足致使的application直接失敗  :

  限制提交一次做業，用線程池確保一個線程執行任務。

採用以下方案：

一、在你的J2EE（咱們這個項目裏面，spark做業的運行，以前說過了，J2EE平臺觸發的，執行spark-submit腳本），限制，同時只能提交一個spark做業到yarn上去執行，確保一個spark做業的資源確定是有的。

二、你應該採用一些簡單的調度區分的方式，好比說，你有的spark做業多是要長時間運行的，好比運行30分鐘；有的spark做業，多是短期運行的，可能就運行2分鐘。此時，都提交到一個隊列上去，確定不合適。極可能出現30分鐘的做業卡住後面一大堆2分鐘的做業。分隊列，能夠申請（跟大家的YARN、Hadoop運維的同窗申請）。你本身給本身搞兩個調度隊列。每一個隊列的根據你要執行的做業的狀況來設置。在你的J2EE程序裏面，要判斷，若是是長時間運行的做業，就乾脆都提交到某一個固定的隊列裏面去把；若是是短期運行的做業，就統一提交到另一個隊列裏面去。這樣，避免了長時間運行的做業，阻塞了短期運行的做業。

三、你的隊列裏面，不管什麼時候，只會有一個做業在裏面運行。那麼此時，就應該用咱們以前講過的性能調優的手段，去將每一個隊列能承載的最大的資源，分配給你的每個spark做業，好比80個executor；6G的內存；3個cpu core。儘可能讓你的spark做業每一次運行，都達到最滿的資源使用率，最快的速度，最好的性能；並行度，240個cpu core，720個task。

四、在J2EE中，經過線程池的方式（一個線程池對應一個資源隊列），來實現上述咱們說的方案。

4.解決yarn-cluster模式的JVM內存溢出沒法執行問題，（PermGen 永久代內存溢出）

既然是JVM的PermGen永久代內存溢出，那麼就是內存不夠用。我們呢，就給yarn-cluster模式下的，driver的PermGen多設置一些。 spark-submit腳本中，加入如下配置便可： --conf spark.driver.extraJavaOptions="-XX:PermSize=128M -XX:MaxPermSize=256M" 這個就設置了driver永久代的大小，默認是128M，最大是256M。那麼，這樣的話，就能夠基本保證你的spark做業不會出現上述的yarn-cluster模式致使的永久代內存溢出的問題。

5.錯誤的持久化方式以及checkpoint的使用

checkpoint原理：

一、在代碼中，用SparkContext，設置一個checkpoint目錄，能夠是一個容錯文件系統的目錄，好比hdfs；

二、在代碼中，對須要進行checkpoint的rdd，執行RDD.checkpoint()；

三、RDDCheckpointData（spark內部的API），接管你的RDD，會標記爲marked for checkpoint，準備進行checkpoint

四、你的job運行完以後，會調用一個finalRDD.doCheckpoint()方法，會順着rdd lineage，回溯掃描，發現有標記爲待checkpoint的rdd，就會進行二次標記，inProgressCheckpoint，正在接受checkpoint操做

五、job執行完以後，就會啓動一個內部的新job，去將標記爲inProgressCheckpoint的rdd的數據，都寫入hdfs文件中。（備註，若是rdd以前cache過，會直接從緩存中獲取數據，寫入hdfs中；若是沒有cache過，那麼就會從新計算一遍這個rdd，再checkpoint）

六、將checkpoint過的rdd以前的依賴rdd，改爲一個CheckpointRDD*，強制改變你的rdd的lineage。後面若是rdd的cache數據獲取失敗，直接會經過它的上游CheckpointRDD，去容錯的文件系統，好比hdfs，中，獲取checkpoint的數據。

數據傾斜

一、定位緣由與出現問題的位置: 根據log去定位 出現數據傾斜的緣由，基本只多是由於發生了shuffle操做，在shuffle的過程當中，出現了數據傾斜的問題。由於某個，或者某些key對應的數據，遠遠的高於其餘的key。 一、你在本身的程序裏面找找，哪些地方用了會產生shuffle的算子，groupByKey、countByKey、reduceByKey、join

二、看log log通常會報是在你的哪一行代碼，致使了OOM異常；或者呢，看log，看看是執行到了第幾個stage！！！ 咱們這裏不會去剖析stage的劃分算法

方案一：聚合源數據

 將致使數據傾斜的一些shuffle操做，好比，groupbykey,reducebykey 或者join等，提早到ETL過程進行處理，通常狀況spark處理的源數據來之hive ETL的hive表。

方案二：過濾致使傾斜的key

 從源頭過濾掉啦

方案三：提升reduce Task的數量，調用方式時，傳入numPartitions

治標不治本的意思，由於，它沒有從根本上改變數據傾斜的本質和問題。不像第一個和第二個方案（直接避免了數據傾斜的發生）。原理沒有改變，只是說，儘量地去緩解和減輕shuffle reduce task的數據壓力，以及數據傾斜的問題。

方案四：雙重groupby 在數據量過多的key上作文章咯。把這個key作分解，打散。好比key是「hello」作個map操做，變成「hello1」「hello2」。

 使用場景 （1）groupByKey （2）reduceByKey 比較適合使用這種方式；

第一輪聚合的時候，對key進行打散，將原先同樣的key，變成不同的key，至關因而將每一個key分爲多組； 先針對多個組，進行key的局部聚合；接着，再去除掉每一個key的前綴，而後對全部的key，進行全局的聚合。 對groupByKey、reduceByKey形成的數據傾斜，有比較好的效果。

方案五：（針對join操做）reduce join轉換爲map join，適合在什麼樣的狀況下，能夠來使用？ 若是兩個RDD要進行join，其中一個RDD是比較小的。一個RDD是100萬數據，一個RDD是1萬數據。（一個RDD是1億數據，一個RDD是100萬數據） 其中一個RDD必須是比較小的，broadcast出去那個小RDD的數據之後，就會在每一個executor的block manager中都駐留一份。要確保你的內存足夠存放那個小RDD中的數據 這種方式下，根本不會發生shuffle操做，確定也不會發生數據傾斜；從根本上杜絕了join操做可能致使的數據傾斜的問題；

方案六：（針對join操做）sample採樣傾斜key進行兩次join：這種方案何時適合使用？ 優先對於join，確定是但願可以採用上一講講的，reduce join轉換map join。兩個RDD數據都比較大，那麼就不要那麼搞了。 針對你的RDD的數據，你能夠本身把它轉換成一箇中間表，或者是直接用countByKey()的方式，你能夠看一下這個RDD各個key對應的數據量；此時若是你發現整個RDD就一個，或者少數幾個key，是對應的數據量特別多；儘可能建議，好比就是一個key對應的數據量特別多。 此時能夠採用我們的這種方案，單拉出來那個最多的key；單獨進行join，儘量地將key分散到各個task上去進行join操做。 何時不適用呢？ 若是一個RDD中，致使數據傾斜的key，特別多；那麼此時，最好仍是不要這樣了；仍是使用咱們最後一個方案，終極的join數據傾斜的解決方案。

方案七：使用隨機數以及擴容表進行join：兩個rdd都很大的jion操做，key進行打散，擴容10倍，在join