Spark编程进阶之共享变量

在向spark传递函数时，可以向驱动器中传递变量，但是集群中运行的每个任务都会得到这些变量的一个副本，跟新副本不会影响驱动器中的值。spark的两个共享变量，累加器和广播变量分别为结果聚合与广播这两种常见的通信模式突破了这一限制

累加器

查找包含电影信息包含1996这一信息出现的次数

from pyspark import SparkContext sc=SparkContext("local","blanksum") file=sc.textFile("file:///G:/Reference/source/ml-100k/u.item") #创建Accumulator[int]并初始化为0 num=sc.accumulator(0) def extractCallSigns(line): global num count = line.count("1996") if(count>0): num+=count return line.split(" ") callSigns=file.flatMap(extractCallSigns) callSigns.saveAsTextFile("record") print("Blank lines:%d" % num.value)

通过创建一个num的Accumulator[Int]对象，在rdd的每个字符串中包含count个“1996”字符串，那么num对象就加上count，但是只有在运行saveAsTextFile()行动操作后，flatMap才会开始执行

累加器有如下的调用方法 1.SparkContext.accumulator(initialValue)创建初始的累加器 2.+=方法增加累加器的值 3..value方法访问累加器的值

工作节点上的任务不能访问累加器的值

累加器与容错性

spark会自动重新执行失败的或较慢的任务来应对有错误的或者比较慢的机器 但是失败的机器对累加器的操作又该如何处理呢？ 对于在行动操作的中使用到的累加器，spark只会把每个任务对各累加器的修改应用一次，因此为了保证累加器的绝对可靠性，我们可以累加操作放在foreach()这样的行动操作之中

而在RDD转化操作中使用的累加器会发生不只一次更新，比如RDD根据谱系重算时，累加器会被多次更新

广播变量

spark第二种共享变量类型是广播变量，它可以让程序高效的向所有的工作节点发送一个较大的只读值。比如向所有的节点发送一个较大的只读查询表。

signPrefixes = sc.broadcast(loadCallSignTable())