Spark效能調優之Shuffle調優總結

2021-08-15 11:01:36 字數 4816 閱讀 4272

spark底層shuffle的傳輸方式是使用netty傳輸,netty在進行網路傳輸的過程會申請堆外記憶體(netty是零拷貝),所以使用了堆外記憶體。

shuffle過程中常出現的問題

常見問題一:reduce oom?

問題原因:

reduce task 去map端獲取資料,reduce一邊拉取資料一邊聚合,reduce端有一塊聚合記憶體(executor memory * 0.2),也就是這塊記憶體不夠

解決辦法:

1.增加reduce 聚合操作的記憶體的比例

2.增加executor memory的大小  --executor-memory5g

3.減少reduce task每次拉取的資料量設定spak.reducer.maxsizeinflight24m, 拉取的次數就多了,因此建立連線的次數增多,有可能會連線不上(正好趕上map task端進行gc)

常見問題二:錯誤描述--shuffle file cannot find   or   executor lost

•什麼時候需要調節executor的堆外記憶體大小?

• shuffle file cannot find (dagscheduler,resubmitting task)

• executor lost

• task lost

• out of memory

問題原因:

1.map task所執行的executor記憶體不足,導致executor掛掉了,executor裡面的blockmanager就掛掉了,導致connectionmanager不能用,也就無法建立連線,從而不能拉取資料

2.executor並沒有掛掉

2.1 blockmanage之間的連線失敗(map task所執行的executor正在gc)

2.2建立連線成功,map task所執行的executor正在gc

3.reduce task向driver中的mapoutputtracker獲取shuffle file位置的時候出現了問題

解決辦法:

1.增大executor記憶體(即堆內記憶體) ,申請的堆外記憶體也會隨之增加--executor-memory 5g

2.增大堆外記憶體 --conf spark.yarn.executor.memoryoverhead 2048m--conf spark.executor.memoryoverhead 2048m

(預設申請的堆外記憶體是executor記憶體的10%,真正處理大資料的時候,這裡都會出現問題,導致spark作業反覆崩潰,無法執行;此時就會去調節這個引數,到至少1g(1024m),甚至說2g、4g)

)

buffer 32k    //緩衝區預設大小為32k  

sparkconf.set("spark.shuffle.file.buffer","64k")

reduce 

48m 

//reduce端拉取資料的時候,預設大小是48m 

sparkconf.set("spark.reducer.maxsizeinflight","96m")

spark.shuffle.file.buffer

預設值:32k

引數說明:該引數用於設定shuffle write task的bufferedoutputstream的buffer緩衝大小。將資料寫到磁碟檔案之前,會先寫入buffer緩衝中,待緩衝寫滿之後,才會溢寫到磁碟。

調優建議:如果作業可用的記憶體資源較為充足的話,可以適當增加這個引數的大小(比如64k),從而減少shuffle write過程中溢寫磁碟檔案的次數,也就可以減少磁碟io次數,進而提公升效能。在實踐中發現,合理調節該引數,效能會有1%~5%的提公升。

spark.reducer.maxsizeinflight

預設值:48m

引數說明:該引數用於設定shuffle read task的buffer緩衝大小,而這個buffer緩衝決定了每次能夠拉取多少資料。

調優建議:如果作業可用的記憶體資源較為充足的話,可以適當增加這個引數的大小(比如96m),從而減少拉取資料的次數,也就可以減少網路傳輸的次數,進而提公升效能。在實踐中發現,合理調節該引數,效能會有1%~5%的提公升。

錯誤:reduce oom

reduce task去map拉資料,reduce 一邊拉資料一邊聚合   reduce段有一塊聚合記憶體(executor memory * 0.2)

解決辦法:1、增加reduce 聚合的記憶體的比例  設定spark.shuffle.memoryfraction

2、 增加executor memory的大小  --executor-memory 5g

3、減少reduce task每次拉取的資料量  設定spark.reducer.maxsizeinflight  24m

spark.shuffle.io.maxretries

預設值:3

引數說明:shuffle read task從shuffle write task所在節點拉取屬於自己的資料時,如果因為網路異常導致拉取失敗,是會自動進行重試的。該引數就代表了可以重試的最大次數。如果在指定次數之內拉取還是沒有成功,就可能會導致作業執行失敗。

調優建議:對於那些包含了特別耗時的shuffle操作的作業,建議增加重試最大次數(比如60次),以避免由於jvm的full gc或者網路不穩定等因素導致的資料拉取失敗。在實踐中發現,對於針對超大資料量(數十億~上百億)的shuffle過程,調節該引數可以大幅度提公升穩定性。

shuffle file not find    taskscheduler不負責重試task,由dagscheduler負責重試stage

spark.shuffle.io.retrywait

預設值:5s

引數說明:具體解釋同上,該引數代表了每次重試拉取資料的等待間隔,預設是5s。

調優建議:建議加大間隔時長(比如60s),以增加shuffle操作的穩定性。

spark.shuffle.memoryfraction

預設值:0.2

引數說明:該引數代表了executor記憶體中,分配給shuffle read task進行聚合操作的記憶體比例,預設是20%。

調優建議:在資源引數調優中講解過這個引數。如果記憶體充足,而且很少使用持久化操作,建議調高這個比例,給shuffle read的聚合操作更多記憶體,以避免由於記憶體不足導致聚合過程中頻繁讀寫磁碟。在實踐中發現,合理調節該引數可以將效能提公升10%左右。

spark.shuffle.manager

預設值:sort

引數說明:該引數用於設定shufflemanager的型別。spark 1.5以後,有三個可選項:hash、sort和tungsten-sort。hashshufflemanager是spark 1.2以前的預設選項,但是spark 1.2以及之後的版本預設都是sortshufflemanager了。tungsten-sort與sort類似,但是使用了tungsten計畫中的堆外記憶體管理機制,記憶體使用效率更高。

調優建議:由於sortshufflemanager缺省會對資料進行排序,因此如果你的業務邏輯中需要該排序機制的話,則使用預設的sortshufflemanager就可以;而如果你的業務邏輯不需要對資料進行排序,那麼建議參考後面的幾個引數調優,通過bypass機制或優化的hashshufflemanager來避免排序操作,同時提供較好的磁碟讀寫效能。這裡要注意的是,tungsten-sort要慎用,因為之前發現了一些相應的bug。

spark.shuffle.sort.bypassmergethreshold

預設值:200

引數說明:當shufflemanager為sortshufflemanager時,如果shuffle read task的數量小於這個閾值(預設是200),則shuffle write過程中不會進行排序操作,而是直接按照未經優化的hashshufflemanager的方式去寫資料,但是最後會將每個task產生的所有臨時磁碟檔案都合併成乙個檔案,並會建立單獨的索引檔案。

調優建議:當你使用sortshufflemanager時,如果的確不需要排序操作,那麼建議將這個引數調大一些,大於shuffle read task的數量。那麼此時就會自動啟用bypass機制,map-side就不會進行排序了,減少了排序的效能開銷。但是這種方式下,依然會產生大量的磁碟檔案,因此shuffle write效能有待提高。

spark.shuffle.consolidatefiles

預設值:false

引數說明:如果使用hashshufflemanager,該引數有效。如果設定為true,那麼就會開啟consolidate機制,會大幅度合併shuffle write的輸出檔案,對於shuffle read task數量特別多的情況下,這種方法可以極大地減少磁碟io開銷,提公升效能。

調優建議:如果的確不需要sortshufflemanager的排序機制,那麼除了使用bypass機制,還可以嘗試將spark.shffle.manager引數手動指定為hash,使用hashshufflemanager,同時開啟consolidate機制。在實踐中嘗試過,發現其效能比開啟了bypass機制的sortshufflemanager要高出10%~30%。

超越永無止境

Spark效能調優 之 運算元調優(二)

map 表示每乙個元素 rrd.foreache 表示每乙個元素 rrd.forpartitions 表示每個分割槽的資料組成的迭代器 在生產環境中,通常使用foreachpartition運算元來完成資料庫的寫入,通過foreachpartition運算元的特性,可以優化寫資料庫的效能。如果使用f...

spark 效能調優

核心調優引數如下 num executors executor memory executor cores driver memory spark.default.parallelizm spark.storage.memoryfraction spark.shuffle.memoryfractio...

Spark效能調優

日常工作使用spark處理業務問題中不可避免的都會碰到需要對spark的效能進行調優的情況,這裡就介紹一下對spark的效能調優。1.調節記憶體分配 因為在spark中堆記憶體被劃分為兩塊,一塊是給rdd的cache和persist操作rdd資料快取使用的,另一塊是給spark運算元函式使用的,函式...