spark not serializable异常分析及解决方案
转载自: http://bigdataer.net/?p=569
1.背景
在使用spark开发分布式数据计算作业过程中或多或少会遇到如下的错误:
SRE实战 互联网时代守护先锋,助力企业售后服务体系运筹帷幄!一键直达领取阿里云限量特价优惠。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(_)) } }
第一种方法比较简单实用。
更多精彩