《深入理解Spark RDD缓存机制》(第4天)

news2024/10/23 2:50:52

文章目录

  • 前言
  • 一、小试牛刀:解剖RDD缓存机制?
    • 1. 什么是Spark RDD缓存策略
      • 1.1 为什幺RDD要做缓存
      • 1.2 缓存相关API:
      • 1.3 缓存案例解析:
      • 1.4 图解缓存效果:
    • 2. 什么是checkpoint缓存
      • 2.1 为什么要做checkpoint缓存
      • 2.2 checkpoint相关API:
      • 2.3 checkpoint案例解析
    • 3. 缓存和checkpoint的区别
      • 3.1 案例解析
  • 二、打铁趁热:面试题思考
    • 1. cache缓存和checkpoint检查点的区别
    • 2. 既然持久化的方案有两种,那么在生产环境中应该使用哪种方案呢?
  • 总结


前言

Apache Spark是一个大规模数据处理框架,它提供了高效、快速和通用的数据处理能力。在Spark中,弹性分布式数据集(RDD, Resilient Distributed Dataset)是一个核心概念,而RDD的缓存机制则是确保Spark性能高效的关键因素之一。本文将通过’案例’,'图文’等解析方式深入探讨Spark RDD的缓存机制。


一、小试牛刀:解剖RDD缓存机制?

1. 什么是Spark RDD缓存策略

  • 在Spark中,RDD的缓存机制允许我们将计算的结果存储在内存中,从而避免在后续的计算中重复计算相同的RDD。这对于迭代计算、机器学习等场景尤为重要,可以显著提高计算效率。
    在这里插入图片描述

1.1 为什幺RDD要做缓存

  • 当RDD被重复使用,或者计算该RDD比较容易出错,而且需要消耗比较多的资源和时间的时候,我们就可以将该RDD缓存起来。
  • 主要作用: 提升Spark程序的计算效率
  • 注意事项: RDD的缓存可以存储在内存或者是磁盘上,甚至可以存储在Executor进程的堆外内存中。主要是放在内存中,因此缓存的数据是不太稳定可靠。
    由于是临时存储,可能会存在丢失,所以缓存操作,并不会将RDD之间的依赖关系给截断掉(丢失掉),因为当缓存失效后,可以全部重新计算且缓存的API都是Lazy惰性的,如果需要触发缓存操作,推荐调用“count算子” ,因为运行效率高

1.2 缓存相关API:

设置缓存的API: 
	rdd.cache(): 将RDD的数据缓存储内存中
	rdd.persist(缓存的级别/位置): 将RDD的数据存储在指定位置

手动清理缓存API:
	rdd.unpersist()
默认情况下,当整个Spark应用程序执行完成后,缓存数据会自动失效,会被自动删除


缓存的级别/位置:
    DISK_ONLY: 只存储在磁盘
    DISK_ONLY_2: 只存储在磁盘,并且有2个副本
    DISK_ONLY_3: 只存储在磁盘,并且有3个副本
    MEMORY_ONLY: 只存储在内存中
    MEMORY_ONLY_2: 只存储在内存中,并且有2个副本
    MEMORY_AND_DISK: 存储在内存和磁盘中,先放在内存,再放在磁盘
    MEMORY_AND_DISK_2: 存储在内存和磁盘中,先放在内存,再放在磁盘,并且有2个副本
    OFF_HEAP: Executor进程的堆外内存
    
工作中最常用的是: MEMORY_AND_DISK和MEMORY_AND_DISK_2

1.3 缓存案例解析:

# 导包
import os
import time

import jieba
from pyspark import SparkConf, SparkContext, StorageLevel

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'


def get_topN_keyword(etlRDD, n):
    r1 = etlRDD.flatMap(lambda line_list: list(jieba.cut(line_list[2]))) \
        .filter(lambda word: word not in ('.', '+', '的')) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda agg, curr: agg + curr) \
        .top(n, lambda t: t[1])
    print(r1)


def get_topN_search(etlRDD, n):
    r2 = etlRDD.map(lambda line_list: ((line_list[1], line_list[2]), 1)) \
        .reduceByKey(lambda agg, curr: agg + curr) \
        .top(n, lambda t: t[1])
    print(r2)


# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    conf = SparkConf().setAppName('pyspark_demo').setMaster('local[*]')
    sc = SparkContext(conf=conf)
    # 2.数据输入
    textRDD = sc.textFile('file:///export/data/spark_project/spark_core/data/SogouQ.sample')
    # 3.数据处理(切分,转换,分组聚合)
    etlRDD = textRDD.filter(lambda line: line.strip() != '').map(lambda line: line.split()).filter(
        lambda line_list: len(line_list) >= 6)
    # 去除搜索内容两端的 [ ]
    etlRDD = etlRDD.map(lambda line_list:
                        [
                            line_list[0],
                            line_list[1],
                            line_list[2][1:-1],
                            line_list[3],
                            line_list[4],
                            line_list[5]
                        ])
    # 不加缓存
    # etlRDD.count()
    # 7.TODO: cache添加缓存,注意: 只能把缓存添加内存!相对用的少
    # etlRDD.cache().count()
    # 8.TODO: persist添加缓存,注意: 可以修改缓存级别
    etlRDD.persist(storageLevel=StorageLevel.MEMORY_AND_DISK_2).count()

    # 4.数据输出
    # 需求一: 统计每个 关键词 出现了多少次, 最终展示top10数据 注意:'.', '+', '的'  都需要过滤
    # 伪SQL:select 关键词 ,count(*)  from  搜狗表 group by 关键词
    get_topN_keyword(etlRDD, 10)

    # 8.TODO: 如果不想用缓存,可以使用unpersist释放缓存,给哪个rdd加的,就给哪个释放
    etlRDD.unpersist()

    # 需求二: 统计每个用户 每个 搜索内容 点击的次数, 最终展示top5数据
    # 伪SQL:select 用户,搜索内容,count(*)  from  搜狗表 group by 用户,搜索内容
    get_topN_search(etlRDD, 5)
    # 6.为了方便查看页面,可以让程序多睡会儿
    time.sleep(500)
    # 5.关闭资源
    sc.stop()

1.4 图解缓存效果:

  • 无缓存的DAG流程图显示:
    在这里插入图片描述

  • 有缓存的DAG流程图显示:
    在这里插入图片描述

  • cache基于内存
    在这里插入图片描述

  • persist可以修改缓存级别: 同时基于内存和磁盘
    在这里插入图片描述

2. 什么是checkpoint缓存

Checkpoint缓存,或称Checkpoint机制,是Apache Spark中用于确保数据一致性和容错性的一种技术。在不同的系统中,其实现方式和用途略有不同,但核心思想是一致的:确保关键数据或中间计算结果被安全地存储,以便在系统崩溃或需要恢复时能够重新使用。

2.1 为什么要做checkpoint缓存

  • RDD缓存主要是将数据存储在内存中,是临时存储,不太稳定,它主要是用来提升程序运行效率的。RDD的checkpoint(检查点)主要是将数据存储在HDFS上,是持久化存储。而HDFS存储数据有3副本的机制,让数据更加安全可靠。

  • checkpoint认为使用磁盘或者HDFS存储数据之后,数据非常的安全可靠,因此checkpoint会将RDD间的依赖关系给删除/丢弃掉。因此如果checkpoint的数据真的出现了问题,是无法在从头开始计算。

  • checkpoint主要作用: 提高程序的容错性
    注意事项: checkpoint可以将数据存储在磁盘或者HDFS上,主要是将数据存储在HDFS上。

2.2 checkpoint相关API:

	sc.setCheckpointDir(存储路径): 设置checkpoint数据存放路径
	rdd.checkpoint(): 对指定RDD启用checkpoint
	rdd.count(): 触发checkpoint

2.3 checkpoint案例解析

# 导包
import os
import time

import jieba
from pyspark import SparkConf, SparkContext, StorageLevel

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'


def get_topN_keyword(etlRDD, n):
    r1 = etlRDD.flatMap(lambda line_list: list(jieba.cut(line_list[2]))) \
        .filter(lambda word: word not in ('.', '+', '的')) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda agg, curr: agg + curr) \
        .top(n, lambda t: t[1])
    print(r1)


def get_topN_search(etlRDD, n):
    r2 = etlRDD.map(lambda line_list: ((line_list[1], line_list[2]), 1)) \
        .reduceByKey(lambda agg, curr: agg + curr) \
        .top(n, lambda t: t[1])
    print(r2)


# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    conf = SparkConf().setAppName('pyspark_demo').setMaster('local[*]')
    sc = SparkContext(conf=conf)
    # 2.数据输入
    textRDD = sc.textFile('file:///export/data/spark_project/spark_core/data/SogouQ.sample')
    # 3.数据处理(切分,转换,分组聚合)
    etlRDD = textRDD.filter(lambda line: line.strip() != '').map(lambda line: line.split()).filter(
        lambda line_list: len(line_list) >= 6)
    # 去除搜索内容两端的 [ ]
    etlRDD = etlRDD.map(lambda line_list:
                        [
                            line_list[0],
                            line_list[1],
                            line_list[2][1:-1],
                            line_list[3],
                            line_list[4],
                            line_list[5]
                        ])
    # 不加缓存
    # etlRDD.count()
    # 7.TODO: 先拿着sc对象设置检查点保存位置, 建议用hdfs,这样能利用hdfs的高可靠高可用性
    sc.setCheckpointDir('hdfs://node1:8020/ckpt')
    # 8.TODO: 添加检查点checkpoint
    etlRDD.checkpoint()
    etlRDD.count()

    # 4.数据输出
    # 需求一: 统计每个 关键词 出现了多少次, 最终展示top10数据 注意:'.', '+', '的'  都需要过滤
    # 伪SQL:select 关键词 ,count(*)  from  搜狗表 group by 关键词
    get_topN_keyword(etlRDD, 10)

    # 需求二: 统计每个用户 每个 搜索内容 点击的次数, 最终展示top5数据
    # 伪SQL:select 用户,搜索内容,count(*)  from  搜狗表 group by 用户,搜索内容
    get_topN_search(etlRDD, 5)
    # 6.为了方便查看页面,可以让程序多睡会儿
    time.sleep(500)
    # 5.关闭资源
    sc.stop()
  • 没有设置检查点正常的DAG执行流图:
    在这里插入图片描述
  • 设置检查点后:
    在这里插入图片描述

3. 缓存和checkpoint的区别

  • 面试题:Spark提供了两种持久化方案。一种为缓存操作,一种为checkpoint方案。请问有什么区别呢?
1- 数据存储位置不同
	缓存: 存储在内存或者磁盘 或者 堆外内存中
	checkpoint检查点: 可以将数据存储在磁盘或者HDFS上, 在集群模式下, 仅能保存到HDFS上

2- 数据生命周期:
	缓存: 当程序执行完成后, 或者手动调用unpersist 缓存都会被删除
	checkpoint检查点: 即使程序退出后, checkpoint检查点的数据依然是存在的, 不会删除, 需要手动删除

3- 血缘关系:
	缓存: 不会截断RDD之间的血缘关系, 因为缓存数据有可能是失效, 当失效后, 需要重新回溯计算操作
	checkpoint检查点: 会截断掉依赖关系, 因为checkpoint将数据保存到更加安全可靠的位置, 不会发生数据丢失的问题, 当执行失败的时候, 也不需要重新回溯执行
	
4- 主要作用不同:
	缓存: 提高Spark程序的运行效率和容错性
	checkpoint检查点: 提高Spark程序的容错性和高可用,高可靠性
  • 思考:既然持久化的方案有两种,那么在生产环境中应该使用哪种方案呢?
在同一个项目中,推荐缓存和checkpoint(检查点)同时配合使用。

使用顺序如下: 在代码中设置缓存和checkpoint检查点,然后再一同使用Action算子触发!!! 
使用count算子触发

实际过程如下: 程序会优先从缓存中读取数据,如果发现缓存中没有数据。
再从checkpoint中读取数据,并且接着将读取到的数据重新在内存中放置一份,
后续还是优先从缓存中读取

在这里插入图片描述

3.1 案例解析

# 导包
import os
import time

import jieba
from pyspark import SparkConf, SparkContext, StorageLevel

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'


def get_topN_keyword(etlRDD, n):
    r1 = etlRDD.flatMap(lambda line_list: list(jieba.cut(line_list[2]))) \
        .filter(lambda word: word not in ('.', '+', '的')) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda agg, curr: agg + curr) \
        .top(n, lambda t: t[1])
    print(r1)


def get_topN_search(etlRDD, n):
    r2 = etlRDD.map(lambda line_list: ((line_list[1], line_list[2]), 1)) \
        .reduceByKey(lambda agg, curr: agg + curr) \
        .top(n, lambda t: t[1])
    print(r2)


# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    conf = SparkConf().setAppName('pyspark_demo').setMaster('local[*]')
    sc = SparkContext(conf=conf)
    # 2.数据输入
    textRDD = sc.textFile('file:///export/data/spark_project/spark_core/data/SogouQ.sample')
    # 3.数据处理(切分,转换,分组聚合)
    etlRDD = textRDD.filter(lambda line: line.strip() != '').map(lambda line: line.split()).filter(
        lambda line_list: len(line_list) >= 6)
    # 去除搜索内容两端的 [ ]
    etlRDD = etlRDD.map(lambda line_list:
                        [
                            line_list[0],
                            line_list[1],
                            line_list[2][1:-1],
                            line_list[3],
                            line_list[4],
                            line_list[5]
                        ])
    # 不加缓存
    # etlRDD.count()
    # 7.TODO: persist添加缓存,注意: 可以修改缓存级别
    etlRDD.persist(storageLevel=StorageLevel.MEMORY_AND_DISK_2)
    # 8.TODO: 先拿着sc对象设置检查点保存位置, 建议用hdfs,这样能利用hdfs的高可靠高可用性
    sc.setCheckpointDir('hdfs://node1:8020/ckpt')
    etlRDD.checkpoint()
    etlRDD.count()
    # TODO:触发缓存和检查点
    etlRDD.count()

    # 4.数据输出
    # 需求一: 统计每个 关键词 出现了多少次, 最终展示top10数据 注意:'.', '+', '的'  都需要过滤
    # 伪SQL:select 关键词 ,count(*)  from  搜狗表 group by 关键词
    get_topN_keyword(etlRDD, 10)

    # 7.TODO: 如果不想用缓存,可以使用unpersist释放缓存,给哪个rdd加的,就给哪个释放
    # etlRDD.unpersist()

    # 需求二: 统计每个用户 每个 搜索内容 点击的次数, 最终展示top5数据
    # 伪SQL:select 用户,搜索内容,count(*)  from  搜狗表 group by 用户,搜索内容
    get_topN_search(etlRDD, 5)
    # 6.为了方便查看页面,可以让程序多睡会儿
    time.sleep(500)
    # 5.关闭资源
    sc.stop()
  • DAG有向无环图:
    在这里插入图片描述

二、打铁趁热:面试题思考

1. cache缓存和checkpoint检查点的区别

1- 数据存储位置不同
	缓存: 存储在内存或者磁盘 或者 堆外内存中
	checkpoint检查点: 可以将数据存储在磁盘或者HDFS上, 在集群模式下, 仅能保存到HDFS上

2- 数据生命周期:
	缓存: 当程序执行完成后, 或者手动调用unpersist 缓存都会被删除
	checkpoint检查点: 即使程序退出后, checkpoint检查点的数据依然是存在的, 不会删除, 需要手动删除

3- 血缘关系:
	缓存: 不会截断RDD之间的血缘关系, 因为缓存数据有可能是失效, 当失效后, 需要重新回溯计算操作
	checkpoint检查点: 会截断掉依赖关系, 因为checkpoint将数据保存到更加安全可靠的位置, 不会发生数据丢失的问题, 当执行失败的时候, 也不需要重新回溯执行
	
4- 主要作用不同:
	缓存: 提高Spark程序的运行效率和容错性
	checkpoint检查点: 提高Spark程序的容错性和高可用,高可靠性

2. 既然持久化的方案有两种,那么在生产环境中应该使用哪种方案呢?

在同一个项目中,推荐缓存和checkpoint(检查点)同时配合使用。

使用顺序如下: 在代码中设置缓存和checkpoint检查点,然后再一同使用Action算子触发!!! 
使用count算子触发

实际过程如下: 程序会优先从缓存中读取数据,如果发现缓存中没有数据。
再从checkpoint中读取数据,并且接着将读取到的数据重新在内存中放置一份,
后续还是优先从缓存中读取

总结

本文主要通过案例和图文的方式详解了Spark RDD 数据持久化的2种方案,重点思考项目中该采取什么方案。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1844472.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

车载测试面试项目看这一套就够了!车载测试___自我讲解项目

面试官您好,我叫xx来自安微,今年xx岁,毕业于安微新华学院,我是从2017年开始接触软件测试行业,目前从事软件测试工作有5年多时间,第一家公司做了电商和进销存项目app和web都有做过,上家公司做了车…

流程图工具评测:十大热门软件对比

流程图是一种用图形符号和箭头表示工作流程的图形表示方法。它展示了一系列相互关联的步骤,以显示过程中数据或物质的流动、决策点和操作步骤。流程图广泛用于各种领域,包括业务流程、软件开发、工程等,以帮助人们更好地理解和分析工作流程。…

大数据助力电商发展||电商API接口接入

伴随互联网尤其是移动互联网的高速发展,电子商务已经成为人们生活中不可或缺的一部分,人们的购物理念和消费模式正在发生颠覆性的转变。基于天然的数据优势,电子商务平台利用大数据计算技术不断实施数据的累积、分析和处理,消费者…

xshell使用vi命令:bash:vim:command not found

你们好,我是金金金。 场景 此时我通过xshell客户端连接到了远程的虚拟机。想用vi命令编辑一个文件时,显示:bash: vim: command not found 排查 看报错提示就可以知道,没找到vim命令 解决 使用包管理器 apt 来安装 vim 更新你的软…

Faiss:加速大规模数据相似性搜索的利器

在机器学习和数据挖掘领域,相似性搜索是一项基本且重要的任务,它涉及到在大型数据集中找到与特定对象最相似的对象。Faiss是一个由Facebook AI Research开发的库,专门用于高效地进行相似性搜索和聚类,它之所以重要,是因…

视频服务网关的特点

一、视频服务网关的介绍 视频服务网关采用Linux操作系统,可支持国内外不同品牌、不同协议、不同设备类型监控产品的统一接入管理,同时提供标准的H5播放接口供其他应用平台快速对接,让您快速拥有视频集成能力。不受开发环境、跨系统跨平台等条…

Linux系统:线程概念 线程控制

Linux系统:线程概念 & 线程控制 线程概念轻量级进程 LWP页表 线程控制POSIX 线程库 - ptherad线程创建pthread_createpthread_self 线程退出pthread_exitpthread_cancelpthread_joinpthread_detach 线程架构线程与地址空间线程与pthread动态库 线程的优缺点 线程…

【ajax核心01】ajax底层原理

一:XMLHttpRequest对象 节选自MDN网站 XMLHttpRequest(XHR)对象用于与服务器交互。通过 XMLHttpRequest 可以在不刷新页面的情况下请求特定 URL,获取数据。这允许网页在不影响用户操作的情况下,更新页面的局部内容。XMLHttpReque…

Excel入门必备:掌握单元格引用,轻松应对数据处理

文章目录 概述引用单个单元格:引用单元格范围:相对引用:绝对引用:混合引用: 概述 在 Excel 中,单元格引用是指引用工作表中的单个单元格或单元格范围。单元格引用通常用于在公式中使用单元格的值或进行数据…

腾讯云开端口

轻量服务器 由于开发者计划,这些腾讯云 阿里云什么的小vps,是非常之便宜,甚至到了白送的地步(小阿),但是作为一个web安全学习者必要的vps操作还是要会的 开启端口 腾讯云的轻量服务器是没有安全组的&…

SQLite 3 优化批量数据存储操作---事务transaction机制

0、事务操作 事务的目的是为了保证数据的一致性和完整性。 事务(Transaction)具有以下四个标准属性,通常根据首字母缩写为 ACID: 原子性(Atomicity):确保工作单位内的所有操作都成功完成&…

蓝牙模块在车载系统中的应用与集成:现状、挑战与未来展望

随着科技的快速发展,蓝牙技术已经深入到我们生活的方方面面,其中车载系统中的应用尤为显著。蓝牙模块作为一种无线通信技术,不仅为驾驶者提供了更加便捷的操作体验,同时也提升了驾驶的安全性。本文旨在分析蓝牙模块在车载系统中的…

计算机网络知识点汇总(二)

计算机网络知识点汇总 第1章计算机网络体系结构 1.1 计算机网络概述 1.1.1 计算机网络的概念 ​ 计算机网络是由若干个结点(node)和连接这些结点的链路(link)组成。网络中的结点可以是就三级、集线器、交换机、或者路由器等,网络之间通过路由器进行互联&#xf…

基于Java的家政服务管理平台

开头语:你好呀,我是计算机学姐码农小野!如果有相关需求,可以私信联系我。 开发语言:Java 数据库:MySQL 技术:B/S结构,SpringBoot框架 工具:MyEclipse,Nav…

在编译内核时添加驱动的固件

最近调驱动时,无法正常加载引导。 使用的内核5.10 内核启动先于文件系统,内核启动时驱动无法访问固件文件,所以无法加载驱动。 有2个办法,可以解决,一是驱动编译KO模块,系统启动后,再动态加载…

小林图解系统-二.硬件结构 2.6什么是软中断?

中断是什么? 中断是系统用来响应硬件设备请求的一种机制,操作系统收到硬件的中断请求,会打断正在执行的进程,然后调用内核中的中断处理程序来相应请求。 中断好比外卖打电话,打了电话可以停下手中的事情 中断是一种…

Qt项目学习-20240617

Qt项目学习 1.0 文件构建 1.1 预处理命令 C预处理命令是编译过程中的第一步,发生在编译器进行实际编译之前。预处理器(preprocessor)执行这些命令,它们不是C语言的一部分,但对源代码的编译过程至关重要。以下是一些常…

【javascript】vsCode我常用的开发插件汇总

console-log – 此插件用于快捷键生成console,当然你也可以在其他模板中自定义快捷生成内容 CodeGeeX – 一个内嵌在vscode插件列中的AI工具,让我们可以直接获取AI信息而不必麻烦的到其他AI软件或平台搜索,我经常用来查询TS定义类型相关的警告,效率坐火…

Window使用Hyper-V进行显卡直通

一、环境配置 处理器 Intel Xeon CPU E5-2680 v3 @ 2.50GHz 2.50 GHz 机带 RAM 64.0 GB 二、安装Hyper-V 控制面板–>程序和功能->启用或关闭winodws功能 三、打开Hyper-V安装windows 安装windows略过,记住(禁用检查点) 四、在本机Winows上以及管理员打开Powe…

【Java】线程池技术(二)ThreadPoolExecutor的基本定义

线程池初始化与定义 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)线程池构造方法的入参含义分别如下&…