Pyspark

注：大家觉得博客好的话，别忘了点赞收藏呀，本人每周都会更新关于人工智能和大数据相关的内容，内容多为原创，Python Java Scala SQL 代码，CV NLP 推荐系统等，Spark Flink Kafka Hbase Hive Flume等等~写的都是纯干货，各种顶会的论文解读，一起进步。
今天继续和大家分享一下Pyspark基础入门6
#博学谷IT学习技术支持

前言

今天和大家分享的是Spark RDD的共享变量

一、广播变量

在Driver端定义一个共享的变量,如果不使用广播变量, 各个线程在运行的时候, 都需要将这个变量拷贝到自己的线程中, 对网络传输, 内存的使用都是一种浪费, 而且影响效率

如果使用广播变量, 会将变量在每个executor上放置一份, 各个线程直接读取executor上的变量即可, 不需要拉取到Task中, 减少副本的数量, 对网络和内存都降低了, 从而提升效率

广播变量是只读的, 各个Task只能读取数据, 不能修改

相关的API:
设置广播变量: 广播变量的对象 = sc.broadcast(变量值)
获取广播变量: 广播变量的对象.value

1.使用方法

from pyspark import SparkContext, SparkConf
import os# 锁定远端环境, 确保环境统一
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'if __name__ == '__main__':print("演示广播变量相关使用")# 1. 创建SparkContext对象:conf = SparkConf().setAppName('sougou').setMaster('local[*]')sc = SparkContext(conf=conf)# a = 100broadcast = sc.broadcast(100)# 2- 初始化数据:rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7])# 3- 处理数据:# 需求: 请为每一个元素累加一个值def fn1(num):return num + broadcast.valuerdd_res = rdd.map(fn1)print(rdd_res.collect())

二、累加器

Spark提供累加器, 可以用于实现全局累加计算的操作, 比如全局共计操作了多少个数据, 可以使用累加器实现

累加器是由Driver端设置初始值, 在各个Task中进行累加操作, 最终在Driver端获取结果

Task只能累加操作, 不能读取累加器的值

相关的API:
1- 在Driver端设置累加器初始值:
acc = sc.accumulator(初始值)
2- 在Task(RDD)中: 执行累加操作
acc.add(累加值)
3- 在Driver中获取值
acc.value

1.使用方法

from pyspark import SparkContext, SparkConf
import os# 锁定远端环境, 确保环境统一
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'if __name__ == '__main__':print("演示累加器相关的操作:")# 1. 创建SparkContext对象:conf = SparkConf().setAppName('sougou').setMaster('local[*]')sc = SparkContext(conf=conf)# 定义一个累加的变量# agg = 0acc = sc.accumulator(0)# 2- 初始化数据:rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7])# 3- 执行相关的操作:# 需求: 对每个元素进行 +1 返回, 在执行操作的过程汇总, 需要统计共计对多少个数据进行 +1操作def fn1(num):acc.add(1)return num + 1rdd_res = rdd.map(fn1)# 3- 获取结果print(rdd_res.collect())print(acc.value)

如果后续多次调用action算子, 会导致累加器重复累加操作主要原因: 每一次调度action算子, 都会触发一个Job任务执行, 每一个Job任务都要重新对象其所依赖的所有RDD进行整个计算操作, 从而导致累加器重复累加解决方案:在调用累加器后的RDD上, 对其设置缓存操作, 即可解决问题, 但是不能单独设置checkpoint, checkpoint和累加器无法直接共用, 可以通过缓存 + 累加器的思路来解决
主要原因: 每一次调度action算子, 都会触发一个Job任务执行, 每一个Job任务都要重新对象其所依赖的所有RDD进行整个计算操作, 从而导致累加器重复累加解决方案:在调用累加器后的RDD上, 对其设置缓存操作, 即可解决问题, 但是不能单独设置checkpoint, checkpoint和累加器无法直接共用, 可以通过缓存 + 累加器的思路来解决

总结

今天和大家分享了RDD的两种共享变量。