Apache Spark-foreach Vs foreachPartitions何时使用什么？

Question

我想知道，与foreach方法相比，foreach是否会由于更高级别的并行性而导致更好的性能，考虑到我正在通过RDD对累加器变量执行一些求和。

Answer 1

foreach和foreachpartitions是操作。

用于调用具有副作用的操作的泛型函数。对于RDD中的每个元素，它调用传递的函数。这通常用于操作累加器或写入外部存储。

注意：在foreach()之外修改累加器以外的变量可能会导致未定义的行为。有关更多详细信息，请参见了解闭包。

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Long = 10

null

/**
    * Insert in to database using foreach partition.
    *
    * @param sqlDatabaseConnectionString
    * @param sqlTableName
    */
  def insertToTable(sqlDatabaseConnectionString: String, sqlTableName: String): Unit = {

    //numPartitions = number of simultaneous DB connections you can planning to give

datframe.repartition(numofpartitionsyouwant)

    val tableHeader: String = dataFrame.columns.mkString(",")
    dataFrame.foreachPartition { partition =>
      // Note : Each partition one connection (more better way is to use connection pools)
      val sqlExecutorConnection: Connection = DriverManager.getConnection(sqlDatabaseConnectionString)
      //Batch size of 1000 is used since some databases cant use batch size more than 1000 for ex : Azure sql
      partition.grouped(1000).foreach {
        group =>
          val insertString: scala.collection.mutable.StringBuilder = new scala.collection.mutable.StringBuilder()
          group.foreach {
            record => insertString.append("('" + record.mkString(",") + "'),")
          }

          sqlExecutorConnection.createStatement()
            .executeUpdate(f"INSERT INTO [$sqlTableName] ($tableHeader) VALUES "
              + insertString.stripSuffix(","))
      }


      sqlExecutorConnection.close() // close the connection so that connections wont exhaust.
    }
  }

• 示例2:

foreachpartition与SparkStreams（dstreams）和kafka producer一起使用

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
// only once per partition You can safely share a thread-safe Kafka //producer instance.
    val producer = createKafkaProducer()
    partitionOfRecords.foreach { message =>
      producer.send(message)
    }
    producer.close()
  }
}

注意：如果您想避免这种每个分区创建一次生产者的方式，最好的方法是使用sparkcontext.broadcast广播生产者，因为Kafka生产者是异步的，并且在发送之前会大量缓冲数据。

累加器示例片段可以使用它...您可以通过它来测试性能

     test("Foreach - Spark") {
        import spark.implicits._
        var accum = sc.longAccumulator
        sc.parallelize(Seq(1,2,3)).foreach(x => accum.add(x))
        assert(accum.value == 6L)
      }

      test("Foreach partition - Spark") {
        import spark.implicits._
        var accum = sc.longAccumulator
        sc.parallelize(Seq(1,2,3)).foreachPartition(x => x.foreach(accum.add(_)))
        assert(accum.value == 6L)
      }

foreachpartition对分区的操作，因此显然，edge比foreach更好

当您访问昂贵的资源（如数据库连接或kafka producer等）时，应使用foreachpartition。它将每个分区初始化一个而不是每个元素初始化一个(foreach)。当涉及到蓄能器时，您可以通过上述测试方法来测量性能，这些测试方法在蓄能器的情况下也应该工作得更快。

还有...参见map vs mappartitions，它们有相似的概念，但它们是转换。