Spark 調(diào)優(yōu)之RDD持久化級(jí)別及kryo序列化性能測(cè)試

我們上篇文章中講了，RDD的持久化是spark優(yōu)化中必須掌握的，并且，在內(nèi)存不足的情況下，我們可以將持久化類型選擇為MEMORY_ONLY_SER，減少內(nèi)存的占用，持久化更多的partition，并且不同的序列化方法也會(huì)影響序列化性能。
下面，我們就來(lái)測(cè)試下，持久化級(jí)別和序列化方法的選擇對(duì)RDD持久化大小的影響。
我選擇了一個(gè)170.9MB的日志文件，傳到了百度網(wǎng)盤
提取碼：ffae
測(cè)試環(huán)境是windows,
IDEA參數(shù)配置

MEMORY_ONLY

代碼為

case class CleanedLog(cdn:String,region:String,level:String,date:String,ip:String, domain:String, url:String, traffic:String)

  object KyroTest {
    def main(args: Array[String]) {

      val inputPath=new Path(args(0))
      val outputPath=new Path(args(1))

      val fsConf=new Configuration()
      val fs= FileSystem.get(fsConf)

      if (fs.exists(outputPath)) {
        fs.delete(outputPath,true)
        val path=args(1).toString

        println(s"已刪除已存在的路徑$path")
      }

      val conf = new SparkConf().setMaster("local[2]").setAppName("KyroTest")
      //conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      //conf.set("spark.kryo.registrationRequired", "true")
      val sc = new SparkContext(conf)

      val logs = sc.textFile(args(0))
      //logs.filter(_.split("\t").length==8).take(10).foreach(println(_))
      val logsCache=logsCahe(logs)
      //序列化的方式將rdd存到內(nèi)存
      saveAtLocal(logsCache,args(1))
      Thread.sleep(100000)
    }
    def logsCahe(logs:RDD[String]): RDD[CleanedLog] ={
      logs.filter(_.split("\t").length==8).map(x=>{
        val fields=x.split("\t")

        CleanedLog(fields(0),fields(1),fields(2),fields(3),fields(4),fields(5),fields(6),fields(7))
      }).cache()
    }
    def saveAtLocal(logsCache: RDD[CleanedLog], outputPath: String) = {
      logsCache.map(x=>{

        x.cdn+"\t"+x.region+"\t"+x.level+"\t"+x.date+"\t"+x.ip+"\t"+x.domain+"\t"+x.url+"\t"+x.traffic

      }).repartition(1).saveAsTextFile(outputPath)
    }
  }

代碼邏輯就是輸入是什么內(nèi)容，輸就是什么內(nèi)容，在中間我將輸入的文本RDD進(jìn)行了memory_only持久化，我們就看這個(gè)持久化內(nèi)存占多少

顯然，input size大小是170.9 MB，但是持久化之后是908.5 MB，顯然占據(jù)內(nèi)存空間增大了好幾倍，如果在生產(chǎn)上，內(nèi)存資源不足的情況下，這種方式顯然緩存不了不少partition
時(shí)間耗費(fèi)14s

MEMORY_ONLY_SER 未使用kryo序列化

def logsCahe(logs:RDD[String]): RDD[CleanedLog] ={
      logs.filter(_.split("\t").length==8).map(x=>{
        val fields=x.split("\t")

        CleanedLog(fields(0),fields(1),fields(2),fields(3),fields(4),fields(5),fields(6),fields(7))
      }).persist(StorageLevel.MEMORY_ONLY_SER)

代碼僅更改了persist(StorageLevel.MEMORY_ONLY_SER)


顯然，input size大小是170.9 MB，但是持久化之后是有204.9MB，所以序列化對(duì)于節(jié)約內(nèi)存空間是很有幫助的。

時(shí)間耗費(fèi)11s

MEMORY_ONLY_SER 使用kryo序列化未注冊(cè)

 val conf = new SparkConf().setMaster("local[2]").setAppName("KyroTest")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

與上一代碼相比，為SparkConf設(shè)置了開啟kryo序列化，不是默認(rèn)的java序列化了，但是沒有進(jìn)行具體的類注冊(cè)！


顯然，input size大小是170.9 MB，但是持久化之后是有230.8MB，使用未注冊(cè)的kryo序列化竟然比使用java序列化還臃腫！原因是:每一個(gè)對(duì)象實(shí)例的序列化結(jié)果都會(huì)包含一份完整的類名，造成了大量的空間浪費(fèi)！

時(shí)間是9s，比java序列化快了一些。

MEMORY_ONLY_SER 使用kryo序列化并注冊(cè)

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

添加了String類和自定義樣例類的kryo注冊(cè)


顯然，input size大小是170.9 MB，使用注冊(cè)的kryo序列化之后，只有175.7MB，時(shí)間也才9秒，很舒服！

所以在目前為止，使用kryo序列化并注冊(cè)是性能最好得了?。?！

如果CPU還是那么悠閑的話，我們還有另外一個(gè)進(jìn)一步優(yōu)化點(diǎn)！

注冊(cè)kryo序列化并開啟RDD壓縮

注意:RDD壓縮只能存在于序列化的情況下

val conf = new SparkConf().setMaster("local[2]").setAppName("KyroTest")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      conf.registerKryoClasses(Array(classOf[CleanedLog], classOf[String]))
      conf.set("spark.rdd.compress","true")

持久化的大小僅有45.6MB?。?！

spark.rdd.compress

這個(gè)參數(shù)決定了RDD Cache的過(guò)程中，RDD數(shù)據(jù)在序列化之后是否進(jìn)一步進(jìn)行壓縮再儲(chǔ)存到內(nèi)存或磁盤上。當(dāng)然是為了進(jìn)一步減小Cache數(shù)據(jù)的尺寸，對(duì)于Cache在磁盤上而言，絕對(duì)大小大概沒有太大關(guān)系，主要是考慮Disk的IO帶寬。而對(duì)于Cache在內(nèi)存中，那主要就是考慮尺寸的影響，是否能夠Cache更多的數(shù)據(jù)，是否能減小Cache數(shù)據(jù)對(duì)GC造成的壓力等。

這兩者，前者通常不會(huì)是主要問(wèn)題，尤其是在RDD Cache本身的目的就是追求速度，減少重算步驟，用IO換CPU的情況下。而后者，GC問(wèn)題當(dāng)然是需要考量的，數(shù)據(jù)量小，占用空間少，GC的問(wèn)題大概會(huì)減輕，但是是否真的需要走到RDD Cache壓縮這一步，或許用其它方式來(lái)解決可能更加有效。

所以這個(gè)值默認(rèn)是關(guān)閉的，但是如果在磁盤IO的確成為問(wèn)題或者GC問(wèn)題真的沒有其它更好的解決辦法的時(shí)候，可以考慮啟用RDD壓縮。

對(duì)比表格