做過一段時間spark的應用開發的小伙伴都會漸漸發現，很沒趣，因為都是調API。那么，真的是沒趣嗎，還是說你本身沒有去深入研究呢？通過本文你就會發現自己沒成長是哪的問題了。
浪尖會花一段時間在spark的算子原理分析和高性能使用對比方面的分析，並將這些知識放到浪尖的知識星球里。有興趣的同學掃描底部二維碼或者點擊閱讀原文加入星球。昨天將spark1.6源碼閱讀視頻已經上傳到星球里。
順便打個廣告，浪尖開了知乎，有興趣的可以關注一下，搜索浪尖即可。
1. mappartition粗介
本問主要想講如何高效的使用mappartition。
首先，說到mappartition大家肯定想到的是map和MapPartition的對比。網上這類教程很多了，以前浪尖也發過類似的，比如
對比foreach和foreachpartition
主要是map和foreach這類的是針對一個元素調用一次我們的函數，也即是我們的函數參數是單個元素，假如函數內部存在數據庫鏈接、文件等的創建及關閉，那么會導致處理每個元素時創建一次鏈接或者句柄，導致性能底下，很多初學者犯過這種毛病。
而foreachpartition是針對每個分區調用一次我們的函數，也即是我們函數傳入的參數是整個分區數據的迭代器，這樣避免了創建過多的臨時鏈接等，提升了性能。
下面的例子都是1-20這20個數字,經過map或者MapPartition然后返回a*3。
2. map栗子
val a=sc.parallelize(1 to 20, 2)def mapTerFunc(a : Int) : Int={ ? ?a*3}val mapResult=a.map(mapTerFunc) println(mapResult.collect().mkString(","))
結果
3,6,9,12,15,18,21,24,27,30,33,36,39,42,45,48,51,54,57,60
3. mappartitions低效用法
大家通常的做法都是申請一個迭代器buffer，將處理后的數據加入迭代器buffer，然后返回迭代器。如下面的demo。
val a=sc.parallelize(1 to 20, 2) ?def terFunc(iter: Iterator[Int]) : Iterator[Int]={ ? ?var res=List[Int]() ? ?while (iter.hasNext) ? ?{ ? ? ?val cur=iter.next; ? ? ?res.::=(cur*3) ; ? ?} ? ?res.iterator ?}val result=a.mapPartitions(terFunc) println(result.collect().mkString(","))
結果
30,27,24,21,18,15,12,9,6,3,60,57,54,51,48,45,42,39,36,33
4. mappartitions高效用法
注意，3中的例子，會在mappartition執行期間，在內存中定義一個數組並且將緩存所有的數據。四級成績假如數據集比較大，內存不足，會導致內存溢出，任務失敗。 對於這樣的案例，Spark的RDD不支持像mapreduce那些有上下文的寫方法。其實，浪尖有個方法是無需緩存數據的，那就是自定義一個迭代器類。如下例：
class CustomIterator(iter: Iterator[Int]) extends Iterator[Int] { ? ?def hasNext : Boolean={ ? ? ?iter.hasNext ? ?} ? ?def next : Int={ ? ? ? ?val cur=iter.next ? ? ? ? cur*3 ? ?} ?} ? ?val result=a.mapPartitions(v=> new CustomIterator(v)) ?println(result.collect().mkString(","))
結果：
3,6,9,12,15,18,21,24,27,30,33,36,39,42,45,48,51,54,57,60
是不是腦洞大開，要多學習的，同志們。
【完】
文章來源：https://blog.csdn.net/rlnLo2pNEfx9c/article/details/81613035