spark not serializable异常分析及解决方案

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(_))

  }

}

第一种方法比较简单实用。

巴特西