RDD 序列化

Posted 2023-03-01 气质&末雨

文章目录

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

RDD 序列化

1、闭包检查

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

2、序列化方法和属性

从计算的角度，算子以外的代码都是在Driver端执行，算子里面的代码都是在Executor端执行，看如下代码：
说明：类的构造器本身就是类的属性，用了类的属性传递到其他的函数，相当于是在调用这个类，所以需要进行序列化，或者用一个变量来进行接收这个类的属性，那么就不需要混入了，因为这个变量是字符串是在Driver端的，本身就可以序列化的。

package com.atguigu.bigdata.spark.core.wc.serial

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

//这个包里是序列化的一些测试代码
class Spark01_RDD_Serial 


object Spark01_RDD_Serial
  def main(args: Array[String]): Unit = 

    val conf = new SparkConf().setMaster("local[*]").setAppName("RDD行动算子")
    val context = new SparkContext(conf)
    val rdd: RDD[String] = context.makeRDD(Array("hello world", "hello spark", "hive", "aex"))

    val search = new Search("h") //如果数据源中的数据包含了这个小写的h，这个数据留下来，不包含的数据就不要了
    search.getMatch1(rdd).collect().foreach(println)
    search.getMatch2(rdd).collect().foreach(println)

    context.stop()
  
  //查询对象
  //类的构造参数其实是类的属性,构造参数需要进行闭包检测，其实就等同于类进行我们的闭包检测
  //混入序列化就没事了
  class Search(query:String)  extends Serializable 
    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)) //第二种是直接使用匿名函数
    

  

3、Kryo序列化框架

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