Spark算子[01]：foreach，foreachPartition

概述

RDD.foreachPartition/foreach这两个action的操作:

这两个action主要用于对每个partition中的iterator实行迭代的处理。通过用户传入的function对iterator进行内容的处理。

foreach的操作

在foreach中，传入一个function，这个函数的传入参数就是每个partition中，每次的foreach得到的一个rdd的kv实例，也就是具体的内容，这种处理你并不知道这个iterator的foreach什么时候结束，只能是foreach的过程中，你得到一条数据，就处理一条数据。

由下面的源码中，foreach操作是直接调迭代rdd中每一条数据进行function操作。

/**
* Applies a function f to all elements of this RDD.
* 将函数应用在RDD的所有元素；
*/
def foreach(f: T => Unit): Unit = withScope {
val cleanF = sc.clean(f)
//runJob job的运行
sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
}

示例说明:

val list = new ArrayBuffer()
Rdd.foreach(record => {
list += record
If (list.size >= 10000) {
list.flush....
}
})

上面这段示例代码中，如果这么使用就会存在一个问题：

迭代的最后，list的结果可能还没有达到10000条，这个时候，你在内部的处理的flush部分就不会执行，也就是迭代的最后如果没有达到10000的数据就会丢失。

所以在foreach中，一般就是拿到一条数据进行下处理Rdd.foreach(record => {record._1 == a return})。

foreachPartition操作

这个函数也是根据传入的function进行处理，但不同处在于，这里function的传入参数是一个partition对应数据的iterator，而不是直接使用iterator的foreach。

/**
* Applies a function f to each partition of this RDD.
* 将函数f应用于该RDD的每个分区。
*/
def foreachPartition(f: Iterator[T] => Unit): Unit = withScope {
val cleanF = sc.clean(f)
sc.runJob(this, (iter: Iterator[T]) => cleanF(iter))
}

示例代码:

这种情况下，如果是上面foreach的示例代码中list这个片段在这个action中就能够正常的去处理。

Val list = new ArrayBuffer
rdd.foreachPartition(it => {
It.foreach(r => {
List += r
If (list.size > 10000) flush
})
If (list.size > 0) flush
})

总结

最后说下这两个action的区别:

Foreach与foreachPartition都是在每个partition中对iterator进行操作，不同的是，foreach是直接在每个partition中直接对iterator执行foreach操作，而传入的function只是在foreach内部使用；

而foreachPartition是在每个partition中把iterator给传入的function，让function自己对iterator进行处理。

在实践中发现，foreachPartitions类的算子，对性能的提升还是很有帮助的。比如在foreach函数中，将RDD中所有数据写MySQL，那么如果是普通的foreach算子，就会一条数据一条数据地写，每次函数调用可能就会创建一个数据库连接，此时就势必会频繁地创建和销毁数据库连接，性能是非常低下；但是如果用foreachPartitions算子一次性处理一个partition的数据，那么对于每个partition，只要创建一个数据库连接即可，然后执
922e
行批量插入操作，此时性能是比较高的。