10 | 广播变量 & 累加器：共享变量是用来做什么的？

1.使用普通变量是无法得到预期结果的，因为lambda里的必须是final型的变量，那这里我用两种方法做个测试： 首先是原子类，因字数限制，只放关键代码： var normal = new AtomicInteger(0) val preRdd = wordRdd.filter(f => if(f.contains("scala")) { println(normal.incrementAndGet()) false } else { true } ) println(normal.get()) 可以看到Executor运行过程中normal是会正常累加的，但是最后println(normal.get())打印出来是0，这里简单分析下，spark会把闭包序列化后 传递给Executor，然后Executor再把闭包反序列化后作用在RDD上。因此Driver里的normal变量和Executor里的normal变量是多个进程里的多个变量。 然后使用自定义类对象 class IntHolder(var value: Int) {} 测试，报了一个Exception in thread "main" org.apache.spark.SparkException: Task not serializable 从侧面也证明，闭包里的对象是要实现序列化的。变量是多个进程里的多个变量改进下再试， class IntHolder(var value: Int) extends Serializable {} 发现每个Task都是从0开始计数，更说明每个Task里的对象是Driver丢过来的副本。这里我多想了下Task里的函数是串行执行还是并行执行的，如果我的IntHolder对象 不是线程安全的，那在Task里有无数据竞争？从我的例子看是串行运行的，但是一时找不到看哪些代码，请老师指正。 多讲两句，Spark闭包序列化和反序列化真的是很重要的知识，打开了我的视野，我感觉Spark是Faas一种非常巧妙的场景，函数式编程真的也很符合Spark 把函数作用在RDD上，存算分离的思路，UCB出品真的名不虚传。 2.Spark基本支持任何类型的广播变量，但是不支持RDD类型的广播变量，从代码中可以看到，有一个验证会判断变量是否为RDD类型，如果想要广播RDD类型的话，可以先 collect收集到driver，再作为普通集合广播。 def broadcast[T: ClassTag](value: T): Broadcast[T] = { assertNotStopped() require(!classOf[RDD[_]].isAssignableFrom(classTag[T].runtimeClass), "Can not directly broadcast RDDs; instead, call collect() and broadcast the result.") val bc = env.broadcastManager.newBroadcast[T](value, isLocal) val callSite = getCallSite logInfo("Created broadcast " + bc.id + " from " + callSite.shortForm) cleaner.foreach(_.registerBroadcastForCleanup(bc)) bc }

第一题： driver中声明的非广播变量，如果executor需要用的话，会给每一个分区拷贝一个副本，所以每个分区都会给自己的这份副本加一加一。最后这些副本随着executor 进程的结束就都丢失了。所以driver 中的这个变量仍然还是0.

python 版的代码对累加器这儿有些不一样，貌似没有longAccumulator，double之分，我这儿是这样的： file='~~~~/chapter01/wikiOfSpark.txt' lineRDD=sc.textFile(file) lineRDD.first() # 会打印出lineRDD的第一行： u'Apache Spark'，如果没有打印出来，则报错 wordRDD=lineRDD.flatMap(lambda line: line.split(" ")) # 定义累加器 ac = sc.accumulator(0) # 这儿的定义方式不一样 # 定义filter算子的判定函数f，注意，f的返回类型必须是Boolean def f(x): if x=='': ac.add(1) return False else: return True # 使用f对RDD进行过滤 cleanWordRDD = wordRDD.filter(f) kvRDD=cleanWordRDD.map(lambda word:(word,1)) wordCounts=kvRDD.reduceByKey(lambda x,y:x+y) # 获取计算结果 wordCounts.collect() # 结果：[('Spark', 63), ('Apache', 34)] # wordCounts.map(lambda (k,v):(v,k)).sortByKey(False).take(5) # 作业执行完毕，通过调用value获取累加器结果 ac.value # 79

以前我写累加器，都是手动继承AccumulatorV2，现在找到更简单的方法了😁

写程序测试了一下老师的第一个思考题，在本机的idea中可以得到正确的结果79，但是在spark- shell或者提交到yarn集群上结果却是0，好纠结，请老师解惑，代码如下。 第二个思考题，我认为广播变量是只读数据，参加计算时不能随着RDD一起变换形态破坏数据的一致性。 import org.apache.spark.sql.SparkSession object BcAcOpt { var n: Long = _ // 全局变量 def main(args: Array[String]): Unit = { val spark = SparkSession.builder.master("local[*]").appName("board").getOrCreate() val lineRDD = spark.sparkContext.textFile("wikiOfSpark.txt") println(lineRDD.partitions.length) val wordRDD = lineRDD.flatMap(word => word.split(" ")) val filterRDD = wordRDD.filter(acf) val kvRDD = filterRDD.map(word => (word, 1)) val wordCount = kvRDD.reduceByKey((x, y) => x + y) wordCount.take(10) println(n) spark.stop() } def acf(word: String): Boolean = { if ("".equals(word)) { n += 1 false } else { true } } }

老师，广播变量有"只读" / "不可变"特性， 但广播引用数据类型时, 广播的实际上是地址值，那么地址值肯定不可变，而地址值指向的内容是可变的; 我今天在idea和shell里做了一个测试，证明了这个说法是对的, 代码如下: (scala-List是不可变的，所以代码里我用了可变的scala-ListBuffer) var buffer: ListBuffer[String] = ListBuffer() var bufferBroad = spark.sparkContext.broadcast(buffer) val cleanWordRDD2: RDD[String] = wordRDD.filter(word => { if (bc.value.contains(word)) { //bc是文章里的广播变量 bufferBroad.value.append("test") //关键的一行 true } else { false } }) println(cleanWordRDD2.count())//触发这条dag println(buffer) //97(34+63)个test println(bufferBroad.value) //97(34+63)个test 老师，这个说法和我的证明没有问题吧？ 这个算不算“不可变”特性的漏洞 呢？

老师，关于第一题， 我的答案：driver和executor是不一样的进程，普通变量会拷贝副本到executor上， “原本”和“副本”没有任何关系， spark-shell打印出来的是“原本”，数值是初始化时的值. 我的问题：为何idea能计算正确呢？

老师您好。关于第一题的实验，我使用自定义类，如下： class MyCount implements Serializable{ int num; public MyCount(){ this.num = 0; } public void add(){ this.num++; } } 将它用于代替广播变量，是不行的，因为executor最后销毁了这些对象。 但是我将这个对象置为静态对象，放在Driver中再运行后，这个对象成功得到了最后的结果。 请问老师，为什么设置为静态对象就可以获取结果呢？静态对象在Driver和Executor是怎么一个工作机制？

广播变量是以executor为粒度分发的，那么累加器是怎么分发的？最终的结果又是怎么算出来的？

老师，我有一个问题：一个worker上可以有多个executor吗？在读取文件阶段，一个executor（一个进程）里可以有多个task（多个线程）读取不同分区（RDD的分区）的数据吗？那RDD的分区的定义是什么呢，是不同机器上的数据，还是？