【回顾】RDD的序列化

文章目录

• ​​RDD 序列化​​

• ​​1、闭包检查​​
• ​​2、序列化方法和属性​​
• ​​3、Kryo 序列化框架​​

RDD 序列化

1、闭包检查

从计算的角度, 算子以外的代码都是在 Driver 端执行, 算子里面的代码都是在 Executor 端执行。那么在 scala 的函数式编程中，就会导致​

​算子内经常会用到算子外的数据，这样就形成了闭包的效果​

​，如果使用的算子外的数据无法序列化，就意味着无法传值给Executor端执行，就会发生错误，所以需要在执行任务计算前，检测闭包内的对象是否可以进行序列化，这个操作我们称之为闭包检测。Scala2.12 版本后闭包编译方式发生了改变。

​​返回顶部​​

2、序列化方法和属性

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

object serializable02_function {
   def main(args: Array[String]): Unit = {
      // 1.创建 SparkConf 并设置 App 名称
      val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
     // 2.创建 SparkContext，该对象是提交 Spark App 的入口
     val sc: SparkContext = new SparkContext(conf)
      // 3.创建一个 RDD
      val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))
      //  3.1 创建一个 Search 对象
      val search = new Search("hello")
      //  3.2 函数传递，打印：ERROR Task not serializable
      search.getMatch1(rdd).collect().foreach(println)
      //  3.3 属性传递，打印：ERROR Task not serializable
      search.getMatch2(rdd).collect().foreach(println)
      // 4.关闭连接
      sc.stop()
    }
}
class Search(query:String) extends Serializable {
   def isMatch(s: String): Boolean = {
      s.contains(query)
   }
   // 函数序列化案例
   def getMatch1 (rdd: RDD[String]): RDD[String] = {
       //rdd.filter(this.isMatch)
       rdd.filter(isMatch)
   }

   // 属性序列化案例
   def getMatch2(rdd: RDD[String]): RDD[String] = {
      //rdd.filter(x => x.contains(this.query))
      rdd.filter(x => x.contains(query))
      //val q = query
      //rdd.filter(x => x.contains(q))
   }
}      

类的构造参数其实是类的属性（即query属性），构造参数需要进行闭包检测，其实就等共同于类进行闭包检测。

无论是调用getMatch1还是getMatch2都直接或间接地需要用到query属性，而算子以外的代码都是在 Driver 端执行, 算子里面的代码都是在 Executor端执行，所以这里的query是存在与Driver中，但是计算存在于Executor，所以在使用query的时候必须将Search类进行序列化。

在作业job提交之前，其中有一行代码，​

​val cleanF = sc.clean(f)​

​, 用于进行闭包检测，之所以叫闭包检测，是因为当前函数的内部访问了外部函数的变量，属于闭包的形式。按照这种逻辑，还有一种解决方法，我们在使用query属性之前将其赋值给函数内部的局部变量 ​

​val q = query​

​，然后我们就使用变量q来完成操作：​

​rdd.filter(x => x.contains(q))​

​。

​​返回顶部​​

3、Kryo 序列化框架

Java 的序列化能够序列化任何的类。但是比较重（字节多），序列化后，对象的提交也比较大。Spark 出于性能的考虑，Spark2.0 开始支持另外一种 ​

​Kryo 序列化机制​

​。Kryo 速度是 Serializable 的 10 倍。当 RDD 在 Shuffle 数据的时候，简单数据类型、数组和字符串类型已经在 Spark 内部使用 Kryo 来序列化。

注意：即使使用 Kryo 序列化，也要继承 Serializable 接口（或者使用局部变量）。

package test03_rdd.serializable

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark02_Kryo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("SerDemo")
      .setMaster("local[*]")
      // 1、替换默认的序列化机制
      .set("spark.serializer",
        "org.apache.spark.serializer.KryoSerializer")
      // 2、注册需要使用 kryo 序列化的自定义类
      .registerKryoClasses(Array(classOf[Searcher]))
    val sc = new SparkContext(conf)
    val rdd = sc.makeRDD(Array("hello world", "hello spark", "scala"), 2)
    val searcher = new Searcher("hello")
    val result = searcher.getMatchedRDD1(rdd)
    result.collect.foreach(println)
  }
}

case class Searcher(val query: String) {
  def isMatch(s: String): Boolean = {
    s.contains(query)
  }

  def getMatchedRDD1(rdd: RDD[String]): RDD[String] = {
    rdd.filter(isMatch)
  }

  def getMatchedRDD2(rdd: RDD[String]): RDD[String] = {
    val q = query
    rdd.filter(_.contains(q))
  }
}