RDD序列化
1. 闭包检查
从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在 Executor端执行。那么在 scala 的函数式编程中,就会导致算子内经常会用到算子外的数据,这样就形成了闭包的效果,如果使用的算子外的数据无法序列化,就意味着无法传值给Executor端执行,就会发生错误,所以需要在执行任务计算前,检测闭包内的对象是否可以进行序列化,这个操作我们称之为闭包检测。
注:Scala2.12 版本后闭包编译方式发生了改变 。
2. 序列化方法和属性
从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在 Executor端执行,看如下代码:
package com.atguigu.bigdata.spark.core.rdd.serial
import 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.创建一个 RDD
val 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 = query
rdd.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.serial
import 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))
}
}