使用 json4s 解析 JSON 时引发不可序列化的异常

Posted 2023-02-22

【中文标题】使用 json4s 解析 JSON 时引发不可序列化的异常

【英文标题】：Spark non-serializable exception when parsing JSON with json4s

【发布时间】：2015-06-22 02:47:00

【问题描述】：

我在 Spark 作业中尝试解析 json 时遇到了问题。我正在使用spark 1.1.0、json4s 和Cassandra Spark Connector。抛出的异常是：

java.io.NotSerializableException: org.json4s.DefaultFormats

检查 DefaultFormats 伴随对象，并通过这个stack 问题，很明显 DefaultFormats 不能被序列化。现在的问题是该怎么做。

我可以看到这个ticket 显然通过添加关键字transient 在spark 代码库中解决了这个问题，但我不确定如何或在哪里将它应用到我的案例中。解决方案是否只在执行程序上实例化 DefaultFormats 类，以避免一起序列化？人们正在使用另一个用于 scala/spark 的 JSON 解析库吗？我最初尝试单独使用杰克逊，但遇到了一些我无法轻松解决的注释错误，并且 json4s 开箱即用。这是我的代码：

import org.json4s._
import org.json4s.jackson.JsonMethods._
implicit val formats = DefaultFormats

val count = rdd.map(r => checkUa(r._2, r._1)).reduce((x, y) => x + y) 

我在 checkUa 函数中进行 json 解析。我尝试让 count 变得懒惰，希望它以某种方式延迟执行，但它没有效果。也许在 checkUA 中移动隐式 val？非常感谢任何建议。

【参考方案1】：

an open ticket with json4s 已经回答了这个问题。解决方法是将implicit 声明放在函数内

val count = rdd
               .map(r => implicit val formats = DefaultFormats; checkUa(r._2, r._1))
               .reduce((x, y) => x + y) 

【讨论】：

谢谢你的回答，它就像一个魅力。这是我遇到的另一个火花 JSON4s 问题......再次卡住了。 ***.com/questions/29666487/…

这似乎对我不起作用：***.com/questions/48454611/…

【参考方案2】：

当我将 implicit val formats = ... 声明放在包含解析的方法中时，我遇到了同样的错误，而不是在类（对象）上声明它。

所以这会引发错误：

object Application 

  //... Lots of other code here, which eventually calls 
  // setupStream(...)

  def setupStream(streamingContext: StreamingContext,
                          brokers: String,
                          topologyTopicName: String) = 
    implicit val formats = DefaultFormats
    _createDStream(streamingContext, brokers, topologyTopicName)
      // Remove the message key, which is always null in our case
      .map(_._2)
      .map((json: String) => parse(json).camelizeKeys
        .extract[Record[TopologyMetadata, Unused]])
      .print()

不过这样就好了：

object Application 

  implicit val formats = DefaultFormats

  //... Lots of other code here, which eventually calls 
  // setupStream(...)

  def setupStream(streamingContext: StreamingContext,
                          brokers: String,
                          topologyTopicName: String) = 
    _createDStream(streamingContext, brokers, topologyTopicName)
      // Remove the message key, which is always null in our case
      .map(_._2)
      .map((json: String) => parse(json).camelizeKeys
        .extract[Record[TopologyMetadata, Unused]])
      .print()

【讨论】：

这似乎对我不起作用：***.com/questions/48454611/…