SparkCore之RDD序列化

Myth丶恋晨 2022-12-31 06:24 194阅读 0赞

# SparkCore之RDD序列化 #

> 在SparkRDD编程中，初始化工作是在Driver端进行的，而实际运行程序是在Executor端进行的，涉及到了跨进程通信，是需要序列化的。  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

## 1. 闭包检查 ##

**1）闭包引入（有闭包就需要进行序列化）**

object serializable01_object { 
    
        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.创建两个对象
            val user1 = new User()
            user1.name = "zhangsan"
    
            val user2 = new User()
            user2.name = "lisi"
    
            val userRDD1: RDD[User] = sc.makeRDD(List(user1, user2))
    
            //3.1 打印，ERROR报java.io.NotSerializableException
            //userRDD1.foreach(user => println(user.name))
            
    
            //3.2 打印，RIGHT （因为没有传对象到Executor端）
            val userRDD2: RDD[User] = sc.makeRDD(List())
            //userRDD2.foreach(user => println(user.name))
    
            //3.3 打印，ERROR Task not serializable 注意：没执行就报错了
            userRDD2.foreach(user => println(user1.name))
    
            //4.关闭连接
            sc.stop()
        }
    }
    
    //class User { 
    // var name: String = _
    //}
    class User extends Serializable { 
        var name: String = _
    }

## 2. 序列化方法和属性 ##

1）说明

> **Driver：算子以外的代码都是在Driver端执行  
> Executor：算子里面的代码都是在Executor端执行**

2）代码实现

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")
    
            // Driver：算子以外的代码都是在Driver端执行
            // Executor：算子里面的代码都是在Executor端执行
            //3.2 函数传递，打印：ERROR Task not serializable
            search.getMatch1(rdd).collect().foreach(println)
    
            //3.3 属性传递，打印：ERROR Task not serializable
            search.getMatche2(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 getMatche2(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))
        }
    }

3）问题一说明

//过滤出包含字符串的RDD
    def getMatch1 (rdd: RDD[String]): RDD[String] = { 
        rdd.filter(isMatch)
    }

> （1）在这个方法中所调用的方法isMatch()是定义在Search这个类中的，实际上调用的是this. isMatch()，this表示Search这个类的对象，程序在运行过程中需要将Search对象序列化以后传递到Executor端。  
> （2）解决方案  
> 类继承scala.Serializable即可。  
> class Search() extends Serializable\{…\}

4）问题二说明

//过滤出包含字符串的RDD
    def getMatche2(rdd: RDD[String]): RDD[String] = { 
        rdd.filter(x => x.contains(query))
    }

> （1）在这个方法中所调用的方法query是定义在Search这个类中的字段，实际上调用的是this. query，this表示Search这个类的对象，程序在运行过程中需要将Search对象序列化以后传递到Executor端。  
> （2）解决方案一  
> （a）类继承scala.Serializable即可。  
> class Search() extends Serializable\{…\}  
> （b）将类变量query赋值给局部变量  
> 修改getMatche2为  
> //过滤出包含字符串的RDD  
> def getMatche2(rdd: RDD\[String\]): RDD\[String\] = \{  
> val q = this.query//将类变量赋值给局部变量  
> rdd.filter(x => x.contains(q))  
> \}  
> （3）解决方案二  
> 把Search类变成样例类，样例类默认是序列化的。  
> case class Search(query:String) \{…\}

## 3. Kryo序列化框架 ##

> 参考地址: https://github.com/EsotericSoftware/kryo  
> Spark默认使用Java的序列化器，Java的序列化能够序列化任何的类。但是比较重，序列化后对象的体积也比较大。  
> Spark出于性能的考虑，Spark2.0开始支持另外一种Kryo序列化机制。Kryo速度是Serializable的10倍。当RDD在Shuffle数据的时候，简单数据类型、数组和字符串类型已经在Spark内部使用Kryo来序列化。  
> 注意：即使使用Kryo序列化，也要继承Serializable接口。

object serializable03_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[Searche]))
    
            val sc = new SparkContext(conf)
    
            val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", "atguigu", "hahah"), 2)
    
            val searche = new Searche("hello")
            val result: RDD[String] = searche.getMatchedRDD1(rdd)
    
            result.collect.foreach(println)
        }
    }
    
    //使用样例类，底层已经封装了继承serizable接口
    case class Searche(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))
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221120/16a962ea1e394fad9123f8308bace6b5.png