云計算大數(shù)據(jù)培訓之Spark調優(yōu)（3）

使用背景：當RDD 引用到了一個外部變量，并且這個外部變量數(shù)據(jù)量還不小，同時這個RDD對應的task數(shù)量特別多，那么此時使用廣播變量就在合適不過了

使用原理：因為每個task都要拷貝一個副本到executor去執(zhí)行，那么我們可以想象一下，如果有1000個task在某個worker上執(zhí)行，而這個副本有100M，那么就意味著我們要拷貝100G的數(shù)據(jù)到某個worker上去執(zhí)行，這樣的話會大大消耗我們的網(wǎng)絡流量，同時會加大executor的內存消耗，從而增加我們spark作業(yè)的運行時間，大大降低了spark作業(yè)的運行效率，增加了作業(yè)失敗的概率，如何解決以上的問題？也就是什么時候該使用廣播變量，我們可以將這種大的外部變量做成廣播變量，那么每個executor的內存中只會有一個外部變量，這樣的一個副本，針對所有的task都是共享的，這樣的話就減少了網(wǎng)絡流量的消耗，降低了executor的內存消耗，提高了spark 作業(yè)運行效率和 縮短了運行時間，同時降低了作業(yè)失敗的概率

使用過程： 
1、某個executor的第一個task先執(zhí)行的時候，首先從自己的blockManager中查找外部變量，如果沒有就從鄰居的executor的blockManager的內存中獲取這個外部變量，如果還是獲取不到，就從driver端獲取，拷貝這個外部變量到本地的executor的blockManager中 
2、當這個executor的其他task執(zhí)行時，就不需要再從外面獲取這個外部變量的副本了，直接從本地的blockManager中獲取即可

【注意】廣播變量可讀，不可寫

五，使用Kryo序列化 
默認情況下，spark內部是使用Java的序列化機制，objectOutputStream/ObjectInputStream對象輸入輸出機制，來進行序列化 
這種默認的序列化機制的好處在于，處理起來比較方便，也不需要手動去做什么事情，只是，你在算子里面使用的變量，必須實現(xiàn)Serializable接口的，可序列化即可 
但是缺點是，默認的序列化的效率不高，序列化速度比較慢，序列化以后的數(shù)據(jù)，占用的內存空間相對還是比較大的 
Spark支持使用Kryo序列化機制。 
Kyro序列化機制，比默認的Java序列化機制，速度要快，序列化后的數(shù)據(jù)要更小，大概是Java序列化機制的1/10。所以Kryo序列化優(yōu)化以后，可以讓網(wǎng)絡傳輸?shù)臄?shù)據(jù)變少，在集群中耗費的內存資源大大減少

Kryo序列化機制，一旦啟用以后，會生效的幾個地方： 
1、算子函數(shù)中使用到的外部變量 
2、持久化RDD時進行序列化，StorageLevel.MEMORY_ONLY_SER 
3、Shuffle

1、算子函數(shù)中使用到的外部變量，使用Kryo以后：優(yōu)化網(wǎng)絡傳輸?shù)男阅?，可以?yōu)化集群中的內存的占用和消耗 
2、持久化RDD，優(yōu)化內存的占用和消耗，持久化RDD占用的內存越少，task執(zhí)行的時候，創(chuàng)建的對象，就不至于頻繁的占滿內存，頻繁的發(fā)生GC 
3、Shuffle：可以優(yōu)化網(wǎng)絡傳輸?shù)男阅?nbsp;
Kryo之所以沒有被作為默認的序列化類庫的原因： 
主要是因為Kryo要求，如果要達到他的最佳性能的話，那么就一定要注冊你自定義的類（比如，你的算子函數(shù)中使用到了外部自定義的對象變量，這時，就要求必須注冊你的類，否則Kryo達不到最佳的性能）。

首先，第一步：在sparkConf中設置一個屬性： 
spark.serializer，org.apache.spark.serializer.KryoSerializer類 
例如：sparkConf.set(“spark.serializer”,”org.apahe.spark.serializer.KryoSerializer”)

其次，第二步：注冊你使用到的。需要通過Kryo序列化的，一些自定義類： 
sparkConf.registerKryoClasses() 
例如：sparkConf.registerKryoClasses(Array(classOf[CategorySecondSort],classOf[…]))