临沂 网站建设,孟州网站开发app,微网站建设哪里便宜,旅游类网站建设的结论第七章 共享变量
在默认情况下#xff0c;当Spark在集群的多个不同节点的多个任务上并行运行一个函数时#xff0c;它会把函数中涉及到的每个变量#xff0c;在每个任务上都生成一个副本。但是#xff0c;有时候需要在多个任务之间共享变量#xff0c;或者在任务(Task)和…第七章 共享变量
在默认情况下当Spark在集群的多个不同节点的多个任务上并行运行一个函数时它会把函数中涉及到的每个变量在每个任务上都生成一个副本。但是有时候需要在多个任务之间共享变量或者在任务(Task)和任务控制节点(Driver Program)之间共享变量。
为了满足这种需求Spark提供了两种类型的变量 1、广播变量Broadcast Variables
广播变量用来把变量在所有节点的内存之间进行共享在每个机器上缓存一个只读的变量而不是为机器上的每个任务都生成一个副本 2、累加器Accumulators累加器支持在所有不同节点之间进行累加计算(比如计数或者求和)
官方文档http://spark.apache.org/docs/2.4.5/rdd-programming-guide.html#shared-variables
7.1 广播变量
广播变量允许开发人员在每个节点Worker or Executor缓存只读变量而不是在Task之间传递这些变量。使用广播变量能够高效地在集群每个节点创建大数据集的副本。同时Spark还使用高效的广播算法分发这些变量从而减少通信的开销。
可以通过调用sc.broadcast(v)创建一个广播变量该广播变量的值封装在v变量中可使用获取 该变量value的方法进行访问。
7.2 累加器
Spark提供的Accumulator主要用于多个节点对一个变量进行共享性的操作。Accumulator只提供了累加的功能即确提供了多个task对一个变量并行操作的功能。但是task只能对Accumulator进行累加操作不能读取Accumulator的值只有Driver程序可以读取Accumulator的值。创建的Accumulator变量的值能够在Spark Web UI上看到在创建时应该尽量为其命名。 Spark内置了三种类型的Accumulator分别是LongAccumulator用来累加整数型DoubleAccumulator用来累加浮点型CollectionAccumulator用来累加集合元素。
当内置的Accumulator无法满足要求时可以继承AccumulatorV2实现自定义的累加器。实现自定义累加器的步骤
第一步、继承AccumulatorV2实现相关方法第二步、创建自定义Accumulator的实例然后在SparkContext上注册它
官方提供实例如下
7.3 案例演示
以词频统计WordCount程序为例假设处理的数据如下所示包括非单词符合统计数据词频时过滤非单词的符合并且统计总的格式。 实现功能 第一、过滤非单词符合
非单词符合存储列表List中 使用广播变量广播列表
第二、累计统计非单词符号出现次数
定义一个LongAccumulator累加器进行计数 范例演示完整代码如下
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}
/**
* 基于Spark框架使用Scala语言编程实现词频统计WordCount程序将符号数据过滤并统计出现的次数
* -a. 过滤标点符号数据
* 使用广播变量
* -b. 统计出标点符号数据出现次数
* 使用累加器
*/
object SparkSharedVariableTest {
def main(args: Array[String]): Unit {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext {
// 1.a 创建SparkConf对象设置应用的配置信息
val sparkConf: SparkConf new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix($))
.setMaster(local[2])
// 1.b 传递SparkConf对象构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel(WARN)
// a. 读取文件数据
val datasRDD: RDD[String] sc.textFile(datas/filter/datas.input, minPartitions 2)
// TODO: 字典数据只要有这些单词就过滤: 特殊字符存储列表List中
val list: List[String] List(,, .,!,#,$,%)
// TODO: 通过广播变量 将列表list广播到各个Executor内存中便于多个Task使用
val listBroadcast: Broadcast[List[String]] sc.broadcast(list)
// TODO: 定义累加器记录单词为符号数据的个数
val accumulator: LongAccumulator sc.longAccumulator(number_accum)
// b. 分割单词过滤数据
val wordsRDD datasRDD
// 1)、过滤数据去除空行数据
.filter(line null ! line line.trim.length 0)
// 2)、分割单词
.flatMap(line line.trim.split(\\s))
// 3)、过滤字典数据符号数据
.filter{word
// 获取符合列表 TODO 从广播变量中获取列表list的值
val listValue listBroadcast.value
// 判断单词是否为符号数据如果是就过滤掉
val isFlag listValue.contains(word)
if(isFlag){
// TODO: 如果单词为符号数据累加器加1
accumulator.add(1L)
}
!isFlag
}
val resultRDD: RDD[(String, Int)] wordsRDD
// 转换为二元组
.mapPartitions{iter iter.map(word (word, 1))}
// 按照单词聚合统计
.reduceByKey((tmp, item) tmp item)
resultRDD.foreach(println)
println(s过滤符合数据的个数${accumulator.value})
// 应用程序运行结束关闭资源
sc.stop()
}
}运行应用查看WEB UI监控定义累加器的值
