目錄

• 1.閉包檢查
• 2.序列化方法和屬性
• 3.序列化異常解決方案
• 4.Kryo序列化框架

1.閉包檢查

​ 從計算的角度, 運算元以外的程式碼都是在Driver端執行, 運算元裡面的程式碼都是在Executor端執行。那麼在scala的函數語言程式設計中，就會導致運算元內經常會用到運算元外的資料，這樣就形成了閉包的效果，如果使用的運算元外的資料無法序列化，就意味著無法傳值給Executor端執行，就會發生錯誤，所以需要在執行任務計算前，檢測閉包內的物件是否可以進行序列化，這個操作我們稱之為閉包檢測。

如圖

2.序列化方法和屬性

從計算的角度, 運算元以外的程式碼都是在Driver端執行, 運算元裡面的程式碼都是在Executor端執行

object serializable02_function {

    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
        val sc: SparkContext = new SparkContext(conf)

        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)//rdd引用了物件的成員方法產生閉包
    }

    // 屬性序列化案例
    def getMatch2(rdd: RDD[String]): RDD[String] = {
        //rdd.filter(x => x.contains(this.query))
        rdd.filter(x => x.contains(query))//rdd引用了物件的成員屬性產生閉包     
        /*解決方案：將成員屬性值賦值給Int區域性變數，基本型別已經實現了序列化。*/
        //val q = query
        //rdd.filter(x => x.contains(q))
    }
}

 

3.序列化異常解決方案

①extends Serializable：類實現序列化

②使用case class：使用樣例類，樣例類預設實現序列化

③不引用物件的成員方法和變數，可以使用區域性變數儲存屬性的值、 使用函式或匿名函式！

4.Kryo序列化框架

​ Java的序列化能夠序列化任何的類。但是比較重（位元組多），序列化後，物件的提交也比較大。Spark出於效能的考慮，Spark2.0開始支援另外一種Kryo序列化機制。Kryo速度是Serializable的10倍。當RDD在Shuffle資料的時候，簡單資料型別、陣列和字串型別已經在Spark內部使用Kryo來序列化。

注意：即使使用Kryo序列化，也要繼承Serializable介面。

使用方式

將需要使用Kryo序列化的物件在conf中註冊，即可使用

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)