Spark算子：RDD基本轉換操作(5)–mapPartitions、

關鍵字：Spark算子、Spark RDD基本轉換、mapPartitions、mapPartitionsWithIndex

mapPartitions

def mapPartitions[U](f: (Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]

該函數和map函數類似，只不過映射函數的參數由RDD中的每一個元素變成了RDD中每一個分區的迭代器。如果在映射的過程中需要頻繁創建額外的對象，使用mapPartitions要比map高效的過。

比如，將RDD中的所有數據通過JDBC連接寫入數據庫，如果使用map函數，可能要為每一個元素都創建一個connection，這樣開銷很大，如果使用mapPartitions，那么只需要針對每一個分區建立一個connection。

參數preservesPartitioning表示是否保留父RDD的partitioner分區信息。

 

1. var rdd1 = sc.makeRDD(1 to 5,2)
2. //rdd1有兩個分區
3. scala> var rdd3 = rdd1.mapPartitions{ x => {
4. | var result = List[Int]()
5. | var i = 0
6. | while(x.hasNext){
7. | i += x.next()
8. | }
9. | result.::(i).iterator
10. | }}
11. rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[84] at mapPartitions at :23
12.  
13. //rdd3將rdd1中每個分區中的數值累加
14. scala> rdd3.collect
15. res65: Array[Int] = Array(3, 12)
16. scala> rdd3.partitions.size
17. res66: Int = 2
18.  

mapPartitionsWithIndex

def mapPartitionsWithIndex[U](f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]

函數作用同mapPartitions，不過提供了兩個參數，第一個參數為分區的索引。

1. var rdd1 = sc.makeRDD(1 to 5,2)
2. //rdd1有兩個分區
3. var rdd2 = rdd1.mapPartitionsWithIndex{
4. (x,iter) => {
5. var result = List[String]()
6. var i = 0
7. while(iter.hasNext){
8. i += iter.next()
9. }
10. result.::(x + "|" + i).iterator
12. }
13. }
14. //rdd2將rdd1中每個分區的數字累加，並在每個分區的累加結果前面加了分區索引
15. scala> rdd2.collect
16. res13: Array[String] = Array(0|3, 1|12)
17.  
18.  

轉載請注明：lxw的大數據田地 » Spark算子：RDD基本轉換操作(5)–mapPartitions、mapPartitionsWithIndex