map和mapPartitions目錄map和mapPartitionmapFun和mapPartitionsFun21222map和mapPartitions1map和mapPartitions

map是對(duì)rdd中的每一個(gè)元素進(jìn)行操作；mapPartitions則是對(duì)rdd中的每個(gè)分區(qū)的迭代器進(jìn)行操作；map和mapPartitions

加入函數(shù)內(nèi)存在數(shù)據(jù)庫連接、文件創(chuàng)建關(guān)閉等操作，則map每次調(diào)用都會(huì)創(chuàng)建一次連接和文件句柄，導(dǎo)致性能十分低，而mapPartitions只在每個(gè)Partition中創(chuàng)建一次，因此效率高。map和mapPartitions

如果是普通的map，比如一個(gè)partition中有1萬條數(shù)據(jù)。那么function要執(zhí)行和計(jì)算1萬次。使用MapPartitions操作之后，一個(gè)task僅僅會(huì)執(zhí)行一次function，function一次接收所有的partition數(shù)據(jù)。只要執(zhí)行一次就可以了，性能比較高。如果在map過程中需要頻繁創(chuàng)建額外的對(duì)象(例如將rdd中的數(shù)據(jù)通過jdbc寫入數(shù)據(jù)庫,map需要為每個(gè)元素創(chuàng)建一個(gè)鏈接而mapPartition為每個(gè)partition創(chuàng)建一個(gè)鏈接),則mapPartitions效率比map高的多。map和mapPartitions

mapPartitions存在OOM問題，即內(nèi)存溢出，例如當(dāng)一個(gè)partition的數(shù)據(jù)量超出內(nèi)存限制時(shí)一次性載入函數(shù)中會(huì)導(dǎo)致OOM，而map就不會(huì)存在這樣的情況因?yàn)樗看翁幚硪粭l數(shù)據(jù)，當(dāng)內(nèi)存不足時(shí)可以將以前處理好的數(shù)據(jù)從內(nèi)存中垃圾回收，騰出空間。mapFun和mapPartitionFun2mapFun和mapPartitionFun

map針對(duì)RDD中的每一個(gè)元素調(diào)用一次函數(shù)，而mapPartitions針對(duì)RDD中每個(gè)Partition調(diào)用一次函數(shù)，假設(shè)RDD有N個(gè)元素，有M個(gè)分區(qū)，那么map的函數(shù)將被調(diào)用N次，而mapPartitions被調(diào)用M次。在編寫map和mapPartitions的函數(shù)時(shí)需要有一個(gè)概念，map傳入的是RDD的一個(gè)元素，而mapPartitions傳入的是RDD的一個(gè)分區(qū)。mapFun和mapPartitionFun通過以下案例，可見mapFun和mapPartitionFun的區(qū)別。defmapFun(x):returnx+2defmappartitionFun(xlist):return[x+2forxinxlist]mapFun和mapPartitionFun

def_main():#setsparkcontextconf=SparkConf().setMaster("local[*]").setAppName("MyApp")sc=SparkContext(conf=conf)sc.setLogLevel("ERROR")rdd=sc.parallelize([1,2,3,4,5,6])#res=rdd.mapPartitions(mappartitionFun)