Spark 序列化问题

在Spark应用开发中，很容易出现以下报错：

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:2058) ... Caused by: java.io.NotSerializableException

 

该报错意思是用户代码的transformation操做中包含不可序列化的对象引用。

本文主要从如下三个方面解释Spark 应用中序列化问题 。 
一、Java序列化含义？ 
二、Spark代码为何须要序列化？ 
三、如何解决Spark序列化问题？

一、Java序列化含义？

Spark是基于JVM运行的进行，其序列化必然遵照Java的序列化规则。

序列化就是指将一个对象转化为二进制的byte流（注意，不是bit流），而后以文件的方式进行保存或经过网络传输，等待被反序列化读取出来。序列化常被用于数据存取和通讯过程当中。

对于java应用实现序列化通常方法：

• class实现序列化操做是让class 实现Serializable接口，但实现该接口不保证该class必定能够序列化，由于序列化必须保证该class引用的全部属性能够序列化。

• 这里须要明白，static和transient修饰的变量不会被序列化，这也是解决序列化问题的方法之一，让不能序列化的引用用static和transient来修饰。（static修饰的是类的状态，而不是对象状态，因此不存在序列化问题。transient修饰的变量，是不会被序列化到文件中，在被反序列化后，transient变量的值被设为初始值，如int是0，对象是null）

• 此外还能够实现readObject()方法和writeObject()方法来自定义实现序列化。（具体用例见参考连接）

二、Spark的transformation操做为何须要序列化？

Spark是分布式执行引擎，其核心抽象是弹性分布式数据集RDD，其表明了分布在不一样节点的数据。Spark的计算是在executor上分布式执行的，故用户开发的关于RDD的map，flatMap，reduceByKey等transformation 操做（闭包）有以下执行过程： 
1. 代码中对象在driver本地序列化 
2. 对象序列化后传输到远程executor节点 
3. 远程executor节点反序列化对象 
4. 最终远程节点执行 
故对象在执行中须要序列化经过网络传输，则必须通过序列化过程。

三、如何解决Spark序列化问题？

若是出现NotSerializableException报错，能够在spark-default.xml文件中加入以下参数来开启SerializationDebugger功能类，从而能够在日志中打印出序列化出问题的类和属性信息。

spark.executor.extraJavaOptions -Dsun.io.serialization.extendedDebugInfo=true spark.driver.extraJavaOption -Dsun.io.serialization.extendedDebugInfo=true

 

对于scala语言开发，解决序列化问题主要以下几点：

• 在Object中声明对象 （每一个class对应有一个Object）
• 若是在闭包中使用SparkContext或者SqlContext，建议使用SparkContext.get() and SQLContext.getActiveOrCreate()
• 使用static或transient修饰不可序列化的属性从而避免序列化。 
  注：scala语言中，class的Object

对于java语言开发，对于不可序列化对象，若是自己不须要存储或传输，则可以使用static或trarnsient修饰；若是须要存储传输，则实现writeObject()/readObject()使用自定义序列化方法。

此外注意

对于Spark Streaming做业，注意哪些操做在driver，哪些操做在executor。由于在driver端（foreachRDD）实例化的对象，极可能不能在foreach中运行，由于对象不能从driver序列化传递到executor端（有些对象有TCP连接，必定不能够序列化）。因此这里通常在foreachPartitions或foreach算子中来实例化对象，这样对象在executor端实例化，没有从driver传输到executor的过程。

dstream.foreachRDD { rdd =>
  val where1 = "on the driver" rdd.foreach { record => val where2 = "on different executors" } } }

 

参考资料： 
Avoid NotSerializable Error in Spark Job 
spark not serializable problem 
Spark Streaming / Tips on Running Streaming Apps inside Databricks 
Java 序列化的高级认识 
什么是writeObject 和readObject？可定制的序列化过程