在Spark上使用json4s的NotSerializableException

Question

基本上，我必须使用Spark分析HDFS上的一些复杂JSON。

我使用“for理解”来（预）过滤json4s的JSON和“提取”方法，将其包装成一个case类

这个很好用！

def foo(rdd: RDD[String]) = {

case class View(C: String,b: Option[Array[List[String]]],  t: Time)
case class Time($numberLong: String)
implicit val formats = DefaultFormats

rdd.map { jsonString =>
  val jsonObj = parse(jsonString)
  val listsOfView = for {
    JObject(value) <- jsonObj
    JField(("v"), JObject(views)) <- value
    normalized <- views.map(x => (x._2))
  } yield normalized
}

到目前为止还不错！

当我尝试将（预）过滤的JSON提取到我的CaseClass时，我得到了以下结果：

线程“main”组织中出现异常。阿帕奇。火花SparkException:作业因阶段失败而中止：任务不可序列化：java。伊奥。NotSerializableException:org。json4s。默认格式$

下面是带抽取的代码：

def foo(rdd: RDD[String]) = {

case class View(C: String,b: Option[Array[List[String]]],  t: Time)
case class Time($numberLong: String)
implicit val formats = DefaultFormats

rdd.map { jsonString =>
  val jsonObj = parse(jsonString)
  val listsOfView = for {
    JObject(value) <- jsonObj
    JField(("v"), JObject(views)) <- value
    normalized <- views.map(x => (x._2))
  } yield normalized.extract[View]
}

我已经在scala ws上尝试了我的代码，并且已经成功了！我对hdfs和spark真的很新，所以我希望能得到一个提示。

Answer 1

解决我的问题的是，我在rdd中使用了implicit val formats=DefaultFormats。foreach{}循环。它解决了我的可序列化异常。

以下是我解决问题的代码片段：

case class rfId(rfId: String) {}

// ... some code here ...

 rdd.foreach { record =>
    val value = record.value()

    // Bring in default date formats etc and makes json4s serializable
    implicit val formats = DefaultFormats
    val json = parse(value)
    println(json.camelizeKeys.extract[rfId])  // Prints `rfId(ABC12345678)`
 }

Answer 2

这实际上已经被修复；JSON4S从3.3.0版开始可序列化：https://github.com/json4s/json4s/issues/137

Answer 3

Spark将RDD转换上的闭包序列化，并将其“发送”给工作人员以进行分布式执行。这就要求闭包（通常也包括包含对象）中的所有代码都应该是可序列化的。

看组织的实施。json4s。DefaultFormat$（该特征的伴随对象）：

object DefaultFormats extends DefaultFormats {
    val losslessDate = new ThreadLocal(new java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"))
    val UTC = TimeZone.getTimeZone("UTC")

}

很明显，这个对象是不可序列化的，不能这样做。（ThreadLocal本身就不可序列化）

你似乎没有在你的代码中使用Date类型，所以你能摆脱implicit val formats=DefaultFormats或者用可序列化的东西替换DefaultFormats吗？