在Spark中使用Kryo序列化

spark序列化

对于优化<网络性能>极为重要,将RDD以序列化格式来保存减少内存占用.
spark.serializer=org.apache.spark.serializer.JavaSerialization

Spark默认 使用Java自带的ObjectOutputStream
框架来序列化对象,这样任何实现了 java.io.Serializable 接口的对象,都能被序列化。同时,还可以通过扩展
java.io.Externalizable 来控制序列化性能。Java序列化很灵活但性能差速度很慢,同时序列化后占用的字节数也较多。

spark.serializer=org.apache.spark.serializer.KryoSerialization

KryoSerialization速度快,可以配置为任何org.apache.spark.serializer的子类。但Kryo也不支持所有实现了
java.io.Serializable 接口的类型,它需要你在程序中 register 需要序列化的类型,以得到最佳性能。

LZO的支持要求先安装 Hadoop-lzo包(每个节点), 并放到
Spark本地库中。如果是Debian包安装,在调用spark-submit时加上 --driver-library-path
/usr/lib/hadoop/lib/native/ --driver-class-path /usr/lib/hadoop/lib/
就可以。 下载lzo http://cn.jarfire.org/hadoop.lzo.html

在 SparkConf 初始化的时候调用
conf.set(“spark.serializer”,
“org.apache.spark.serializer.KryoSerializer”) 使用
Kryo。这个设置不仅控制各个worker节点之间的混洗数据序列化格式,同时还控制RDD存到磁盘上的序列化格式。需要在使用时注册需要序列化的类型,建议在对网络敏感的应用场景下使用Kryo。
如果你的自定义类型需要使用Kryo序列化,可以用 registerKryoClasses 方法先注册:

val conf = new SparkConf.setMaster(...).setAppName(...)

conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

val sc = new SparkContext(conf)
最后,如果你不注册需要序列化的自定义类型,Kryo也能工作,不过每一个对象实例的序列化结果都会包含一份完整的类名,这有点浪费空间。

在Scala中使用New API (Twitter Elephant Bird 包) lzo JsonInputFormat读取 LZO 算法压缩的 JSON 文件:

val input = sc.newAPIHadoopFile(inputFile, classOf[lzoJsonInputFormat], classOf[LongWritable], classOf[MapWritable], conf)

inputFile: 输入路径

接收第一个类:“格式”类,输入格式

接收第二个类:“键”

接收第二个类:“值”

conf:设置一些额外的压缩选项

在Scala中使用老API直接读取 KeyValueTextInputFormat()最简单的Hadoop输入格式 :
val input = sc.HadoopFile[Text, Text, KeyValueTextInputFormat](inputFile).map{ case (x, y) => (x.toString, y.toString) }

注:如果读取单个压缩过的输入,做好不要考虑使用Spark的封装(textFile/SequenceFile..),而是使用
newAPIHadoopFile 或者 HadoopFile,并指定正确的压缩解码器。
有些输入格式(如SequenceFile)允许我们只压缩键值对数据中的值,这在查询时很有用。其它一些输入格式也有自己的压缩控制,如:Twitter
Elephant Bird 包中的许多格式都可以使用LZO算法压缩数据。

时间: 2024-12-28 02:01:41

在Spark中使用Kryo序列化的相关文章

Spark 调优之RDD持久化级别及kryo序列化性能测试

我们上篇文章中讲了,RDD的持久化是spark优化中必须掌握的,并且,在内存不足的情况下,我们可以将持久化类型选择为MEMORY_ONLY_SER,减少内存的占用,持久化更多的partition,并且不同的序列化方法也会影响序列化性能.下面,我们就来测试下,持久化级别和序列化方法的选择对RDD持久化大小的影响.我选择了一个170.9MB的日志文件,传到了百度网盘 提取码:ffae 测试环境是windows,IDEA参数配置 MEMORY_ONLY 代码为 case class CleanedLo

Spark设置Kryo序列化缓冲区大小

背景 今天在开发SparkRDD的过程中出现Buffer Overflow错误,查看具体Yarn日志后发现是因为Kryo序列化缓冲区溢出了,日志建议调大spark.kryoserializer.buffer.max的value,搜索了一下设置keyo序列化缓冲区的方法,特此整理记录下来. 20/01/08 17:12:55 WARN scheduler.TaskSetManager: Lost task 1.0 in stage 1.0 (TID 4, s015.test.com, execut

java原生序列化和Kryo序列化性能比较

简介 最近几年,各种新的高效序列化方式层出不穷,不断刷新序列化性能的上限,最典型的包括: 专门针对Java语言的:Kryo,FST等等 跨语言的:Protostuff,ProtoBuf,Thrift,Avro,MsgPack等等 这些序列化方式的性能多数都显著优于hessian2(甚至包括尚未成熟的dubbo序列化).有鉴于此,我们为dubbo引入Kryo和FST这 两种高效Java序列化实现,来逐步取代hessian2.其中,Kryo是一种非常成熟的序列化实现,已经在Twitter.Group

spark中的广播变量broadcast

Spark中的Broadcast处理 首先先来看一看broadcast的使用代码: val values = List[Int](1,2,3) val broadcastValues = sparkContext.broadcast(values) rdd.mapPartitions(iter => { broadcastValues.getValue.foreach(println) }) 在上面的代码中,首先生成了一个集合变量,把这个变量通过sparkContext的broadcast函数进

spark 中的RDD编程 -以下基于Java api

1.RDD介绍:     RDD,弹性分布式数据集,即分布式的元素集合.在spark中,对所有数据的操作不外乎是创建RDD.转化已有的RDD以及调用RDD操作进行求值.在这一切的背后,Spark会自动将RDD中的数据分发到集群中,并将操作并行化. Spark中的RDD就是一个不可变的分布式对象集合.每个RDD都被分为多个分区,这些分区运行在集群中的不同节点上.RDD可以包含Python,Java,Scala中任意类型的对象,甚至可以包含用户自定义的对象. 用户可以使用两种方法创建RDD:读取一个

Tachyon在Spark中的作用(Tachyon: Reliable, Memory Speed Storage for Cluster Computing Frameworks 论文阅读翻译)

摘要: Tachyon是一种分布式文件系统,能够借助集群计算框架使得数据以内存的速度进行共享.当今的缓存技术优化了read过程,可是,write过程由于须要容错机制,就须要通过网络或者是磁盘进行复制操作.Tachyon通过将"血统"技术引入到存储层进而消除了这个瓶颈.创建一个长期的以"血统机制"为基础的存储系统的关键挑战是失败情况发生的时候及时地进行数据恢复.Tachyon通过引入一种检查点的算法来解决问题,这样的方法保证了恢复过程的有限开销以及通过资源调度器下进行

Hadoop中的Shuffle 与 Spark中的Shuffle得区别与联系

MapReduce过程.Spark和Hadoop以Shuffle为中心的对比分析 mapreduce与Spark的map-Shuffle-reduce过程 mapreduce过程解析(mapreduce采用的是sort-based shuffle) 将获取到的数据分片partition进行解析,获得k/v对,之后交由map()进行处理. map函数处理完成之后,进入collect阶段,对处理后的k/v对进行收集,存储在内存的环形缓冲区中. 当环形缓冲区中的数据达到阀值之后(也可能一直没有达到阀值

Hadoop与 Spark中的Shuffle之区别与联系

Hadoop与 Spark中的Shuffle之区别与联系 2018年08月22日 20:24:46 小爷欣欣 阅读数:175 转自:http://mini.eastday.com/mobile/180114141035935.html mapreduce过程解析(mapreduce采用的是sort-based shuffle),将获取到的数据分片partition进行解析,获得k/v对,之后交由map()进行处理.map函数处理完成之后,进入collect阶段,对处理后的k/v对进行收集,存储在

在Apache Spark中使用UDF

用户自定义函数(UDF)是大多数SQL环境的一个关键特性,其主要用于扩展系统的内置功能.UDF允许开发人员通过抽象其低级语言实现在更高级语言(如SQL)中应用的新函数.Apache Spark也不例外,其为UDF与Spark SQL工作流集成提供了各种选项. 在本篇博文中,我们将回顾Python.Java和Scala上的Apache Spark UDF和UDAF(用户自定义的聚合函数)实现的简单示例.我们还将讨论重要的UDF API功能和集成点,包括各发行版本之间的当前可用性.总而言之,我们将介