foreachRDD、foreach和foreachPartition的区别

首先foreachRDD 是sparkStreaming的算子，

另外两个是sparkCore的算子。

一、首先说一下foreach和foreachPartition的区别，类似于map和mapPartition的区别。

（一）map和mapPartition的区别

map是对RDD的每一个元素进行操作，mapPartition是对每个partition的迭代器进行操作。

MapPartitions的优点：
如果是普通的map，比如一个partition中有1万条数据。ok，那么你的function要执行和计算1万次。使用MapPartitions操作之后，一个task仅仅会执行一次function，function一次接收所有的partition数据。只要执行一次就可以了，性能比较高。如果在map过程中需要频繁创建额外的对象(例如将rdd中的数据通过jdbc写入数据库,map需要为每个元素创建一个链接而mapPartition为每个partition创建一个链接),则mapPartitions效率比map高的多。SparkSql或DataFrame默认会对程序进行mapPartition的优化。
MapPartitions的缺点：
如果是普通的map操作，一次function的执行就处理一条数据；那么如果内存不够用的情况下， 比如处理了1千条数据了，那么这个时候内存不够了，那么就可以将已经处理完的1千条数据从内存里面垃圾回收掉，或者用其他方法，腾出空间来吧。
所以说普通的map操作通常不会导致内存的OOM异常。 但是MapPartitions操作，对于大量数据来说，比如甚至一个partition，100万数据，一次传入一个function以后，那么可能一下子内存不够，但是又没有办法去腾出内存空间来，可能就OOM，内存溢出。

（二）foreach和foreachPartition的区别

1）foreachPartition

        foreachPartition是spark-core的action算子,该算子源码中的注释是:Applies a function func to each parition of this RDD.(将函数func应用于此RDD的每个分区)

foreachPartition是对每个partition中的iterator实行迭代的处理,通过用户传入的function(即函数func)iterator进行内容的处理,源码中函数func传入的参数是一个迭代器,也就是说在functionPartition中函数处理的是分区迭代器,而非具体的数据.

2）foreach

      foreach也是spark-core的action算子,与foreachPartition类似的是,foreach也是对每个partition中的iterator实行迭代处理,通过用户传入的function(即函数func)对iterator进行内容的处理,而不同的是,函数func中的参数传入的不再是一个迭代器,而是每次foreach得到的一个rdd的kv实例,也就是具体的数据，源码中的注释是:Applies  a function fun to all elements of this RDD.(将函数func用于此RDD的所有元素).

（三）foreachRDD与另外两个的区别

foreachRDD是sparkStreaming的OutputOperation算子。但是foreachRDD并不会触发立即处理，必须在碰到sparkcore的foreach或者foreachPartition算子后，才会触发action动作。同时要注意，func的应用在的driver端进行，而不是Executor端进行。

通常将数据写入到外部系统需要创建一个连接对象（如 TCP 连接到远程服务器），并用它来 发送数据到远程系统。出于这个目的，开发者可能在不经意间在 Spark Driver 端创建了连接 对象，并尝试使用它保存 RDD 中的记录到 Spark Worker 上，如下面代码：

lines.foreachRDD(rdd => {

val connection = createNewConnection() // 在 driver 端执行

rdd.foreach( record => {

connection.send(record) // 在 worker 端执行 })

})

这是不正确的，这需要连接对象进行序列化并从 Driver 端发送到 Worker 上。连接对象很少 在不同机器间进行这种操作，此错误可能表现为序列化错误（连接对不可序列化），初始化 错误（连接对象在需要在 Worker 上进行需要初始化）等等，正确的解决办法是在 worker 上 创建的连接对象。

通常情况下，创建一个连接对象有时间和资源开销。因此，创建和销毁的每条记录的连接对 象可能招致不必要的资源开销，并显著降低系统整体的吞吐量 。一个更好的解决方案是使 用 rdd.foreachPartition 方法创建一个单独的连接对象，然后使用该连接对象输出的所有 RDD 分区中的数据到外部系统。

lines.foreachRDD(rdd => {

rdd.foreachPartition( record => {

val connection = createNewConnection() // 在 worker 端创建 connection 对象

connection.send(record) // 在 worker 端执行

}) })

这缓解了创建多条记录连接的开销。最后，还可以进一步通过在多个 RDDs/batches 上重用 连接对象进行优化。一个保持连接对象的静态池可以重用在多个批处理的 RDD 上将其输出 到外部系统，从而进一步降低了开销。

需要注意的是，在静态池中的连接应该按需延迟创建，这样可以更有效地把数据发送到外部 系统。另外需要要注意的是：DStreams 延迟执行的，就像 RDD 的操作是由 actions 触发一样。 默认情况下，输出操作会按照它们在 Streaming 应用程序中定义的顺序一个个执行。