当前位置： 首页 > article >正文

Spark 的容错机制：保障数据处理的稳定性与高效性

article 2024/11/14 22:42:46

Spark 的介绍与搭建：从理论到实践_spark环境搭建-CSDN博客

Spark 的Standalone集群环境安装与测试-CSDN博客

PySpark 本地开发环境搭建与实践-CSDN博客

Spark 程序开发与提交：本地与集群模式全解析-CSDN博客

Spark on YARN：Spark集群模式之Yarn模式的原理、搭建与实践-CSDN博客

Spark 中 RDD 的诞生：原理、操作与分区规则-CSDN博客

Spark 中的 RDD 分区的设定规则与高阶函数、Lambda 表达式详解-CSDN博客

RDD 算子全面解析：从基础到进阶与面试要点-CSDN博客

PySpark 数据处理实战：从基础操作到案例分析-CSDN博客

目录

一、Spark 容错机制概述

1、各个软件为了防止数据丢失的解决方案

2、Spark如何保障数据的安全

二、RDD 持久化机制

（一）cache 算子

（二）persist 算子

（三）unpersist 算子

（四）示例代码分析

三、RDD 检查点机制

（一）功能与原理

（二）适用场景

四、RDD 的 cache、persist 持久化机制和 checkpoint 检查点机制的区别

五、将日志分析案例进行优化

六、总结

        在大数据处理领域，Spark 作为一款强大的分布式计算框架，面临着数据丢失和性能优化的双重挑战。为了确保数据的安全性和处理效率，Spark 构建了一套完善的容错机制。本文将深入探讨 Spark 的容错机制，包括 RDD 的持久化机制（persist 和 cache 算子）以及检查点机制（checkpoint），并分析它们的特点、适用场景以及相互之间的区别。

一、Spark 容错机制概述

1、各个软件为了防止数据丢失的解决方案

操作日志：

将内存变化操作日志追加记录在一个文件中，下一次读取文件对内存重新操作
  - NAMENODE：元数据的操作日志记录在edits
  - MySQL：日志记录binlog ()


 副本机制：

将数据构建多份冗余副本
  - HDFS：构建每个数据块的3个副本


依赖关系：

每份数据保留与其他数据之间的一个转换关系
  - RDD：保留RDD与其他RDD之间的依赖关系

2、Spark如何保障数据的安全

        每个RDD在构建数据时，会根据自己来源一步步导到数据来源，然后再一步步开始构建RDD数据。

问题：如果一个RDD被触发多次，这个RDD就会按照依赖关系被构建多次，性能相对较差，怎么解决？

例如：日志分析的时候，三个问题，tupleRdd 之前的所有操作都要执行三次，每次读取100M多的数据，效率非常的低

  • 第一次：一定会通过血脉构建这个RDD的数据
  • 希望从第二次开始，就不要重复构建，直接使用第一个构建的内容
  • 实现：Spark持久化机制：主动将RDD进行保存，供多次使用，避免重复构建

二、RDD 持久化机制

（一）cache 算子

  • 功能：将 RDD 缓存在内存中，以便后续多次使用时无需重新计算。
  • 语法：cache()。
  • 本质：底层实际调用的是 persist(StorageLevel.MEMORY_ONLY)，即只尝试将 RDD 缓存在内存。但如果内存资源不足，缓存操作可能会失败。
  • 场景：适用于资源充足且确定 RDD 只需在内存中缓存的情况，例如对于一些频繁使用且数据量较小能够完全容纳在内存中的 RDD，可以使用 cache 算子提高数据读取速度。

（二）persist 算子

  • 功能：能够将 RDD（包含其依赖关系）进行缓存，并且可以根据需求自行指定缓存的级别，这是它与 cache 算子的主要区别。
  • 语法：persist(StorageLevel)。
  • 级别：
    • 将 RDD 缓存在磁盘中：
      • StorageLevel.DISK_ONLY = StorageLevel(True, False, False, False)：将庞大且暂时不急需使用的 RDD 放入磁盘，释放 Executor 内存。
      • StorageLevel.DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)：在磁盘中多存储一个缓存副本，提高数据的冗余性和可用性。
      • StorageLevel.DISK_ONLY_3 = StorageLevel(True, False, False, False, 3)：类似地，存储三个副本。 
        StorageLevel.DISK_ONLY = StorageLevel(True, False, False, False)
StorageLevel.DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)
StorageLevel.DISK_ONLY_3 = StorageLevel(True, False, False, False, 3)

    • 将 RDD 缓存在内存中：
      • StorageLevel.MEMORY_ONLY = StorageLevel(False, True, False, False)：仅使用内存进行缓存，常用于高频使用且数据量不大能适应内存容量的 RDD。
      • StorageLevel.MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)：增加一个内存缓存副本。 
        StorageLevel.MEMORY_ONLY = StorageLevel(False, True, False, False)
StorageLevel.MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)

    • 将 RDD 优先缓存在内存中，如果内存不足，就缓存在磁盘中：
      • StorageLevel.MEMORY_AND_DISK = StorageLevel(True, True, False, False)：对于高频使用的大 RDD 较为合适，先利用内存缓存，当内存空间不足时，自动将多余数据溢出到磁盘。
      • StorageLevel.MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)：多一个缓存副本。 
        StorageLevel.MEMORY_AND_DISK = StorageLevel(True, True, False, False)
StorageLevel.MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)

    • 使用堆外内存：StorageLevel.OFF_HEAP = StorageLevel(True, True, True, False, 1)：提供了除内存和磁盘之外的存储选择，可利用堆外内存资源。 
      StorageLevel.OFF_HEAP = StorageLevel(True, True, True, False, 1)

    • 使用序列化：StorageLevel.MEMORY_AND_DISK_DESER = StorageLevel(True, True, False, True)：在内存和磁盘缓存时采用序列化方式，有助于减少内存占用，但在读取时需要进行反序列化操作，会有一定性能开销。 
      StorageLevel.MEMORY_AND_DISK_DESER = StorageLevel(True, True, False, True)

  • 场景：根据实际的资源状况，灵活地将 RDD 缓存在不同的存储介质中或者设置多个缓存副本，以平衡内存使用、数据读取速度和数据安全性。例如在内存和磁盘资源都较为充裕但内存使用较为紧张的情况下，可以选择 MEMORY_AND_DISK 级别进行缓存，既能保证数据的快速读取，又能在内存不足时利用磁盘空间。

  • 总结：Spark的StorageLevel共有9个缓存级别

DISK_ONLY：缓存入硬盘。这个级别主要是讲那些庞大的Rdd，之后仍需使用但暂时不用的，放进磁盘，腾出Executor内存。
DISK_ONLY_2：多一个缓存副本。
MEMORY_ONLY：只使用内存进行缓存。这个级别最为常用，对于马上用到的高频rdd，推荐使用。
MEMORY_ONLY_2：多一个缓存副本。
MEMORY_AND_DISK：先使用内存，多出来的溢出到磁盘，对于高频的大rdd可以使用。
MEMORY_AND_DISK_2：多一个缓存副本。
OFF_HEAP：除了内存、磁盘，还可以存储在OFF_HEAP

常用的：

项目中经常使用

MEMORY_AND_DISK_2

MEMORY_AND_DISK_DESER

（三）unpersist 算子

  • 功能：释放已缓存的 RDD，回收缓存占用的资源。
  • 语法：unpersist，还可以使用 unpersist(blocking=True)，表示等待 RDD 释放完资源后再继续执行下一步操作。
  • 场景：当确定某个 RDD 不再被使用，且后续还有大量代码需要执行时，及时调用 unpersist 算子将其数据从缓存中释放，避免资源的浪费。需要注意的是，如果不手动释放缓存，在 Spark 程序结束时，系统也会自动清理该程序中的所有缓存内存。

（四）示例代码分析

以下是一个简单的 Spark 程序示例，展示了如何使用 cache 、 persist 和 unpersist 算子：

import os
import time

# 导入pyspark模块
from pyspark import SparkContext, SparkConf
from pyspark.storagelevel import StorageLevel

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("spark的持久化机制")
    # 假如我想设置压缩
    # conf.set("spark.eventLog.compression.codec","snappy")
    # 根据配置文件，得到一个SC对象，第一个conf 是 形参的名字，第二个conf 是实参的名字
    sc = SparkContext(conf=conf)
    print(sc)

    fileRdd = sc.textFile("../resources/1.dat")
    # cache 是转换算子
    #cacheRdd = fileRdd.cache()
    cacheRdd = fileRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
    print(type(cacheRdd))

    cacheRdd.foreach(lambda x: print(x))

    time.sleep(20)

    cacheRdd.unpersist(blocking=True)

    time.sleep(10)

    # 使用完后，记得关闭
    sc.stop()

# unpersist(blocking=True)：等RDD释放完再继续下一步
# blocking = True：阻塞

        在上述代码中，首先配置了 Spark 运行所需的环境变量，然后创建了 SparkConf 和 SparkContext 对象。通过 textFile 方法读取文本文件创建了 fileRdd，接着使用 persist 算子将其缓存到内存和磁盘，并设置了两个副本。之后对缓存的 cacheRdd 进行了遍历操作，模拟了对 RDD 的使用。在暂停 20 秒后，调用 unpersist 算子释放缓存，最后关闭 SparkContext。

三、RDD 检查点机制

（一）功能与原理

  • 功能：将 RDD 的数据（不包含 RDD 依赖关系）存储在可靠的存储系统（如 HDFS）中。可以将其类比为虚拟机中的快照，作为数据处理过程中的一个重要里程碑。
  • 设置与使用：
    • 首先需要设置一个检查点目录，例如：sc.setCheckpointDir("../datas/chk/chk1")。
    • 然后对需要设置检查点的 RDD 调用 checkpoint() 方法，如：rs_rdd.checkpoint()。需要注意的是，一定要在触发该 RDD 的算子之前调用 checkpoint() 方法，否则检查点中可能没有数据。
  • 注意事项：启用检查点机制后，在代码执行过程中会专门多一个 job，用于将 RDD 数据持久化存储到 HDFS 中。​

（二）适用场景

        适用于对 RDD 数据安全性要求极高，但对性能要求相对不那么苛刻的场景。例如在一些数据处理任务中，数据的准确性和完整性至关重要，不容许因为任何故障导致数据丢失或错误，即使这可能会带来一定的性能开销，如在金融数据处理、关键业务数据分析等领域。

四、RDD 的 cache、persist 持久化机制和 checkpoint 检查点机制的区别

​

  1. 存储位置：
    • persist：可以将 RDD 缓存在内存或者磁盘中，根据指定的缓存级别灵活选择存储介质。
    • checkpoint：将 RDD 的数据存储在文件系统磁盘（通常是 HDFS）中，提供更可靠的持久化存储。
  2. 生命周期：
    • persist：当代码中遇到了 unpersist 算子调用或者整个 Spark 程序结束时，缓存会被自动清理，释放资源。
    • checkpoint：检查点的数据不会被自动清理，需要手动删除，这使得数据在长时间内都能保持可用状态，有利于数据的长期保存和回溯。
  3. 存储内容：
    • persist：会保留 RDD 的血脉关系，即与其他 RDD 的依赖关系。这样在缓存丢失时，可以依据依赖关系重新构建 RDD，恢复数据。
    • checkpoint：会斩断 RDD 的血脉关系，仅存储 RDD 的数据本身。这意味着一旦检查点数据可用，就不再依赖之前的 RDD 依赖链，简化了数据恢复过程，但也失去了基于依赖关系的灵活重建能力。

五、将日志分析案例进行优化

 对前文的一个案例进行优化

import os
import re

# 导入pyspark模块
from pyspark import SparkContext, SparkConf
import jieba
from pyspark.storagelevel import StorageLevel

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/sogou.tsv")
	print(fileRdd.count())
	print(fileRdd.first())
	listRdd = fileRdd.map(lambda line: re.split("\\s+", line))
	filterList = listRdd.filter(lambda l1: len(l1) == 6)
	# 这个结果只获取而来时间 uid 以及热词，热词将左右两边的[] 去掉了
	tupleRdd = filterList.map(lambda l1: (l1[0], l1[1], l1[2][1:-1]))
	# 将tupleRdd 缓存到内存中
	tupleRdd.cache()
	#tupleRdd.persist(storageLevel=StorageLevel.MEMORY_AND_DISK)

	# 求热词
	wordRdd = tupleRdd.flatMap(lambda t1: jieba.cut_for_search(t1[2]))
	filterRdd2 = wordRdd.filter(lambda word: len(word.strip()) != 0 and word != "的").filter(
		lambda word: re.fullmatch("[\u4e00-\u9fa5]+", word) is not None)
	# filterRdd2.foreach(print)
	result = filterRdd2.map(lambda word: (word, 1)).reduceByKey(lambda sum, num: sum + num).sortBy(
		keyfunc=lambda tup: tup[1], ascending=False).take(10)

	for ele in result:
		print(ele)


	# 第二问：  (（uid,"功夫")  10)
	# [(time,uid,"中华人民"),()]
	def splitWord(tupl):
		li1 = jieba.cut_for_search(tupl[2]) # 中国 中华 共和国
		li2 = list()
		for word in li1:
			li2.append(((tupl[1], word),1))
		return li2
	newRdd = tupleRdd.flatMap(splitWord)
	#newRdd.foreach(print)
	reduceByUIDAndWordRdd = newRdd.reduceByKey(lambda sum,num : sum + num)
	# reduceByUIDAndWordRdd.foreach(print)

	valList =reduceByUIDAndWordRdd.values()
	
	print(valList.max())
	print(valList.min())
	print(valList.mean()) # 中位数
	print(valList.sum() / valList.count()) #

	# 第三问 统计一天每小时点击量并按照点击量降序排序
	reductByKeyRDD = tupleRdd.map(lambda tup: (tup[0][0:2],1)).reduceByKey(lambda sum,num : sum + num)
	sortRdd = reductByKeyRDD.sortBy(keyfunc=lambda tup:tup[1],ascending=False)
	listNum = sortRdd.take(24)
	for ele in listNum:
		print(ele)

	tupleRdd.unpersist(blocking=True)
	# 使用完后，记得关闭
	sc.stop()

 将任务运行，运行过程中，发现内存中存储了50M的缓存数据

​

 适用场景：RDD需要多次使用，或者RDD是经过非常复杂的转换过程所构建。

一般缓存的RDD都是经过过滤，经过转换之后重复利用的rdd,可以添加缓存，否则不要加。

六、总结

        Spark 的容错机制通过多种方式保障了数据处理的稳定性和高效性。RDD 的持久化机制（包括 cache 和 persist 算子）为频繁使用的 RDD 提供了灵活的缓存策略，能够有效减少重复计算，提高处理效率。而检查点机制则侧重于数据的高安全性存储，在面对可能的缓存丢失或系统故障时，确保数据的完整性和可用性。在实际应用中，需要根据数据处理任务的特点、资源状况以及对数据安全性和性能的要求，合理选择使用持久化机制和检查点机制，以充分发挥 Spark 框架的优势，构建可靠高效的大数据处理应用。


http://www.kler.cn/a/390340.html

相关文章：

  • 某app最新版 vmp算法分析一
  • 如何在算家云搭建Peach-9B-8k-Roleplay（文本生成）
  • 轻松上手：使用Docker部署Java服务
  • rockylinux 8安装 gcc11.2
  • 人脸识别技术：从算法到深度学习的全面解析
  • 【知识科普】SPA单页应用程序介绍
  • 「IDE」集成开发环境专栏目录大纲
  • 【c++篇】：掌握list--实用的基本使用方法与模拟实现思路
  • 练习LabVIEW第四十二题
  • [CKS] 关闭API凭据自动挂载
  • 深入浅出rust内存对齐
  • 【机器学习】机器学习中用到的高等数学知识-1.线性代数 (Linear Algebra)
  • 初始JavaEE篇 —— 文件操作与IO
  • Tomcat漏洞利用工具-TomcatScanPro(二)
  • vue3 基于element-plus进行的一个可拖动改变导航与内容区域大小的简单方法
  • web——sqliabs靶场——第三关
  • Redis的使用场景及简单优化
  • cherno引擎课 -
  • 前端开发中ES6的技术细节二
  • 24/11/11 算法笔记 泊松融合
  • 开源项目低代码表单设计器FcDesigner扩展组件分组
  • 基于汇编语言实现的彩色黑白棋游戏
  • gitlab项目如何修改主分支main为master，以及可能遇到的问题
  • Electron 项目中获取 Windows 进程列表的深入剖析
  • Microsoft 365 Exchange如何设置可信发件IP白名单
  • MFC中 error C2440错误分析及解决方法