spark中map与mapPartitions区别

Posted 2021-01-15 wubinghuan

在spark中，map与mapPartitions两个函数都是比较常用，这里使用代码来解释一下两者区别

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

object MapAndPartitions {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setAppName("map_mapPartitions_demo").setMaster("local"))
    val arrayRDD =sc.parallelize(Array(1,2,3,4,5,6,7,8,9))

    //map函数每次处理一个/行数据
    arrayRDD.map(element=>{
      element
    }).foreach(println)

    //mapPartitions每次处理一批数据
    //将 arrayRDD分成x批数据进行处理
    //elements是其中一批数据
    //mapPartitions返回一批数据（iterator）
    arrayRDD.mapPartitions(elements=>{
      var result = new ArrayBuffer[Int]()
      elements.foreach(element=>{
        result.+=(element)
      })
      result.iterator
    }).foreach(println)
  }
}

两个函数最终处理得到的结果是一样的

mapPartitions比较适合需要分批处理数据的情况，比如将数据插入某个表，每批数据只需要开启一次数据库连接，大大减少了连接开支，伪代码如下：

    arrayRDD.mapPartitions(datas=>{
      dbConnect = getDbConnect() //获取数据库连接
      datas.foreach(data=>{
        dbConnect.insert(data) //循环插入数据
      })
      dbConnect.commit() //提交数据库事务
      dbConnect.close() //关闭数据库连接
    })