org.apache.spark.SparkException：任务不可序列化——Scala答案

【问题标题】：org.apache.spark.SparkException: Task not serializable -- Scalaorg.apache.spark.SparkException：任务不可序列化——Scala
【发布时间】：2019-05-19 18:07:37
【问题描述】：

我正在阅读一个文本文件，它是一个固定宽度的文件，我需要将其转换为 csv。我的程序在本地机器上运行良好，但是当我在集群上运行它时，它会抛出“Task not serializable”异常。

我尝试用 map 和 mapPartition 解决同样的问题。

在 RDD 上使用 toLocalIterator 可以正常工作。但它不适用于大文件（我有 8GB 的文件）

下面是我最近尝试过的使用 mapPartition 的代码

//读取源文件并创建RDD

def main(){
    var inpData = sc.textFile(s3File)
    LOG.info(s"\n inpData >>>>>>>>>>>>>>> [${inpData.count()}]")

    val rowRDD = inpData.mapPartitions(iter=>{
    var listOfRow = new ListBuffer[Row]
    while(iter.hasNext){
       var line = iter.next()
       if(line.length() >= maxIndex){
          listOfRow += getRow(line,indexList)
        }else{
          counter+=1
        }
     }
    listOfRow.toIterator
    })

    rowRDD .foreach(println)
}

case class StartEnd(startingPosition: Int, endingPosition: Int) extends Serializable

def getRow(x: String, inst: List[StartEnd]): Row = {
    val columnArray = new Array[String](inst.size)
    for (f <- 0 to inst.size - 1) {
      columnArray(f) = x.substring(inst(f).startingPosition, inst(f).endingPosition)
    }
    Row.fromSeq(columnArray)
}

//注意：我使用StartEnd案例类创建的indexList，供您参考，创建后如下所示

[List(StartEnd(0,4), StartEnd(4,10), StartEnd(7,12), StartEnd(10,14))]

这个程序在我的本地机器上运行良好。但是当我使用集群（AWS）时，它会抛出异常，如下所示。

>>>>>>>>Map(ResultantDF -> [], ExceptionString -> 
Exception occurred while applying the FileConversion transformation and the exception Message is :Task not serializable
Exception occurred while applying the FileConversion transformation and the exception Message is :Task not serializable)
[Driver] TRACE reflection.ReflectionInvoker$.invokeDTMethod - Exit

我无法理解这里出了什么问题，什么是不可序列化的，为什么会抛出异常。

感谢任何帮助。提前致谢！

【问题讨论】：

基本上一个包含记录器的实例（可能是LOG）被序列化导致问题。可能StartEnd是一个内部类什么的，使用提供的代码无法确定确切原因

标签： scala apache-spark apache-spark-sql

【解决方案1】：

您在 Spark mapPartition 转换中调用 getRow 方法。它迫使 spark 将主类的实例传递给工人。主类包含LOG 作为字段。似乎此日志对序列化不友好。你可以

a) 将 getRow 的 LOG 移动到不同的 object（解决此类问题的一般方法）

b) 将 LOG 设为 lazy val

c) 使用另一个日志库

【讨论】：

我将 getRow 函数放在不同的对象中