首页 > 其他 > 详细

RDD缓存与RDD序列化缓存的大小对比

时间：2019-05-23 23:14:17      阅读：125      评论：0      收藏：0      [点我收藏+]

RDD的缓存一般就是用cache和persist，那二者之间有什么关系，又有什么区别呢？

关系：

persist()内部调用了persist(StorageLevel.MEMORY_ONLY)

cache()调用了persist()

区别：

persist有一个 StorageLevel 类型的参数，这个表示的是RDD的缓存级别。

cache只有一个默认的缓存级别MEMORY_ONLY ，而persist可以根据情况设置其它的缓存级别。

StorageLevel部分源码
object StorageLevel {
  val NONE = new StorageLevel(false, false, false, false)
  val DISK_ONLY = new StorageLevel(true, false, false, false)
  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
  val MEMORY_ONLY = new StorageLevel(false, true, false, true)
  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
  val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
          。。。。。。
}

我们介绍6种常用的

MEMORY_ONLY
 使用未序列化的Java对象格式，将数据保存在内存中。如果内存不够存放的所有数据，则数据可能就不会进行持久化。那么下次对这个RDD执行算子操作时，那些没有被持久化的数据，需要从源头上重新计算一遍。这是默认的持久化策略。使用cache()方法时，实际就是使用这种持久化策略  
MEMORY_ONLY_SER
  基本上含义同MEMORY_ONLY一样。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个patition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多的内存导致频繁的GC 
MEMORY_AND_DISK
  用未序列化的java对象格式，优先尝试将数据保存在内存中。如果内存不够存放数据，会将数据写入磁盘文件中，下次对这个RDD执行算子时，持久化在磁盘文件中的数据会被读取出来使用 
MEMORY_AND_DISK_SER
  基本含义同MEMORY_AND_DISK。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。
  
DISK_ONLY
 使用未序列化的Java对象格式，将数据全部写入磁盘文件中。
  
MEMORY_ONLY_2，MEMORY_AND_DISK_2
等		  

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

 

持久化级别的选择

  • MEMORY_ONLY是我们最理想的持久化选择，因为它不需要序列化与反序列化的操作，就避免了这块性能的开销，对于这个RDD的后续操作都是基于内存里的数据，不需要从磁盘中读取数据，这样就实现了数据本地化，性能非常高，也是我们最理想的状态，相比于带“_2”的持久化级别，MEMORY_ONLY也不需要传送一份备份数据到远程节点上。但是，这一切都要基于内存足够大的基础上，并且如果你的数据量很大，这种情况基本上就是不存在的，因为没有序列化，就会导致数据很大，基本上是原来的2-3倍，假设每天有10亿数据，那每天你要缓存相当于30亿的数据。这时，直接用这种持久化级别，会导致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的级别，必须将所有数据都复制一份副本，并发送到其他节点上，数据复制以及网络传输会导致较大的性能开销，除非是要求作业的高可用性，否则不建议使用。

持久化级别测试

 1.MEMORY_ONLY

 没有注册没有序列化

package com.ruozedata.spark1
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 使用kryo序列化数据对象,
  * 1、生产山spark.serializer这个配置直接写到spark-default配置文件中即可
  */
object KryoApp {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("KyroApp").setMaster("local[2]")
//      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
//    conf.registerKryoClasses(Array(classOf[Logger]))

    val sc = new SparkContext(conf)
    val logRDD = sc.textFile("C:\\Users\\小十七\\Desktop/CDN_click.log")
      .map(x => {
      val logFields = x.split(",")
      Logger(logFields(0), logFields(1), logFields(2), logFields(3),
        logFields(4), logFields(5), logFields(6), logFields(7))
    })
     val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY)
 //val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY_SER)

    println("总条数" + pesisitRDD.count())
    Thread.sleep(2000000)
    sc.stop()

  }

  case class Logger(filed1: String, filed2: String, filed3: String,
                    filed4: String, filed5: String, filed6: String,
                    filed7: String, filed8: String)

}

 

 

 技术分享图片

从图中可以看出读入的时候是93.6M

技术分享图片

缓存以后就是560.4M，是原来的6倍

2.MEMORY_ONLY_SER

package com.ruozedata.spark1
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 使用kryo序列化数据对象,
  * 1、生产山spark.serializer这个配置直接写到spark-default配置文件中即可
  */
object KryoApp {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("KyroApp").setMaster("local[2]")
//      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
//    conf.registerKryoClasses(Array(classOf[Logger]))

    val sc = new SparkContext(conf)
    val logRDD = sc.textFile("C:\\Users\\小十七\\Desktop/CDN_click.log")
      .map(x => {
      val logFields = x.split(",")
      Logger(logFields(0), logFields(1), logFields(2), logFields(3),
        logFields(4), logFields(5), logFields(6), logFields(7))
    })


    // val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY)
 val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY_SER)


    println("总条数" + pesisitRDD.count())


    Thread.sleep(2000000)
    sc.stop()

  }

  case class Logger(filed1: String, filed2: String, filed3: String,
                    filed4: String, filed5: String, filed6: String,
                    filed7: String, filed8: String)


}

 

 

技术分享图片

由上图可以看出序列化以后，缓存减小到了116M

3.使用Kryo但不注册不序列化

package com.ruozedata.spark1
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 使用kryo序列化数据对象,
  * 1、生产山spark.serializer这个配置直接写到spark-default配置文件中即可
  */
object KryoApp {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("KyroApp").setMaster("local[2]")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //conf.registerKryoClasses(Array(classOf[Logger]))

    val sc = new SparkContext(conf)
    val logRDD = sc.textFile("C:\\Users\\小十七\\Desktop/CDN_click.log")
      .map(x => {
      val logFields = x.split(",")
      Logger(logFields(0), logFields(1), logFields(2), logFields(3),
        logFields(4), logFields(5), logFields(6), logFields(7))
    })


   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY)
// val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY_SER)


    println("总条数" + pesisitRDD.count())


    Thread.sleep(2000000)
    sc.stop()

  }

  case class Logger(filed1: String, filed2: String, filed3: String,
                    filed4: String, filed5: String, filed6: String,
                    filed7: String, filed8: String)


}

 技术分享图片

4.使用Kryo但不注册序列化

package com.ruozedata.spark1
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 使用kryo序列化数据对象,
  * 1、生产山spark.serializer这个配置直接写到spark-default配置文件中即可
  */
object KryoApp {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("KyroApp").setMaster("local[2]")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //conf.registerKryoClasses(Array(classOf[Logger]))

    val sc = new SparkContext(conf)
    val logRDD = sc.textFile("C:\\Users\\小十七\\Desktop/CDN_click.log")
      .map(x => {
      val logFields = x.split(",")
      Logger(logFields(0), logFields(1), logFields(2), logFields(3),
        logFields(4), logFields(5), logFields(6), logFields(7))
    })


//   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY)
   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY_SER)


    println("总条数" + pesisitRDD.count())


    Thread.sleep(2000000)
    sc.stop()

  }

  case class Logger(filed1: String, filed2: String, filed3: String,
                    filed4: String, filed5: String, filed6: String,
                    filed7: String, filed8: String)


}

技术分享图片

5.使用Kryo但注册不序列化

package com.ruozedata.spark1
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 使用kryo序列化数据对象,
  * 1、生产山spark.serializer这个配置直接写到spark-default配置文件中即可
  */
object KryoApp {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("KyroApp").setMaster("local[2]")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.registerKryoClasses(Array(classOf[Logger]))

    val sc = new SparkContext(conf)
    val logRDD = sc.textFile("C:\\Users\\小十七\\Desktop/CDN_click.log")
      .map(x => {
      val logFields = x.split(",")
      Logger(logFields(0), logFields(1), logFields(2), logFields(3),
        logFields(4), logFields(5), logFields(6), logFields(7))
    })


   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY)
//   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY_SER)


    println("总条数" + pesisitRDD.count())


    Thread.sleep(2000000)
    sc.stop()

  }

  case class Logger(filed1: String, filed2: String, filed3: String,
                    filed4: String, filed5: String, filed6: String,
                    filed7: String, filed8: String)


}

技术分享图片

6.使用Kryo注册并序列化

package com.ruozedata.spark1
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 使用kryo序列化数据对象,
  * 1、生产山spark.serializer这个配置直接写到spark-default配置文件中即可
  */
object KryoApp {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("KyroApp").setMaster("local[2]")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.registerKryoClasses(Array(classOf[Logger]))

    val sc = new SparkContext(conf)
    val logRDD = sc.textFile("C:\\Users\\小十七\\Desktop/CDN_click.log")
      .map(x => {
      val logFields = x.split(",")
      Logger(logFields(0), logFields(1), logFields(2), logFields(3),
        logFields(4), logFields(5), logFields(6), logFields(7))
    })


//   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY)
   val pesisitRDD = logRDD.persist(StorageLevel.MEMORY_ONLY_SER)


    println("总条数" + pesisitRDD.count())


    Thread.sleep(2000000)
    sc.stop()

  }

  case class Logger(filed1: String, filed2: String, filed3: String,
                    filed4: String, filed5: String, filed6: String,
                    filed7: String, filed8: String)


}

总结：上边的6种情况分别对MEMORY_ONLY，MEMORY_ONLY_SER做了是否序列化和是否使用Kryo注册的测试，从Web界面的Size in Memory这个参数可以看出凡是经过序列化的缓存以后的值不会很大，但是不序列化，无论注册与否值都是非常大的，对于序列化的使用Kryo并注册以后的的值是最理想的，和原始文件的大小非常贴近，只是使用Kryo序列化，而不注册，缓存值稍微大一点113M左右

技术分享图片

 

RDD缓存与RDD序列化缓存的大小对比

原文：https://www.cnblogs.com/xuziyu/p/10914701.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
分享档案
更多>
2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)
最新文章
更多>
教程昨日排行
更多>
友情链接
汇智网   PHP教程   插件网  
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣，分享程序人生！