累加器和广播变量分别在什么场景使用？

累加器和广播变量分别在什么场景使用？累加器分布式共享只写变量，如果在转换算子中调用累加器后续没有行动算子，累加器不会执行。后续如果调用了两次行动算子，会执行两次累加器出现多加的情况。

1、广播变量的使用方法介绍

解决的场景：

将Driver进程的共享数据发送给所有子节点Executor进程的每个任务中。如果不用广播变量技术，那么Driver端默认会将共享数据分发到每个【Task】中，造成网络分发压力大。

如果使用了广播变量技术，则Driver端将共享数据只会发送到每【Executor】一份。Executor中的所有【Task】都复用这个对象。要保证该共享对象是可【序列化】的。因为跨节点传输的数据都要是可序列化的。

在Driver端将共享对象广播到每个Executor：

val bc = sc.broadcast( 共享对象 )

在Executor中获取：

bc.value

2、累加器的使用方法介绍

集群中所有Executor对同一个变量进行累计操作。Spark目前只支持累【加】操作。有3种内置的累加器：【LongAccumulator】、【DoubleAccumulator】、【CollectionAccumulator】。

整数累加器使用方法

在Driver端定义整数累加器，赋初始值。

acc=sc.accumulator(0)

在Executor端每次累加1

acc+=1

或者acc.add(1)

3、综合案例

# -*- coding:utf-8 -*-
# Desc:This is Code Desc
frompyspark importSparkConf,SparkContext
importos
os.environ['SPARK_HOME'] ='/export/server/spark'
PYSPARK_PYTHON ="/root/anaconda3/bin/python3.8"
# 当存在多个版本时，不指定很可能会导致出错
os.environ["PYSPARK_PYTHON"] =PYSPARK_PYTHON
os.environ["PYSPARK_DRIVER_PYTHON"] =PYSPARK_PYTHON
if__name__=='__main__':
#需求1：从大量用户中，剔除掉黑名单用户
conf=SparkConf().setAppName('sharevalue_review')
.setMaster('local[*]')
sc=SparkContext(conf=conf)
sc.setLogLevel('WARN')
#创建大量用户
rdd_all=sc.parallelize(['zs','ls','ww','zl'])
#创建黑名单用户
black_list=['zs','ls']
#定义广播变量
bc=sc.broadcast(black_list)
#从大量用户中剔除掉黑名单用户
deffilter_black(str):
#获取广播变量
black_list2=bc.value
ifstrinblack_list2:
returnFalse
else:
returnTrue
filterd_rdd=rdd_all.filter(filter_black)
print('过滤后：')
print(filterd_rdd.collect())
#需求2：从大量数字中，挑选出带有7的数字，并计算他们的平均值。
#定义大量数字
rdd_all2=sc.parallelize(range(1,1001))
#定义累加器
#定义累加器1 ，记录有多少个7
acc =sc.accumulator(0)
#定义累加器2 ，将带有7的数字加起来
acc2=sc.accumulator(0)
deffind7(i):
globalacc
globalacc2
if'7'instr(i):
acc+=1
acc2+=i
rdd2=rdd_all2.map(find7)
rdd2.count()
num_7=acc.value
sum_7=acc2.value
avg_7=sum_7/num_7
print('带有7数字的个数是',num_7,'他们的平均数是',avg_7)

小伙伴们一定要自己亲手敲代码进行练习，以上代码不仅练习了累加器和广播变量如何使用，还涉及了函数式编程（Map、Filter）如何使用，上下文变量如何创建、如何用并行化集合的方式创建RDD等，这些练习比较综合，希望可以帮助大家学到更多的技能。