扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
我们上篇文章中讲了,RDD的持久化是spark优化中必须掌握的,并且,在内存不足的情况下,我们可以将持久化类型选择为MEMORY_ONLY_SER,减少内存的占用,持久化更多的partition,并且不同的序列化方法也会影响序列化性能。
下面,我们就来测试下,持久化级别和序列化方法的选择对RDD持久化大小的影响。
我选择了一个170.9MB的日志文件,传到了百度网盘
提取码:ffae
测试环境是windows,
IDEA参数配置
成都创新互联公司是一家专业提供芒市企业网站建设,专注与成都网站制作、成都网站设计、外贸营销网站建设、H5响应式网站、小程序制作等业务。10年已为芒市众多企业、政府机构等服务。创新互联专业的建站公司优惠进行中。
代码为
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存到内存
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)
}
}
代码逻辑就是输入是什么内容,输就是什么内容,在中间我将输入的文本RDD进行了memory_only持久化,我们就看这个持久化内存占多少
显然,input size大小是170.9 MB,但是持久化之后是908.5 MB,显然占据内存空间增大了好几倍,如果在生产上,内存资源不足的情况下,这种方式显然缓存不了不少partition
时间耗费14s
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,所以序列化对于节约内存空间是很有帮助的。
时间耗费11s
val conf = new SparkConf().setMaster("local[2]").setAppName("KyroTest")
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
与上一代码相比,为SparkConf设置了开启kryo序列化,不是默认的java序列化了,但是没有进行具体的类注册!
显然,input size大小是170.9 MB,但是持久化之后是有230.8MB,使用未注册的kryo序列化竟然比使用java序列化还臃肿!原因是:每一个对象实例的序列化结果都会包含一份完整的类名,造成了大量的空间浪费!
时间是9s,比java序列化快了一些。
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注册
显然,input size大小是170.9 MB,使用注册的kryo序列化之后,只有175.7MB,时间也才9秒,很舒服!
所以在目前为止,使用kryo序列化并注册是性能最好得了!!!
如果CPU还是那么悠闲的话,我们还有另外一个进一步优化点!
注意: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!!!
这个参数决定了RDD Cache的过程中,RDD数据在序列化之后是否进一步进行压缩再储存到内存或磁盘上。当然是为了进一步减小Cache数据的尺寸,对于Cache在磁盘上而言,绝对大小大概没有太大关系,主要是考虑Disk的IO带宽。而对于Cache在内存中,那主要就是考虑尺寸的影响,是否能够Cache更多的数据,是否能减小Cache数据对GC造成的压力等。
这两者,前者通常不会是主要问题,尤其是在RDD Cache本身的目的就是追求速度,减少重算步骤,用IO换CPU的情况下。而后者,GC问题当然是需要考量的,数据量小,占用空间少,GC的问题大概会减轻,但是是否真的需要走到RDD Cache压缩这一步,或许用其它方式来解决可能更加有效。
所以这个值默认是关闭的,但是如果在磁盘IO的确成为问题或者GC问题真的没有其它更好的解决办法的时候,可以考虑启用RDD压缩。
类型 | 输入大小 | 持久化大小 | 时间 |
---|---|---|---|
MEMORY_ONLY | 170.9 MB | 908.5 MB | 14s |
MEMORY_ONLY_SER | 170.9 MB | 204.9MB | 11s |
kyro序列化未注册 | 170.9 MB | 230.8MB | 9s |
kyro序列化注册 | 170.9 MB | 175.7MB | 9s |
注册kryo序列化并开启RDD压缩 | 170.9 MB | 45.6MB | 9s |
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流