Spark性能优化用foreachPartition还是与foreach

这篇文章主要介绍“Spark性能优化用foreachPartition还是与foreach”，在日常操作中，相信很多人在Spark性能优化用foreachPartition还是与foreach问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Spark性能优化用foreachPartition还是与foreach”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

首先，我们对比一下foreachPartition和foreach两个方法的实现，有什么不同的地方：

  def foreach(f: T => Unit): Unit = withScope {
    val cleanF = sc.clean(f)
    sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
  }

  def foreachPartition(f: Iterator[T] => Unit): Unit = withScope {
    val cleanF = sc.clean(f)
    sc.runJob(this, (iter: Iterator[T]) => cleanF(iter))
  }

2个方法，参数都是一个函数文本，不同的是foreach当中，函数文本希望的参数是T，也就是RDD当中的元素类型；foreachPartition当中，函数文本希望的参数是Iterator[T]，也就是一个partition。

而在内部实现上，其实是大同小异的。对于foreachPartition而言，直接在各个partition上运行传入的函数文本；而对于foreach而言，是把传入的函数文本，交给各个partition的foreach去执行。
我们查看一些spark性能优化指南，会提到用foreachPartition替代foreach，有助于性能的提高。那么我们要怎样来理解这句话呢？看看下面这段代码：

rdd.foreach { x => {

    val dbClient = new DBClient

    dbClient.ins(x)

}}

在上面这段代码当中，针对RDD当中的每一条数据，都会new一个db client，这样的效率，显然是无比底下的。正确的写法应该是这个样子的：

rdd.foreachPartition { part => {

    val dbClient = new DBClient

    part.foreach{ x => {

        dbClient.ins(x)

    }}

}}
那么这种写法究竟好在哪里，还是要从spark的核心概念开始讲起，我们都知道spark是一个分布式的实时计算系统，而RDD是分布式计算的基础，而partition分区又是这个当中的关键，比如我们搭建一个3*4core的spark集群，对于一个大任务而言，我们往往是希望有12个线程一起来完成这个任务，用下面的代码来构建rdd就能够达到我们的目的：

val rdd = sc.textFile("hdfs://master:9000/woozoom/mavlink1.log", 12)

注意红色字体的部分，代表着构建出来的rdd的分区数量。之后，rdd.foreachPartition，spark集群会把12个分区分别交给12个线程来分别进行处理。结合上面的代码，dbClient 会在每个线程当中分别构建，会有12个db client被构建。

那么有没有另一种可能性，我们只构建一个db client，12个线程都用这一个db client来执行数据库操作，像下面这样：

val dbClient = new DBClient

rdd.foreach { x => {   

    dbClient.ins(x)

}}

要这么写，需要有2个前提：1、dbClient 是线程安全的，2、dbClient 实现了java的序列化接口。而在很多情况下，例如在对hbase进行访问的时候，这两个条件都是不满足的。

到此，关于“Spark性能优化用foreachPartition还是与foreach”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注亿速云网站，小编会继续努力为大家带来更多实用的文章！