Pyspark
注：大家觉得博客好的话，别忘了点赞收藏呀，本人每周都会更新关于人工智能和大数据相关的内容，内容多为原创，Python Java Scala SQL 代码，CV NLP 推荐系统等，Spark Flink Kafka Hbase Hive Flume等等~写的都是纯干货，各种顶会的论文解读，一起进步。
今天继续和大家分享一下Pyspark基础入门5
#博学谷IT学习技术支持
`
文章目录
- Pyspark
- 前言
- 一、RDD的缓存
- 二、使用步骤
- 1.演示缓存的使用操作
- 三、RDD的checkpoint检查点
- 四、缓存和检查点区别
- 总结
前言
今天和大家分享的是Spark RDD的持久化方法。
一、RDD的缓存
缓存:
一般当一个RDD的计算非常的耗时|昂贵(计算规则比较复杂),或者说这个RDD需要被重复(多方)使用,此时可以将这个RDD计算完的结果缓存起来, 便于后续的使用, 从而提升效率
通过缓存也可以提升RDD的容错能力, 当后续计算失败后, 尽量不让RDD进行回溯所有的依赖链条, 从而减少重新计算时间
注意:
缓存仅仅是一种临时的存储, 缓存数据可以保存到内存(executor内存空间),也可以保存到磁盘中, 甚至支持将缓存数据保存到堆外内存中(executor以外的系统内容)
由于临时存储, 可能会存在数据丢失, 所以缓存操作, 并不会将RDD之间的依赖关系给截断掉(丢失掉),因为当缓存失效后, 可以基于原有依赖关系重新计算
缓存的API都是LAZY的, 如果需要触发缓存操作, 必须后续跟上一个action算子, 一般建议使用count
如果不添加action算子, 只有当后续遇到第一个action算子后, 才会触发缓存
二、使用步骤
设置缓存的API:
rdd.cache(): 执行缓存操作 仅能将数据缓存到内存中
rdd.persist(缓存的级别(位置)): 执行缓存操作, 默认将数据缓存到内存中, 当然也可以自定义缓存位置
手动清理缓存的API:
rdd.unpersist()
默认情况下, 当整个Spark应用程序执行完成后, 缓存也会自动失效的, 自动删除
常用的缓存级别:
MEMORY_ONLY : 仅缓存到内存中
DISK_ONLY: 仅缓存到磁盘
MEMORY_AND_DISK: 内存 + 磁盘 优先缓存到内存中, 当内存不足的时候, 剩余数据缓存到磁盘中
OFF_HEAP: 缓存到堆外内存
最为常用的: MEMORY_AND_DISK
1.演示缓存的使用操作
import time
import jieba
from pyspark import SparkContext, SparkConf, StorageLevel
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'
"""
清洗需求:
需要先对数据进行清洗转换处理操作, 清洗掉为空的数据,
以及数据字段个数不足6个的数据, 并且将每一行的数据放置到一个元组中,
元组中每一个元素就是一个字段的数据
"""
def xuqiu1():
# 需求一: 统计每个关键词出现了多少次, 获取前10个
res = rdd_map \
.flatMap(lambda field_tuple: jieba.cut(field_tuple[2])) \
.map(lambda keyWord: (keyWord, 1)) \
.reduceByKey(lambda agg, curr: agg + curr) \
.sortBy(lambda res_tup: res_tup[1], ascending=False).take(10)
print(res)
def xuqiu2():
res = rdd_map \
.map(lambda field_tuple: ((field_tuple[1], field_tuple[2]), 1)) \
.reduceByKey(lambda agg, curr: agg + curr) \
.top(10, lambda res_tup: res_tup[1])
print(res)
if __name__ == '__main__':
print("Spark的Python模板")
# 1. 创建SparkContext核心对象
conf = SparkConf().setAppName('sougou').setMaster('local[*]')
sc = SparkContext(conf=conf)
# 2. 读取外部文件数据
rdd = sc.textFile(name='file:///export/data/workspace/ky06_pyspark/_02_SparkCore/data/SogouQ.sample')
# 3. 执行相关的操作:
# 3.1 执行清洗操作
rdd_filter = rdd.filter(lambda line: line.strip() != '' and len(line.split()) == 6)
rdd_map = rdd_filter.map(lambda line: (
line.split()[0],
line.split()[1],
line.split()[2][1:-1],
line.split()[3],
line.split()[4],
line.split()[5]
))
# 由于 rdd_map 被多方使用了, 此时可以将其设置为缓存
rdd_map.persist(storageLevel=StorageLevel.MEMORY_AND_DISK).count()
# 3.2 : 实现需求
# 需求一: 统计每个关键词出现了多少次, 获取前10个
# 快速抽取函数: ctrl + alt + M
xuqiu1()
# 当需求1执行完成, 让缓存失效
rdd_map.unpersist().count()
# 需求二:统计每个用户每个搜索词点击的次数
xuqiu2()
time.sleep(100)
三、RDD的checkpoint检查点
checkpoint比较类似于缓存操作, 只不过缓存是将数据保存到内存 或者 磁盘上, 而checkpoint是将数据保存到磁盘或者HDFS(主要)上
checkpoint提供了更加安全可靠的持久化的方案, 确保RDD的数据不会发生丢失, 一旦构建checkpoint操作后, 会将RDD之间的依赖关系(血缘关系)进行截断,后续计算出来了问题, 可以直接从检查点的位置恢复数据
主要作用: 容错 也可以在一定程度上提升效率(性能) (不如缓存)
在后续计算失败后, 从检查点直接恢复数据, 不需要重新计算
相关的API:
第一步: 设置检查点保存数据位置
sc.setCheckpointDir(‘路径地址’)
第二步: 在对应RDD开启检查点
rdd.checkpoint()
rdd.count()
注意:
如果运行在集群模式中, checkpoint的保存的路径地址必须是HDFS, 如果是local模式 可以支持在本地路径
checkpoint数据不会自动删除, 必须同时手动方式将其删除掉
import time
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("演示checkpoint相关的操作")
# 1- 创建SparkContext对象
conf = SparkConf().setAppName('sougou').setMaster('local[*]')
sc = SparkContext(conf=conf)
# 开启检查点, 设置检查点的路径
sc.setCheckpointDir('/spark/chk') # 默认的地址为HDFS
# 2- 获取数据集
rdd = sc.parallelize(['张三 李四 王五 赵六', '田七 周八 李九 老张 老王 老李'])
# 3- 执行相关的操作: 以下操作仅仅是为了让依赖链条更长, 并没有太多的实际意义
rdd1 = rdd.flatMap(lambda line: line.split())
rdd2 = rdd1.map(lambda name: (name, 1))
rdd3 = rdd2.map(lambda name_tuple: (f'{name_tuple[0]}_itcast', name_tuple[1]))
rdd3 = rdd3.repartition(3)
rdd4 = rdd3.map(lambda name_tuple: name_tuple[0])
# RDD4设置检查点:
rdd4.checkpoint()
rdd4.count()
rdd5 = rdd4.flatMap(lambda name: name.split('_'))
rdd5 = rdd5.repartition(4)
rdd6 = rdd5.map(lambda name: (name, 1))
rdd_res = rdd6.reduceByKey(lambda agg, curr: agg + curr)
print(rdd_res.collect())
time.sleep(1000)
四、缓存和检查点区别
1- 存储位置不同:
缓存: 存储在内存或者磁盘 或者 堆外内存中
检查点: 可以将数据存储在磁盘 或者 HDFS上, 在集群模式下, 仅能保存到HDFS上
2- 血缘关系:
缓存: 不会截断RDD之间的血缘关系, 因为缓存数据有可能会失效, 当失效后, 需要重新回溯计算操作
检查点: 会截断RDD的之间的血缘关系, 因为检查点将数据保存到更加安全可靠的位置, 认为数据不会发生丢失问题, 当执行失败的时候, 也不需要重新回溯计算
3- 生命周期:
缓存: 当程序执行完成后, 或者手动调度unpersist 缓存都会被删除
检查点: 即使程序退出后, 检查点的数据依然是存在的, 不会删除, 需要手动删除的
一般建议将两种持久化的方案一同作用于项目环境中, 先设置缓存 然后再设置检查点, 最后统一触发执行(底层: 会将数据先缓存好, 然后将缓存好的数据, 保存到checkpoint对应的路径中, 后续在使用的时候, 优先从缓存中读取, 如果缓存中没有, 会从checkpoint中获取, 同时再把读取数据放置到缓存中)
总结
今天和大家分享了RDD的两种持久化方法。
