共享变量:

 

共享变量

通常情况下，当向Spark操作(如map,reduce)传递一个函数时，它会在一个远程集群节点上执行，它会使用函数中所有变量的副本。这些变量被复制到所有的机器上，远程机器上并没有被更新的变量会向驱动程序回传。在任务之间使用通用的，支持读写的共享变量是低效的。尽管如此，Spark提供了两种有限类型的共享变量，广播变量和累加器。

 

广播变量

广播变量允许程序员将一个只读的变量缓存在每台机器上，而不用在任务之间传递变量。广播变量可被用于有效地给每个节点一个大输入数据集的副本。Spark还尝试使用高效地广播算法来分发变量，进而减少通信的开销。

import org.apache.spark.SparkContextimport org.apache.spark.broadcast.Broadcastimport org.apache.spark.rdd.RDDimport org.apache.spark.sql.SparkSessionimport org.apache.spark.util.LongAccumulator

object TransformationsDemo {  def main(args: Array[String]): Unit = {    val sparkSession = SparkSession.builder().appName("TransformationsDemo").master("local[1]").getOrCreate()    val sc = sparkSession.sparkContext    testBroadcast(sc)  }  private def testBroadcast(sc: SparkContext) = {    val factor = 3    val factorBroadcast: Broadcast[Int] = sc.broadcast(factor)    var data = sc.parallelize(Array(1, 2, 3, 4, 5), 2)    data.map(num => num * factorBroadcast.value).collect().foreach(println)  }}

运行结果：

 

累加器

累加器是仅仅被相关操作累加的变量，因此可以在并行中被有效地支持。它可以被用来实现计数器和总和。

import org.apache.spark.SparkContextimport org.apache.spark.broadcast.Broadcastimport org.apache.spark.rdd.RDDimport org.apache.spark.sql.SparkSessionimport org.apache.spark.util.LongAccumulatorobject TransformationsDemo {  def main(args: Array[String]): Unit = {    val sparkSession = SparkSession.builder().appName("TransformationsDemo").master("local[1]").getOrCreate()    val sc = sparkSession.sparkContext    testAccumulator(sc)  }  private def testAccumulator(sc: SparkContext) = {    val sum: LongAccumulator = sc.longAccumulator("My Accumulator")    var data = sc.parallelize(Array(1, 2, 3, 4, 5), 2)    data.foreach(num => sum.add(num))    println(sum.value)  }}

运行结果：