spark not serializable異常分析及解決方案

轉載自: http://bigdataer.net/?p=569

1.背景

在使用spark開發分布式數據計算作業過程中或多或少會遇到如下的錯誤：

Serialization stack:
object not serializable (class:class: org.apache.hadoop.hbase.io.ImmutableBytesWritable, value: 30 30 30 30 30 30 32 34 32 30 32 37 37 32 31)
field (class: scala.Tuple2, name: _1, type: class java.lang.Object) ……

或者如下的錯誤：

org.apache.spark.SparkException: Task not serializable at org.apache.spark.util.ClosureCleaner …

表面意思都是無法序列化導致的。spark運行過程中為什么要序列化？下面來分析一下。

2.分析

spark處理的數據單元為RDD（即彈性分布式數據集），當我們要對RDD做諸如map,filter等操作的時候是在excutor上完成的。但是如果我們在driver中定義了一個變量，在map等操作中使用，則這個變量就要被分發到各個excutor，因為driver和excutor的運行在不同的jvm中，勢必會涉及到對象的序列化與反序列化。如果這個變量沒法序列化就會報異常。還有一種情況就是引用的對象可以序列化，但是引用的對象本身引用的其他對象無法序列化，也會有異常。

 

3.解決方案

(1) 舉例

class UnserializableClass {
    def method(x:Int):Int={
        x*x
    }
}

另外，有如下的spark代碼塊：

object SparkTest {
  def main(args: Array[String]): Unit = {
  val conf = new SparkConf().setMaster("local[*]").setAppName("test")
  val sc = new SparkContext(conf)
  val rdd = sc.parallelize(1 to 10, 3)
  val usz = new UnserializableClass()
  rdd.map(x=>usz.method(x)).foreach(println(_))
  }
}

那么運行的時候就會拋出異常

Exception in thread “main” org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)
at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:324)
at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:323)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
at org.apache.spark.rdd.RDD.map(RDD.scala:323)
at net.bigdataer.spark.SparkTest$.main(SparkTest.scala:16)

(2) 解決方案

1. 將不可序列化的對象定義在閉包內

object SparkTest {
  def main(args: Array[String]): Unit = {
  val conf = new SparkConf().setMaster("local[*]").setAppName("test")
  val sc = new SparkContext(conf)
  val rdd = sc.parallelize(1 to 10,3)
  rdd.map(x=>new UnserializableClass().method(x)).foreach(println(_)) //在map中創建UnserializableClass對象
  }
 }

2.將所調用的方法改為函數，在高階函數中使用

class UnserializableClass {
  //method方法
  /*def method(x:Int):Int={
    x*x
  }*/

//method函數
  val method = (x:Int)=>x*x
}

在SparkTest中傳入函數

object SparkTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("test")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(1 to 10,3)
    val usz  = new UnserializableClass()
    rdd.map(usz.method).foreach(println(_)) //注意這里傳入的是函數
  }
}

3.給無法序列化的類加上java.io.Serializable接口

class UnserializableClass extends java.io.Serializable{ //加接口
  def method(x:Int):Int={
    x*x
  }
}

4.注冊序列化類

以上三個方法基於UnserializableClass可以被修改來說的，假如UnserializableClass來自於第三方，你無法修改其源碼就可以使用為其注冊序列化類的方法。

object SparkTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("test")

    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //指定序列化類為KryoSerializer
    conf.registerKryoClasses(Array(classOf[net.bigdataer.spark.UnserializableClass])) //將UnserializableClass注冊到kryo需要序列化的類中

    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(1 to 10,3)
    val usz  = new UnserializableClass()
    rdd.map(x=>usz.method(x)).foreach(println(_))
  }
}

 

第一種方法比較簡單實用。