【Spark】Task not serializable exception while running apache spark job

小咪咪 2023-06-08 03:05 20阅读 0赞

### 文章目录 ###

*   *  1、概述
     *  2、问题重现
     *  3、问题分析
     *  4、解决方法
     *   *  4.1、仅在map中传递lambda函数中声明实例
         *  4.2、将方法封装为高阶函数
         *  4.3、使未序列化的类继承 java.io.Serializable 接口
         *  4.4、注册序列化类（适用第三方包）
     *  5、避免序列化问题的经验

## 1、概述 ##

在使用spark编写分布式数据计算作业的过程中，我遇到了很多问题，今天跟大家分享一个 spark 作业序列化的问题，我们看一下异常信息，是不是觉得很眼熟：

org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: ...

## 2、问题重现 ##

object SparkTaskNotSerializable {
      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 50, 5)
        val usz = new UnserializableClass()
        rdd.map(x=>usz.method(x)).foreach(println(_))
      }
    }
    class UnserializableClass {
      def method(x:Int):Int={
        x*2
      }
    }

运行以上代码，将会出现异常信息：

Exception in thread "main" org.apache.spark.SparkException: Task not serializable
    	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:345)
    	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:335)
    	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:159)
    	at org.apache.spark.SparkContext.clean(SparkContext.scala:2292)
    	at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:371)
    	at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:370)
    	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    	at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
    	at org.apache.spark.rdd.RDD.map(RDD.scala:370)
    	at com.lkf.spark.SparkTaskNotSerializable$.main(SparkTaskNotSerializable.scala:17)
    	at com.lkf.spark.SparkTaskNotSerializable.main(SparkTaskNotSerializable.scala)
    Caused by: java.io.NotSerializableException: com.lkf.spark.UnserializableClass
    Serialization stack:
    	- object not serializable (class: com.lkf.spark.UnserializableClass, value: com.lkf.spark.UnserializableClass@136ccbfe)
    	- field (class: com.lkf.spark.SparkTaskNotSerializable$$anonfun$main$1, name: usz$1, type: class com.lkf.spark.UnserializableClass)
    	- object (class com.lkf.spark.SparkTaskNotSerializable$$anonfun$main$1, <function1>)
    	at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
    	at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
    	at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
    	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:342)
    	... 11 more

## 3、问题分析 ##

spark 处理的数据单元为RDD（即弹性分布式数据集），当我们要对RDD做map,filter等操作的时候是在excutor上完成的。但是如果我们在 driver 中定义了一个变量，在 map 等操作中使用了，则这个变量就会被分发到各 个excutor。

因为 driver 和 excutor 运行在不同的jvm中，会涉及到对象的序列化与反序列化。如果这个变量没法序列化就会报异常。还有一种情况就是引用的对象可以序列化，但是该对象本身引用的其他对象无法序列化，也会有异常。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2NDcwMzU_size_16_color_FFFFFF_t_70]

## 4、解决方法 ##

### 4.1、仅在map中传递lambda函数中声明实例 ###

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 50, 5)
        // 在 map 中实例化对象 UnserializableClass
        rdd.map(x => new UnserializableClass().method(x)).foreach(println(_))
      }

### 4.2、将方法封装为高阶函数 ###

将方法修改为函数

class UnserializableClass {
      //method方法
      /*def method(x:Int):Int={
        x * 2
      }*/
    
      //method函数
      val method = (x:Int)=>x*2
    }

直接使用函数

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 50, 5)
        val usz  = new UnserializableClass()
        //传入函数
        rdd.map(usz.method).foreach(println(_))
      }

### 4.3、使未序列化的类继承 java.io.Serializable 接口 ###

class UnserializableClass extends java.io.Serializable {
      def method(x: Int): Int = {
        x * 2
      }
    }

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 50, 5)
        val usz = new UnserializableClass()
        rdd.map(usz.method).foreach(println(_))
      }

### 4.4、注册序列化类（适用第三方包） ###

如果 UnserializableClass 来自于第三方包，我们将无法修改其源码该怎么办，此时我们可以使用注册序列化类的方法。

def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]").setAppName("test")
        //指定序列化类为KryoSerializer
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        //将UnserializableClass注册到kryo需要序列化的类中
        conf.registerKryoClasses(Array(classOf[com.lkf.spark.UnserializableClass]))
        val sc = new SparkContext(conf)
        val rdd = sc.parallelize(1 to 50, 5)
        val usz = new UnserializableClass()
        rdd.map(x => usz.method(x)).foreach(println(_))
      }

## 5、避免序列化问题的经验 ##

1、避免使用匿名类，使用静态类，因为匿名类将迫使您将外部类序列化。

2、避免使用静态变量来解决序列化问题，因为“多个任务”可以在同一JVM内运行，并且静态实例可能不是线程安全的。

3、使用Transient变量来避免序列化问题

4、使用静态类代替匿名类。

5、在“ lambda函数”内部永远不要直接引用outclass方法，因为这将导致外部类的序列化。

6、如果需要直接在Lambda函数中使用方法，请将方法设为静态；否则，请使用Class :: func（），而不要直接使用func（）

7、Java Map <>没有实现Serializable，但是HashMap实现了。

8、在决定使用广播还是原始数据结构时要斟酌考虑，如果您很确定请尽量使用广播。

参考资料：

https://stackoverflow.com/questions/25914057/task-not-serializable-exception-while-running-apache-spark-job

https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/troubleshooting/javaionotserializableexception.html

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2NDcwMzU_size_16_color_FFFFFF_t_70]: /images/20230601/1aef0980e0bd40019a03e44428ac17c3.png