Scala Spark报错 task not serializable的解决办法

你的名字 2024-04-17 18:20 56阅读 0赞

*  更新时间：2019-08-29

### 错误描述 ###

今天在写spark程序时，遇到task not serializable的报错，提示task未进行序列化。  
在正常spark程序在执行时会进行task序列化，当一些函数里面有外部变量时，不会序列化外部变量，由此报错。  
上面报错的主要原因是在map函数的闭包内引入外部函数，外部变量没有进行序列化，我的代码结构如下：

def test(sc: SparkContext, rdd1: RDD[((String, String), Float)], rdd2: [((String, String), Float)], isFunc: Boolean): Unit = { 
       
        val rdd_joined = rdd1.leftOuterJoin(rdd2).map({  x =>
          if (x._2._2.getOrElse(0) != 0 ) (x._1._1, (x._1._2,  if(isFunc == false) 1 else Func(0).toFloat))
          else (x._1._1, (x.1._2, 0.0.toFloat))
        }).filter(_._2._2 != 0.0)
        ... ...    
      }
    
    def Func(z: Float): Float = { 
            z
          }

### 解决办法 ###

（1）不在map这类函数的闭包内引入外部变量  
（2）如果一定要引入，提前对相应的类做好序列化操作  
由于上面报错是因为在map闭包内引入了Func函数，所以干脆把Func函数放到map中，这样就修复了，如下：

def test(sc: SparkContext, rdd1: RDD[((String, String), Float)], rdd2: [((String, String), Float)], isFunc: Boolean): Unit = { 
       
        val rdd_joined = rdd1.leftOuterJoin(rdd2).map({  x =>
          def Func(z: Float): Float = { 
            z
          }
          if (x._2._2.getOrElse(0) != 0 ) (x._1._1, (x._1._2,  if(isFunc == false) 1 else Func(0).toFloat))
          else (x._1._1, (x.1._2, 0.0.toFloat))
        }).filter(_._2._2 != 0.0)
        ... ...    
      }

参考：  
https://bbs.csdn.net/topics/390844956  
https://blog.csdn.net/qq\_38899793/article/details/80351205  
https://blog.csdn.net/achilles12345/article/details/77778431  
https://blog.csdn.net/javastart/article/details/51206715