spark的共享变量

因为RDD在spark中是分布式存储

1、python中定义的变量仅仅在driver中运行，在excutor中是获取不到值的——广播变量

2、若定义了一个变量进行累加，先分别在driver和excutor中进行累加，但是结果是不会主动返回给driver的——累加器

Broadcast Variables广播变量

• driver中存放python变量广播到别的excutor中

• 若不使用，就会每个task存放一个

• 不能修改，只能读

• 通过value使用该变量

if __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'D:/Program Files/Java/jdk1.8.0_271'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1/hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 获取 conf 对象# setMaster  按照什么模式运行，local  bigdata01:7077  yarn#  local[2]  使用2核CPU   * 你本地资源有多少核就用多少核#  appName 任务的名字conf = SparkConf().setMaster("local[*]").setAppName("第一个Spark程序")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字sc = SparkContext(conf=conf)fileRdd = sc.textFile("../datas/user.tsv",2)city_dict = {1: "北京",2: "上海",3: "广州",4: "深圳",5: "苏州",6: "无锡",7: "重庆",8: "厦门",9: "大理",10: "成都"}# 将一个变量广播出去，广播到executor中，不是task中city_dict_broad = sc.broadcast(city_dict)# 广播变量# <class 'pyspark.broadcast.Broadcast'>print(type(city_dict_broad ))# <class 'dict'>print(type(city_dict_broad.value))def getLine(line):list01 = line.split(" ")#cityName = city_dict.get(int(list01[3]))# 使用广播变量的变量获取数据cityName = city_dict_broad.value.get(int(list01[3]))# print(cityName)return line + " " + cityNamemapRdd = fileRdd.map(getLine)mapRdd.foreach(print)# 释放广播变量city_dict_broad.unpersist()# 使用完后，记得关闭sc.stop()

 累加器

将所有的excutor中的变量返回到driver中，进行汇总。

否则变量是放在excutor中的，而打印的是driver中，变量值不会改变。

用于修改——汇总

import os
import reimport jieba
# 导入pyspark模块
from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel"""
------------------------------------------Description : TODO：SourceFile : _06SouGou案例Author  : yangeDate  : 2024/10/31 星期四
-------------------------------------------
"""
if __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_241'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 获取 conf 对象# setMaster  按照什么模式运行，local  bigdata01:7077  yarn#  local[2]  使用2核CPU   * 你本地资源有多少核就用多少核#  appName 任务的名字conf = SparkConf().setMaster("local[*]").setAppName("搜索热词案例")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字sc = SparkContext(conf=conf)mapRdd = sc.textFile("../../datas/zuoye/sogou.tsv",minPartitions=8) \.filter(lambda line:len(re.split("\s+",line)) == 6) \.map(lambda line:(re.split("\s+",line)[0],re.split("\s+",line)[1],re.split("\s+",line)[2][1:-1])).persist(StorageLevel.MEMORY_AND_DISK_2)# 统计一天每小时点击量并按照点击量降序排序_sum = 0def sumTotalLine(tuple1):global _sum # 把_sum 设置为全局变量timeStr = tuple1[0] # 10:19:18if timeStr[0:2] == '10':_sum += 1mapRdd.foreach(lambda tuple1:sumTotalLine(tuple1))print(_sum) # 结果是0# 使用完后，记得关闭sc.stop()

上面程序最终结果是：0，因为 sum=0 是在 Driver 端的内存中的，executor 中程序再累加也是无法改变 Driver 端的结果的。下面的则为正确的

import os
import reimport jieba
# 导入pyspark模块
from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel"""
------------------------------------------Description : TODO：SourceFile : _06SouGou案例Author  : yangeDate  : 2024/10/31 星期四
-------------------------------------------
"""
if __name__ == '__main__':# 配置环境os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_241'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 获取 conf 对象# setMaster  按照什么模式运行，local  bigdata01:7077  yarn#  local[2]  使用2核CPU   * 你本地资源有多少核就用多少核#  appName 任务的名字conf = SparkConf().setMaster("local[*]").setAppName("搜索热词案例")# 假如我想设置压缩# conf.set("spark.eventLog.compression.codec","snappy")# 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字sc = SparkContext(conf=conf)accCounter = sc.accumulator(0)mapRdd = sc.textFile("../../datas/zuoye/sogou.tsv",minPartitions=8) \.filter(lambda line:len(re.split("\s+",line)) == 6) \.map(lambda line:(re.split("\s+",line)[0],re.split("\s+",line)[1],re.split("\s+",line)[2][1:-1])).persist(StorageLevel.MEMORY_AND_DISK_2)# 统计一天每小时点击量并按照点击量降序排序#_sum = 0def sumTotalLine(tuple1):#global _sum # 把_sum 设置为全局变量timeStr = tuple1[0] # 10:19:18if timeStr[0:2] == '10':accCounter.add(1)mapRdd.foreach(lambda tuple1:sumTotalLine(tuple1))print(accCounter.value) # 104694# 假如我不知道累加器这个操作，这个题目怎么做？print(mapRdd.filter(lambda tuple1: tuple1[0][0:2] == '10').count())# 使用完后，记得关闭sc.stop()