Spark RDD 缓存机制

这篇具有很好参考价值的文章主要介绍了Spark RDD 缓存机制。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Spark RDD 缓存机制

Spark RDD 缓存是在内存存储RDD计算结果的一种优化技术。把中间结果缓存起来以便在需要的时候重复使用,这样才能有效减轻计算压力,提升运算性能。

当对RDD执行持久化操作时,每个节点都会将自己操作的RDD的partition持久化到内存中,并且在之后对该RDD的反复使用中,直接使用内存缓存的partition。这样的话,对于针对一个RDD反复执行多个操作的场景,就只要对RDD计算一次即可,后面直接使用该RDD,而不需要反复计算多次该RDD。

巧妙使用RDD持久化,甚至在某些场景下,可以将spark应用程序的性能提升10倍。对于迭代式算法和快速交互式应用来说,RDD持久化,是非常重要的。

如何持久化

要持久化一个RDD,只要调用其cache()或者persist()方法即可。在该RDD第一次被计算出来时,就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的,如果持久化的RDD的任何partition丢失了,那么Spark会自动通过其源RDD,使用transformation操作重新计算该partition。

cache()persist()的区别在于,cache()persist()的一种简化方式,cache()的底层就是调用的persist()的无参版本,同时就是调用persist(MEMORY_ONLY),将数据持久化到内存中。如果需要从内存中去除缓存,那么可以使用unpersist()方法。

RDD持久化存储级别

RDD存储级别主要有以下几种。

级别 使用空间 CPU时间 是否在内存中 是否在磁盘上 备注
MEMORY_ONLY
MEMORY_ONLY_2 数据存2份
MEMORY_ONLY_SER 数据序列化
MEMORY_ONLY_SER_2 数据序列化,数据存2份
MEMORY_AND_DISK 中等 部分 部分 如果数据在内存中放不下,则溢写到磁盘
MEMORY_AND_DISK_2 中等 部分 部分 数据存2份
MEMORY_AND_DISK_SER 部分 部分
MEMORY_AND_DISK_SER_2 部分 部分 数据存2份
DISK_ONLY
DISK_ONLY_2 数据存2份
OFF_HEAP

注意:只能设置一种:不然会抛异常:
Cannot change storage level of an RDD after it was already assigned a level

异常源码如下:

 private def persist(newLevel: StorageLevel, allowOverride: Boolean): this.type = {    // TODO: Handle changes of StorageLevel    if (storageLevel != StorageLevel.NONE && newLevel != storageLevel && !allowOverride) {      throw new UnsupportedOperationException(        "Cannot change storage level of an RDD after it was already assigned a level")    }    // If this is the first time this RDD is marked for persisting, register it    // with the SparkContext for cleanups and accounting. Do this only once.    if (storageLevel == StorageLevel.NONE) {      sc.cleaner.foreach(_.registerRDDForCleanup(this))      sc.persistRDD(this)    }    storageLevel = newLevel    this  }

MEMORY_ONLY

使用未序列化的Java对象格式,将数据保存在内存中。如果内存不够存放所有的数据,则数据可能就不会进行持久化。那么下次对这个RDD执行算子操作时,那些没有被持久化的数据,需要从源头处重新计算一遍。这是默认的持久化策略,使用cache()方法时,实际就是使用的这种持久化策略。

MEMORY_ONLY_SER

基本含义同MEMORY_ONLY。唯一的区别是,会将RDD中的数据进行序列化,RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存,从而可以避免持久化的数据占用过多内存导致频繁GC。

MEMORY_AND_DISK

使用未序列化的Java对象格式,优先尝试将数据保存在内存中。如果内存不够存放所有的数据,会将数据写入磁盘文件中,下次对这个RDD执行算子时,持久化在磁盘文件中的数据会被读取出来使用。

MEMORY_AND_DISK_SER

基本含义同MEMORY_AND_DISK。唯一的区别是,会将RDD中的数据进行序列化,RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存,从而可以避免持久化的数据占用过多内存导致频繁GC。

DISK_ONLY

使用未序列化的Java对象格式,将数据全部写入磁盘文件中。

OFF_HEAP

这个目前是试验型选项,类似MEMORY_ONLY_SER,但是数据是存储在堆外内存的。

后缀带“_2”的存储级别

对于上述任意一种持久化策略,如果加上后缀_2,代表的是将每个持久化的数据,都复制一份副本,并将副本保存到其他节点上。这种基于副本的持久化机制主要用于进行容错。假如某个节点挂掉了,节点的内存或磁盘中的持久化数据丢失了,那么后续对RDD计算时还可以使用该数据在其他节点上的副本。如果没有副本的话,就只能将这些数据从源头处重新计算一遍了。

如何选择一种最合适的持久化策略

  • 默认情况下,性能最高的当然是MEMORY_ONLY,但前提是你的内存必须足够足够大,可以绰绰有余地存放下整个RDD的所有数据。因为不进行序列化与反序列化操作,就避免了这部分的性能开销;对这个RDD的后续算子操作,都是基于纯内存中的数据的操作,不需要从磁盘文件中读取数据,性能也很高;而且不需要复制一份数据副本,并远程传送到其他节点上。但是这里必须要注意的是,在实际的生产环境中,恐怕能够直接用这种策略的场景还是有限的,如果RDD中数据比较多时(比如几十亿),直接用这种持久化级别,会导致JVM的OOM内存溢出异常。
  • 如果使用MEMORY_ONLY级别时发生了内存溢出,那么建议尝试使用MEMORY_ONLY_SER级别。该级别会将RDD数据序列化后再保存在内存中,此时每个partition仅仅是一个字节数组而已,大大减少了对象数量,并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销,主要就是序列化与反序列化的开销。但是后续算子可以基于纯内存进行操作,因此性能总体还是比较高的。此外,可能发生的问题同上,如果RDD中的数据量过多的话,还是可能会导致OOM内存溢出的异常。
  • 如果纯内存的级别都无法使用,那么建议使用MEMORY_AND_DISK_SER策略,而不是MEMORY_AND_DISK策略。因为既然到了这一步,就说明RDD的数据量很大,内存无法完全放下。序列化后的数据比较少,可以节省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中,内存缓存不下才会写入磁盘。
  • 通常不建议使用DISK_ONLY和后缀为_2的级别:因为完全基于磁盘文件进行数据的读写,会导致性能急剧降低,有时还不如重新计算一次所有RDD。后缀为_2的级别,必须将所有数据都复制一份副本,并发送到其他节点上,数据复制以及网络传输会导致较大的性能开销,除非是要求作业的高可用性,否则不建议使用。

如何使用 Spark rdd 缓存

调用rdd.persist()

变量可以这样设置,如:rdd.persist(StorageLevel.MEMORY_ONLY);这里使用了MEMORY_ONLY级别存储。当然也可以选择其他的如:rdd.persist(StorageLevel.DISK_ONLY())

调用rdd.cache()

cache()rdd.persist(StorageLevel.MEMORY_ONLY)的简写,效果和他一模一样的。

调用rdd.unpersist()清除缓存

rdd.unpersist()把缓存起来的RDD清除,后续如果用到该RDD,则需要重新计算。文章来源地址https://www.toymoban.com/news/detail-843311.html

到了这里,关于Spark RDD 缓存机制的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

原文地址:https://blog.csdn.net/king14bhhb/article/details/136978508

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spark 【RDD编程(一)RDD编程基础】

            在Spark中,RDD是弹性分布式数据集(Resilient Distributed Dataset)的缩写。通俗来讲,RDD是一种抽象的数据结构,用于表示分布式计算中的数据集合。它是Spark中最基本的数据模型,可以看作是一个不可变的、可分区、可并行处理的数据集合。这个数据集的全部或部分可

    2024年02月09日
    浏览(12)
  • Spark【RDD编程(三)键值对RDD】

            键值对 RDD 就是每个RDD的元素都是 (key,value)类型的键值对,是一种常见的 RDD,可以应用于很多场景。                 因为毕竟通过我们之前Hadoop的学习中,我们就可以看到对数据的处理,基本都是以键值对的形式进行统一批处理的,因为MapReduce模型中

    2024年02月09日
    浏览(8)
  • SPARK-RDD

    分区列表 RDD 数据结构中存在分区列表,用于执行任务时并行计算,是实现分布式计算的重要属性。 分区计算函数 Spark 在计算时,是使用分区函数对每一个分区进行计算 RDD之间的依赖关系 RDD 是计算模型的封装,当需求中需要将多个计算模型进行组合时,就需要将多个 RDD 建

    2024年02月04日
    浏览(7)
  • Spark---RDD依赖关系

    1.1 RDD依赖关系 在Spark中,一个RDD的形成依赖于另一个RDD,则称这两个RDD具有依赖关系(一般指相邻的两个RDD之间的关系) ,RDD的依赖关系对于优化Spark应用程序的性能和可靠性非常重要。通过合理地设计RDD的转换和动作操作,可以避免不必要的Shuffle操作,提高计算效率。 words的

    2024年01月19日
    浏览(19)
  • 【Spark】RDD转换算子

    目录 map mapPartitions mapPartitionsWithIndex flatMap glom groupBy shuffle filter sample distinct coalesce repartition sortBy ByKey intersection union subtract zip partitionBy reduceByKey groupByKey reduceByKey 和 groupByKey 的区别 aggregateByKey foldByKey combineByKey reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别 join leftOuterJoin

    2024年02月12日
    浏览(12)
  • Spark核心--RDD介绍

    rdd  弹性分布式数据集  是spark框架自己封装的数据类型,用来管理内存数据 数据集: rdd数据的格式  类似Python中 []     。 hive中的 该结构[] 叫 数组 rdd提供算子(方法)  方便开发人员进行调用计算数据 在pysaprk中本质是定义一个rdd类型用来管理和计算内存数据 分布式 : r

    2024年01月16日
    浏览(7)
  • Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】

    Saprk 提供了两种方法来实现从 RDD 转换得到 DataFrame: 利用反射机制推断 RDD 模式 使用编程方式定义 RDD 模式 下面使用到的数据 people.txt :         在利用反射机制推断 RDD 模式的过程时,需要先定义一个 case 类,因为只有 case 类才能被 Spark 隐式地转换为DataFrame对象。 注意

    2024年02月09日
    浏览(7)
  • Spark大数据处理讲课笔记---Spark RDD典型案例

    利用RDD计算总分与平均分 利用RDD统计每日新增用户 利用RDD实现分组排行榜 针对成绩表,计算每个学生总分和平均分   读取成绩文件,生成lines;定义二元组成绩列表;遍历lines,填充二元组成绩列表;基于二元组成绩列表创建RDD;对rdd按键归约得到rdd1,计算总分;将rdd1映射

    2024年02月06日
    浏览(17)
  • 【Spark练习】RDD分区操作

    2.1 textFile 对于textFile而言,如果 没有在方法中指定分区数 ,则sc.defaultMinPartitions默认为 min(defaultParallelism,2) ,其中,defaultParallelism对应的就是spark.default.parallelism,如果是从HDFS中读取文件,则分区数为文件分片数(比如,128MB/片) rdd的分区数 = max(本地file的分片数, sc.default

    2024年02月02日
    浏览(7)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包