上次講到避免使用shuffle類算子，接下來繼續(xù)

5、使用map-side預(yù)聚合的shuffle操作

如果因?yàn)闃I(yè)務(wù)需要，一定要使用shuffle操作，無法用map類的算子來替代，那么盡量使用可以map-side預(yù)聚合的算子。

所謂的map-side預(yù)聚合，說的是在每個節(jié)點(diǎn)本地對相同的key進(jìn)行一次聚合操作，類似于MapReduce中的本地combiner。 map-side預(yù)聚合之后，每個節(jié)點(diǎn)本地就只會有一條相同的key，因?yàn)槎鄺l相同的key都被聚合起來了。其他節(jié)點(diǎn)在拉取所有節(jié)點(diǎn)上的相同key時(shí)，就 會大大減少需要拉取的數(shù)據(jù)數(shù)量，從而也就減少了磁盤IO以及網(wǎng)絡(luò)傳輸開銷。通常來說，在可能的情況下，建議使用reduceByKey或者 aggregateByKey算子來替代掉groupByKey算子。因?yàn)閞educeByKey和aggregateByKey算子都會使用用戶自定義 的函數(shù)對每個節(jié)點(diǎn)本地的相同key進(jìn)行預(yù)聚合。而groupByKey算子是不會進(jìn)行預(yù)聚合的，全量的數(shù)據(jù)會在集群的各個節(jié)點(diǎn)之間分發(fā)和傳輸，性能相對來 說比較差。

比如如下兩幅圖，就是典型的例子，分別基于reduceByKey和groupByKey進(jìn)行單詞計(jì)數(shù)。其中第一張圖是groupByKey的原理 圖，可以看到，沒有進(jìn)行任何本地聚合時(shí)，所有數(shù)據(jù)都會在集群節(jié)點(diǎn)之間傳輸；第二張圖是reduceByKey的原理圖，可以看到，每個節(jié)點(diǎn)本地的相同 key數(shù)據(jù)，都進(jìn)行了預(yù)聚合，然后才傳輸?shù)狡渌?jié)點(diǎn)上進(jìn)行全局聚合。

6、使用高性能算子

使用reduceByKey/aggregateByKey替代groupByKey

詳情見上一優(yōu)化原則

使用mapPartitions替代普通map

mapPartitions類的算子，一次函數(shù)調(diào)用會處理一個partition所有的數(shù)據(jù)，而不是一次函數(shù)調(diào)用處理一條，性能相對來說會高一些。?但是有的時(shí)候，使用mapPartitions會出現(xiàn)OOM（內(nèi)存溢出）的問題。因?yàn)閱未魏瘮?shù)調(diào)用就要處理掉一個partition所有的數(shù)據(jù)，如果內(nèi)存 不夠，垃圾回收時(shí)是無法回收掉太多對象的，很可能出現(xiàn)OOM異常。所以使用這類操作時(shí)要慎重！

使用foreachPartitions替代foreach

原理類似于“使用mapPartitions替代map”，也是一次函數(shù)調(diào)用處理一個partition的所有數(shù)據(jù)，而不是一次函數(shù)調(diào)用處理一條數(shù) 據(jù)。在實(shí)踐中發(fā)現(xiàn)，foreachPartitions類的算子，對性能的提升還是很有幫助的。比如在foreach函數(shù)中，將RDD中所有數(shù)據(jù)寫 MySQL，那么如果是普通的foreach算子，就會一條數(shù)據(jù)一條數(shù)據(jù)地寫，每次函數(shù)調(diào)用可能就會創(chuàng)建一個數(shù)據(jù)庫連接，此時(shí)就勢必會頻繁地創(chuàng)建和銷毀數(shù) 據(jù)庫連接，性能是非常低下；但是如果用foreachPartitions算子一次性處理一個partition的數(shù)據(jù)，那么對于每個 partition，只要創(chuàng)建一個數(shù)據(jù)庫連接即可，然后執(zhí)行批量插入操作，此時(shí)性能是比較高的。實(shí)踐中發(fā)現(xiàn)，對于1萬條左右的數(shù)據(jù)量寫MySQL，性能可 以提升30%以上。

使用filter之后進(jìn)行coalesce操作

通常對一個RDD執(zhí)行filter算子過濾掉RDD中較多數(shù)據(jù)后（比如30%以上的數(shù)據(jù)），建議使用coalesce算子，手動減少RDD的 partition數(shù)量，將RDD中的數(shù)據(jù)壓縮到更少的partition中去。因?yàn)閒ilter之后，RDD的每個partition中都會有很多數(shù)據(jù) 被過濾掉，此時(shí)如果照常進(jìn)行后續(xù)的計(jì)算，其實(shí)每個task處理的partition中的數(shù)據(jù)量并不是很多，有一點(diǎn)資源浪費(fèi)，而且此時(shí)處理的task越多， 可能速度反而越慢。因此用coalesce減少partition數(shù)量，將RDD中的數(shù)據(jù)壓縮到更少的partition之后，只要使用更少的task即 可處理完所有的partition。在某些場景下，對于性能的提升會有一定的幫助。

使用repartitionAndSortWithinPartitions替代repartition與sort類操作

repartitionAndSortWithinPartitions是Spark官網(wǎng)推薦的一個算子，官方建議，如果需要在 repartition重分區(qū)之后，還要進(jìn)行排序，建議直接使用repartitionAndSortWithinPartitions算子。因?yàn)樵撍阕?可以一邊進(jìn)行重分區(qū)的shuffle操作，一邊進(jìn)行排序。shuffle與sort兩個操作同時(shí)進(jìn)行，比先shuffle再sort來說，性能可能是要高 的。

7、廣播大變量

有時(shí)在開發(fā)過程中，會遇到需要在算子函數(shù)中使用外部變量的場景（尤其是大變量，比如100M以上的大集合），那么此時(shí)就應(yīng)該使用Spark的廣播（Broadcast）功能來提升性能。

在算子函數(shù)中使用到外部變量時(shí)，默認(rèn)情況下，Spark會將該變量復(fù)制多個副本，通過網(wǎng)絡(luò)傳輸?shù)絫ask中，此時(shí)每個task都有一個變量副本。如 果變量本身比較大的話（比如100M，甚至1G），那么大量的變量副本在網(wǎng)絡(luò)中傳輸?shù)男阅荛_銷，以及在各個節(jié)點(diǎn)的Executor中占用過多內(nèi)存導(dǎo)致的頻 繁GC，都會極大地影響性能。

因此對于上述情況，如果使用的外部變量比較大，建議使用Spark的廣播功能，對該變量進(jìn)行廣播。廣播后的變量，會保證每個Executor的內(nèi)存 中，只駐留一份變量副本，而Executor中的task執(zhí)行時(shí)共享該Executor中的那份變量副本。這樣的話，可以大大減少變量副本的數(shù)量，從而減 少網(wǎng)絡(luò)傳輸?shù)男阅荛_銷，并減少對Executor內(nèi)存的占用開銷，降低GC的頻率。

8、使用Kryo優(yōu)化序列化性能

序列化詳情參考 ? http://www.itdecent.cn/p/e1e19aa51eeb

9、優(yōu)化數(shù)據(jù)結(jié)構(gòu)

Java中，有三種類型比較耗費(fèi)內(nèi)存：

對象，每個Java對象都有對象頭、引用等額外的信息，因此比較占用內(nèi)存空間。

字符串，每個字符串內(nèi)部都有一個字符數(shù)組以及長度等額外信息。

集合類型，比如HashMap、LinkedList等，因?yàn)榧项愋蛢?nèi)部通常會使用一些內(nèi)部類來封裝集合元素，比如Map.Entry。

因此Spark官方建議，在Spark編碼實(shí)現(xiàn)中，特別是對于算子函數(shù)中的代碼，盡量不要使用上述三種數(shù)據(jù)結(jié)構(gòu)，盡量使用字符串替代對象，使用原始類型（比如Int、Long）替代字符串，使用數(shù)組替代集合類型，這樣盡可能地減少內(nèi)存占用，從而降低GC頻率，提升性能。

但是在編碼實(shí)踐中發(fā)現(xiàn)，要做到該原則其實(shí)并不容易。因?yàn)槲覀兺瑫r(shí)要考慮到代碼的可維護(hù)性，如果一個代碼中，完全沒有任何對象抽象，全部是字符 串拼接的方式，那么對于后續(xù)的代碼維護(hù)和修改，無疑是一場巨大的災(zāi)難。同理，如果所有操作都基于數(shù)組實(shí)現(xiàn)，而不使用HashMap、LinkedList 等集合類型，那么對于我們的編碼難度以及代碼可維護(hù)性，也是一個極大的挑戰(zhàn)。因此建議，在可能以及合適的情況下，使用占用內(nèi)存較少的數(shù)據(jù)結(jié)構(gòu)，但是前 提是要保證代碼的可維護(hù)性。