对于spark而言，这两个算子应该算是用的最多的了，而关于这两个算子的数据处理效率问题，讨论其实一直都有很多，尤其对于很多刚入行不久的小伙伴来说，会单纯的以为map这个函数因为针对的是一条数据，而mapPartitions针对的是整个分区的数据，所以会自然的以为：mapPartitions会比map的效率高。

而我一直坚持的认为，这两者单纯从对数据的处理效率来说是不存在高低优劣势之分的，这两者最大的区别仅仅只是暴露给使用者的API中面对的数据粒度不一样而已，这个区别只在你对数据处理过程中，如果涉及到【外部IO】情况下才会区别明显。

最典型就是在算子内部调用数据库连接，如果这个连接是在算子内部创建的，那么会涉及到一个创建【过频】问题，但是这问题其实解决办法有很多，连接池+单例就能很容易搞定。

对于这两者的效率问题，我们来分析一波。

|理论分析

我们知道，对于spark而言，数据进行分布式处理的单位是partition，当我们运行一个spark任务时，如果对一个executor分配N个CPU核心，那么就意味着这个executor进程可以并行处理N个partition的数据，但是对于任何一个独立的partition而言，你的业务处理逻辑都是要串行运用到每一条数据上的，因此无论对于map还是mapPartitions来说，其运行业务处理逻辑的效率理论是一样的。

只不过对于开发者而言，当你用这两者提供的api时，暴露的数据粒度不一样，map面对的是一个具体的数据对象，而mapPartition面对的是一个迭代器，但是从spark的运行层面来说，无论你用map还是mapPartitions都是调用MapPartitionsRDD这个隐藏类中的对应方法。

这个是理论分析，但是实际情况真的这样吗？

那么不妨，咱来验证一下！

|验证准备

我用之前提供给大家做实战项目的数据集来充当验证的数据源，然后将该数据经过一番简单的转换处理，最终把结果以CSV格式写入到HDFS中，其处理流程如下图：

为了比较快达到验证目的，我截取了约500MB数据作为测试集，数据量约400W条，为了尽可能让验证效果准确，对数据的处理逻辑只用map和mapPartitions算子，对于其数据处理逻辑完全一样。

且在实际操作过程中，为了尽可能避免程序对数据处理过程中的其他因素干扰，对数据的处理逻辑不涉及任何IO操作、不涉及shuffle，单纯只是CPU运算，而且为了防止本地电脑的网络抖动问题，将程序放到我的测试集群中运行。其代码分别如下：

1，map算子代码：

object MapTest {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("MapTest")/*.setMaster("local")*/
        val spark = SparkSession.builder().config(conf).getOrCreate()
        val rawDF = spark.read.csv(args(0))//读取HDFS的数据源
        import spark.implicits._
        rawDF.printSchema() //spark job 1
        rawDF.show() //spark job 2
        val resultDF = rawDF.map(row => {
            val clientIP = row.getAs[String]("_c0")
            val domain = row.getAs[String]("_c1").toLowerCase//将域名转成小写
            val time = row.getAs[String]("_c2")
            val targetIP = row.getAs[String]("_c3")
            val rcode = row.getAs[String]("_c4")
            val queryType = row.getAs[String]("_c5")
            val authRecord = if (row.getAs[String]("_c6") == null ) "" else row.getAs[String]("_c6").toLowerCase
            val addMsg = if (row.getAs[String]("_c7") == null ) "" else row.getAs[String]("_c7")
            val dnsIP = row.getAs[String]("_c8")
            (clientIP,domain,time,targetIP,rcode,queryType,authRecord,addMsg,dnsIP)
        }).toDF("client_ip","domain","time","target_ip","rcode","query_type","authority_record","add_msg","dns_ip")

        /**将转换后的数据写入HDFS*/
        resultDF.write.csv(args(1))//spark job 3
    }
}

2，mapPartitions算子代码：

object MapPartitionTest {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("MapPartitionTest")/*.setMaster("local")*/
        val spark = SparkSession.builder().config(conf).getOrCreate()
        val rawDF = spark.read/*.option("header",true)*/.csv(args(0))

        import spark.implicits._
        rawDF.printSchema()
        rawDF.show()
        val resultDF = rawDF.mapPartitions(iterator => {
            val array = new mutable.ArrayBuffer[(String,String,String,String,String,String,String,String,String)]
            while (iterator.hasNext){
                val next = iterator.next()
                val clientIP = next.getAs[String]("_c0")
                val domain = next.getAs[String]("_c1").toLowerCase//将域名转成小写
                val time = next.getAs[String]("_c2")
                val targetIP = next.getAs[String]("_c3")
                val rcode = next.getAs[String]("_c4")
                val queryType = next.getAs[String]("_c5")
                val authRecord = if (next.getAs[String]("_c6") == null ) "" else next.getAs[String]("_c6").toLowerCase
                val addMsg = if (next.getAs[String]("_c7") == null ) "" else next.getAs[String]("_c7")
                val dnsIP = next.getAs[String]("_c8")
                array.+=((clientIP,domain,time,targetIP,rcode,queryType,authRecord,addMsg,dnsIP))
            }
            array.toIterator
        }).toDF("client_ip","domain","time","target_ip","rcode","query_type","authority_record","add_msg","dns_ip")

        resultDF.write.csv(args(1))
    }
}

然后为了屏蔽因为分布式任务，不同服务器之间的资源分配速度不同，导致不同executor之间运行效率会存在差异而带来的干扰，实验只用1个executor，1个CPU核心。启动命令如下：

spark-submit --master yarn --deploy-mode client --conf spark.driver.memory=1g --conf spark.yarn.max.executor.failures=1 --conf spark.memory.fraction=0.8 --conf spark.task.maxFailures=1 --class com.anryg.bigdata.test.map_pk_mappartition.MapTest --executor-cores 1 --num-executors 1 --executor-memory 2g --name MapTest --jars ./spark-coding-1.0-SNAPSHOT-with-dependencies.jar ./spark-coding-1.0-SNAPSHOT.jar /the/path/to/source/data /dir/of/output