重庆网站建设 夹夹虫,昆明网站建设公司,网易企业邮箱登录入口怎么登录,建设部执业考试中心网站RDD的缓存缓存: 一般当一个RDD的计算非常的耗时|昂贵(计算规则比较复杂),或者说这个RDD需要被重复(多方)使用,此时可以将这个RDD计算完的结果缓存起来, 便于后续的使用, 从而提升效率通过缓存也可以提升RDD的容错能力, 当后续计算失败后, 尽量不让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_DISKimport timeimport 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], ascendingFalse).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(confconf)# 2. 读取外部文件数据rdd sc.textFile(namefile:///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(storageLevelStorageLevel.MEMORY_AND_DISK).count()# 3.2 : 实现需求# 需求一: 统计每个关键词出现了多少次, 获取前10个# 快速抽取函数: ctrl alt Mxuqiu1()# 当需求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数据不会自动删除, 必须同时手动方式将其删除掉
