Spark性能調優之合理設置並行度

Spark性能調優之合理設置並行度

1.Spark的並行度指的是什么？

     spark作業中，各個stage的task的數量，也就代表了spark作業在各個階段stage的並行度！

    當分配完所能分配的最大資源了，然后對應資源去調節程序的並行度，如果並行度沒有與資源相匹配，那么導致你分配下去的資源都浪費掉了。同時並行運行，還可以讓每個task要處理的數量變少（很簡單的原理。合理設置並行度，可以充分利用集群資源，減少每個task處理數據量，而增加性能加快運行速度。）

 

    舉例：

         假如， 現在已經在spark-submit 腳本里面，給我們的spark作業分配了足夠多的資源，比如50個executor ，每個executor 有10G內存，每個executor有3個cpu core 。 基本已經達到了集群或者yarn隊列的資源上限。

task沒有設置，或者設置的很少，比如就設置了，100個task 。 50個executor ，每個executor 有3個core ，也就是說
Application 任何一個stage運行的時候，都有總數150個cpu core ，可以並行運行。但是，你現在只有100個task ，平均分配一下，每個executor 分配到2個task，ok，那么同時在運行的task，只有100個task，每個executor 只會並行運行 2個task。 每個executor 剩下的一個cpu core 就浪費掉了！你的資源，雖然分配充足了，但是問題是， 並行度沒有與資源相匹配，導致你分配下去的資源都浪費掉了。合理的並行度的設置，應該要設置的足夠大，大到可以完全合理的利用你的集群資源； 比如上面的例子，總共集群有150個cpu core ，可以並行運行150個task。那么你就應該將你的Application 的並行度，至少設置成150個，才能完全有效的利用你的集群資源，讓150個task ，並行執行，而且task增加到150個以后，即可以同時並行運行，還可以讓每個task要處理的數量變少； 比如總共 150G 的數據要處理， 如果是100個task ，每個task 要計算1.5G的數據。 現在增加到150個task，每個task只要處理1G數據。

2.如何去提高並行度？

   1、task數量，至少設置成與spark Application 的總cpu core 數量相同（最理性情況，150個core，分配150task，一起運行，差不多同一時間運行完畢）官方推薦，task數量，設置成spark Application 總cpu core數量的2~3倍 ，比如150個cpu core ，基本設置 task數量為 300~ 500. 與理性情況不同的，有些task 會運行快一點，比如50s 就完了，有些task 可能會慢一點，要一分半才運行完，所以如果你的task數量，剛好設置的跟cpu core 數量相同，可能會導致資源的浪費，因為 比如150task ，10個先運行完了，剩余140個還在運行，但是這個時候，就有10個cpu core空閑出來了，導致浪費。如果設置2~3倍，那么一個task運行完以后，另外一個task馬上補上來，盡量讓cpu core不要空閑。同時盡量提升spark運行效率和速度。提升性能。

    2、如何設置一個Spark Application的並行度？

      spark.defalut.parallelism 默認是沒有值的，如果設置了值比如說10，是在shuffle的過程才會起作用（val rdd2 = rdd1.reduceByKey(_+_) //rdd2的分區數就是10，rdd1的分區數不受這個參數的影響）

      new SparkConf().set(“spark.defalut.parallelism”,”“500)

 

     3、如果讀取的數據在HDFS上，增加block數，默認情況下split與block是一對一的，而split又與RDD中的partition對應，所以增加了block數，也就提高了並行度。

     4、RDD.repartition，給RDD重新設置partition的數量

     5、reduceByKey的算子指定partition的數量

                 val rdd2 = rdd1.reduceByKey(_+_,10)  val rdd3 = rdd2.map.filter.reduceByKey(_+_)

     6、 val rdd3 = rdd1.join（rdd2）  rdd3里面partiiton的數量是由父RDD中最多的partition數量來決定，因此使用join算子的時候，增加父RDD中partition的數量。

     7、spark.sql.shuffle.partitions //spark sql中shuffle過程中partitions的數量