Spark中foreachPartition和mapPartitions的区别

ゝ一纸荒年。 2022-10-16 07:44 143阅读 0赞

# Spark-core 中foreachPartition和mapPartitions的区别 #

1.  foreachPartition应该属于action运算操作，而mapPartitions是在Transformation中，所以是转化操作，
2.  此外在应用场景上区别是mapPartitions可以获取返回值，继续在返回RDD上做其他的操作，而foreachPartition因为没有返回值并且是action操作，所以使用它一般都是在程序末尾比如说要落地数据到存储系统中如mysql，es，或者hbase中，可以用它。
3.  当然在Transformation中也可以落地数据，但是它必须依赖action操作来触发它，因为Transformation操作是延迟执行的，如果没有任何action方法来触发，那么Transformation操作是不会被执行的，这一点需要注意。

一个foreachPartition例子：

spark-core

val sparkConf=new SparkConf()
         val sc=new SparkContext(sparkConf)
          sparkConf.setAppName("spark demo example ")
        val rdd=sc.parallelize(Seq(1,2,3,4,5),3)
        
        rdd.foreachPartition(partiton=>{ 
          // partiton.size 不能执行这个方法，否则下面的foreach方法里面会没有数据，
          //因为iterator只能被执行一次
          partiton.foreach(line=>{ 
            //save(line) 落地数据
          })
     
        })
        
        sc.stop()

SparkStreaming

dstream.foreachRDD {  rdd =>
      rdd.foreachPartition {  partitionOfRecords =>
        val connection = createNewConnection()
        partitionOfRecords.foreach(record => connection.send(record))
        connection.close()
      }
    }

一个mapPartitions例子

val sparkConf=new SparkConf()
         val sc=new SparkContext(sparkConf)
          sparkConf.setAppName("spark demo example ")
        val rdd=sc.parallelize(Seq(1,2,3,4,5),3)
     
        rdd.mapPartitions(partiton=>{ 
          //只能用map，不能用foreach，因为foreach没有返回值
          partiton.map(line=>{ 
            //save line
          }
          )
        })
     
        rdd.count()//需要action，来触发执行
        sc.stop()

最后，需要注意一点，如果操作是iterator类型，我们是不能在循环外打印这个iterator的size，一旦执行size方法，相当于iterato就会被执行，所以后续的foreach你会发现是空值的，切记iterator迭代器只能被执行一次。