Spark的RDD序列化

RDD序列化

1. 闭包检查

从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在 Executor端执行。那么在 scala 的函数式编程中，就会导致算子内经常会用到算子外的数据，这样就形成了闭包的效果，如果使用的算子外的数据无法序列化，就意味着无法传值给Executor端执行，就会发生错误，所以需要在执行任务计算前，检测闭包内的对象是否可以进行序列化，这个操作我们称之为闭包检测。
注：Scala2.12 版本后闭包编译方式发生了改变。

2. 序列化方法和属性

从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在 Executor端执行，看如下代码：

package com.atguigu.bigdata.spark.core.rdd.serialimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object Spark01_RDD_Serial {def main(args: Array[String]): Unit = {//TODO 准备环境val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")val sc = new SparkContext(sparkConf)//3.创建一个 RDDval rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))val search = new Search("h")//    search.getMatch1(rdd).collect().foreach(println)search.getMatch2(rdd).collect().foreach(println)//TODO 关闭环境sc.stop()}//  class Search(query:String) extends Serializable {class Search(query:String){def isMatch(s: String): Boolean = {s.contains(query)}// 函数序列化案例def getMatch1 (rdd: RDD[String]): RDD[String] = {rdd.filter(isMatch)}// 属性序列化案例def getMatch2(rdd: RDD[String]): RDD[String] = {val s = queryrdd.filter(x => x.contains(s))// 对应第二个rdd}}}

3. Kryo 序列化框架

Java 的序列化能够序列化任何的类。但是比较重（字节多），序列化后，对象的提交也比较大。Spark 出于性能的考虑，Spark2.0 开始支持另外一种Kryo 序列化机制。Kryo 速度是Serializable 的10 倍。当 RDD 在Shuffle 数据的时候，简单数据类型、数组和字符串类型已经在Spark 内部使用 Kryo 来序列化。
注意：即使使用Kryo 序列化，也要继承Serializable 接口。

package com.atguigu.bigdata.spark.core.rdd.serialimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object serializable_Kryo { def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf() .setAppName("SerDemo") .setMaster("local[*]") // 替换默认的序列化机制 .set("spark.serializer",
"org.apache.spark.serializer.KryoSerializer") // 注册需要使用 kryo 序列化的自定义类 .registerKryoClasses(Array(classOf[Searcher])) val sc = new SparkContext(conf) val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu
"atguigu", "hahah"), 2) val searcher = new Searcher("hello") val result: RDD[String] = searcher.getMatchedRDD1(rdd) result.collect.foreach(println) }
} case class Searcher(val query: String) { def isMatch(s: String) = { s.contains(query) } def getMatchedRDD1(rdd: RDD[String]) = { rdd.filter(isMatch)  } def getMatchedRDD2(rdd: RDD[String]) = { val q = query rdd.filter(_.contains(q)) }
}

Spark的RDD序列化相关推荐

Spark的RDD 文件读取与保存
RDD 文件读取与保存 Spark 的数据读取及数据保存可以从两个维度来作区分:文件格式以及文件系统.其中: (1)文件格式分为:text 文件.csv 文件.sequence 文件以及Object ...
【Spark】Spark的Kryo序列化
1.美图 2.Spark序列化概述在Spark的架构中,在网络中传递的或者缓存在内存.硬盘中的对象需要进行序列化操作,序列化的作用主要是利用时间换空间: 分发给Executor上的Task 需要缓存 ...
spark之kryo序列化及其使用
spark之kryo序列化 spark之kryo 序列化 Spark 中使用 Kryo序列化中文切词案例: spark之kryo 序列化 1.定义:把对象转换为字节序列的过程称为对象的序列化. 把字 ...
spark数据处理-RDD
文章目录 spark数据处理笔记 spark核心介绍 RDD编程 RDD介绍-弹性分布式数据集创建RDD两种方式函数传递常见RDD转化操作和行动操作常用的转化操作: 类集合操作行动操作不同 ...
spark之kryo 序列化
几乎所有的资料都显示kryo 序列化方式优于java自带的序列化方式,而且在spark2.*版本中都是默认采用kryo 序列化.因此本文将做kryo 做一个测试以验证其性能. 1.先给出定义: ...
Spark：RDD编程总结(概述、算子、分区、共享变量)
目录 1.RDD概述 1.1.RDD是什么 1.2.RDD的弹性 1.3.RDD的特点 1.3.1.分区 1.3.2.只读 1.3.3.依赖 1.3.4.缓存 1.3.5.检查点 2.RDD编程 2. ...
spark的RDD中的action(执行)和transformation(转换)两种操作中常见函数介绍
参考文章:spark的RDD中的action(执行)和transformation(转换)两种操作中常见函数介绍 spark常见的RDD 1. 函数概览 2. 常见的Transformations 操 ...
Spark 把RDD数据保存到hdfs单个文件中，而不是目录
相比于Hadoop,Spark在数据的处理方面更加灵活方便.然而在最近的使用中遇到了一点小麻烦:Spark保存文件的的函数(如saveAsTextFile)在保存数据时都需要新建一个目录,然后在这个目 ...
Spark的RDD分区器
RDD 分区器基本介绍 Spark 目前支持Hash 分区.Range 分区和用户自定义分区.Hash 分区为当前的默认分区.分区器直接决定了RDD 中分区的个数.RDD 中每条数据经过Shuffl ...