列表的Spark/Scala序列化。任务不可序列化：java.io.NotSerializable异常

Question

问题在于Spark数据集和INT列表的序列化。Scala版本是2.10.4，Spark版本是1.6。

这和其他问题类似，但是我不能基于这些回答让它工作。我已经简化了代码，以便仅仅显示问题。

我有一门案例课：

case class FlightExt(callsign: Option[String], serials: List[Int])

我的主要方法是：

    val (ctx, sctx) = SparkUtil.createContext() // just a helper function to build context
    val flightsDataFrame = separateFlightsMock(sctx) // reads data from Parquet file

    import sctx.implicits._
    flightsDataFrame.as[FlightExt]
      .map(flight => flight.callsign)
      .show()

我得到以下错误：

SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: scala.reflect.internal.Symbols$PackageClassSymbol
Serialization stack:
    - object not serializable (class: scala.reflect.internal.Symbols$PackageClassSymbol, value: package scala)
    - field (class: scala.reflect.internal.Types$ThisType, name: sym, type: class scala.reflect.internal.Symbols$Symbol)
    - object (class scala.reflect.internal.Types$UniqueThisType, scala.type)
    - field (class: scala.reflect.internal.Types$TypeRef, name: pre, type: class scala.reflect.internal.Types$Type)
    - object (class scala.reflect.internal.Types$TypeRef$$anon$6, scala.Int)
    - field (class: org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$5, name: elementType$2, type: class scala.reflect.api.Types$TypeApi)
    - object (class org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$5, <function1>)
    - field (class: org.apache.spark.sql.catalyst.expressions.MapObjects, name: function, type: interface scala.Function1)
    - object (class org.apache.spark.sql.catalyst.expressions.MapObjects, mapobjects(<function1>,cast(serials#7 as array<int>),IntegerType))
    - field (class: org.apache.spark.sql.catalyst.expressions.Invoke, name: targetObject, type: class org.apache.spark.sql.catalyst.expressions.Expression)
    - object (class org.apache.spark.sql.catalyst.expressions.Invoke, invoke(mapobjects(<function1>,cast(serials#7 as array<int>),IntegerType),array,ObjectType(class [Ljava.lang.Object;)))
    - writeObject data (class: scala.collection.immutable.$colon$colon)
    - object (class scala.collection.immutable.$colon$colon, List(invoke(mapobjects(<function1>,cast(serials#7 as array<int>),IntegerType),array,ObjectType(class [Ljava.lang.Object;))))
    - field (class: org.apache.spark.sql.catalyst.expressions.StaticInvoke, name: arguments, type: interface scala.collection.Seq)
    - object (class org.apache.spark.sql.catalyst.expressions.StaticInvoke, staticinvoke(class scala.collection.mutable.WrappedArray$,ObjectType(interface scala.collection.Seq),make,invoke(mapobjects(<function1>,cast(serials#7 as array<int>),IntegerType),array,ObjectType(class [Ljava.lang.Object;)),true))
    - writeObject data (class: scala.collection.immutable.$colon$colon)

如果我从FlightExt中删除列表，那么一切正常，这表明lambda函数序列化没有问题。

Scala本身似乎序列化了一系列Int的优点。也许Spark在序列化列表方面有问题？

我还尝试过使用Java整数。

编辑：

如果我将列表更改为数组，它会工作，但如果我有如下内容：

case class FlightExt(callsign: Option[String], other: Array[AnotherCaseClass])

它也会因同样的错误而失败

我是Scala和Spark的新手，可能遗漏了一些东西，但如果有任何解释，我将不胜感激。

Answer 1

将FlightExt类放入对象中，检查下面的代码。

object Flight {
 case class FlightExt(callsign: Option[String], var serials: List[Int])
}

使用Flight。FlightExt

val (ctx, sctx) = SparkUtil.createContext() // just a helper function to build context
    val flightsDataFrame = separateFlightsMock(sctx) // reads data from Parquet file

    import sctx.implicits._
    flightsDataFrame.as[Flight.FlightExt]
      .map(flight => flight.callsign)
      .show()