json - Spark + Json4s 序列化问题

Question

我在 Spark 2.2.0 闭包中使用 Json4s 类。序列化失败的“解决方法”DefaultFormats是包括他们的定义 inside every closure executed by Spark那需要他们。我相信我在下面所做的比我需要的要多，但仍然出现序列化失败。

使用 Spark 2.2.0、Scala 2.11、Json4s 3.2.x(Spark 中的任何内容)，并尝试使用 Json4s 3.5.3，通过使用 sbt 将其拉入我的工作中。在所有情况下，我都使用了如下所示的解决方法。

有谁知道我做错了什么？

logger.info(s"Creating an RDD for $actionName")
implicit val formats = DefaultFormats
val itemProps = df.rdd.map[(ItemID, ItemProps)](row => { <--- error points to this line
  implicit val formats = DefaultFormats
  val itemId = row.getString(0)
  val correlators = row.getSeq[String](1).toList
  (itemId, Map(actionName -> JArray(correlators.map { t =>
    implicit val formats = DefaultFormats
    JsonAST.JString(t)
  })))
})

我还尝试了另一个建议，即设置 DefaultFormats隐含在类构造函数区域而不是闭包中，任何地方都没有运气。

JVM 错误跟踪来自 Spark，提示该任务不可序列化并指向上面的行(无论如何都是我代码中的最后一行)，然后解释了根本原因:

Serialization stack:
- object not serializable (class: org.json4s.DefaultFormats$, value: org.json4s.DefaultFormats$@7fdd29f3)
- field (class: com.actionml.URAlgorithm, name: formats, type: class org.json4s.DefaultFormats$)
- object (class com.actionml.URAlgorithm, com.actionml.URAlgorithm@2dbfa972)
- field (class: com.actionml.URAlgorithm$$anonfun$udfLLR$1, name: $outer, type: class com.actionml.URAlgorithm)
- object (class com.actionml.URAlgorithm$$anonfun$udfLLR$1, <function3>)
- field (class: org.apache.spark.sql.catalyst.expressions.ScalaUDF$$anonfun$4, name: func$4, type: interface scala.Function3)
- object (class org.apache.spark.sql.catalyst.expressions.ScalaUDF$$anonfun$4, <function1>)
- field (class: org.apache.spark.sql.catalyst.expressions.ScalaUDF, name: f, type: interface scala.Function1)
- object (class org.apache.spark.sql.catalyst.expressions.ScalaUDF, UDF(input[2, bigint, false], input[3, bigint, false], input[5, bigint, false]))
- element of array (index: 1)
- array (class [Ljava.lang.Object;, size 3)
- field (class: org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, name: references$1, type: class [Ljava.lang.Object;)
- object (class org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, <function2>)
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295)
... 128 more

Answer 1

我还有一个例子。您可以使用 spark-shell 进行尝试。我希望它能帮助你。

import org.json4s._
import org.json4s.jackson.JsonMethods._

def getValue(x: String): (Int, String) = {
  implicit val formats: DefaultFormats.type = DefaultFormats
  val obj = parse(x).asInstanceOf[JObject]
  val id = (obj \ "id").extract[Int]
  val name = (obj \ "name").extract[String]
  (id, name)
}

val rdd = sc.parallelize(Array("{\"id\":0, \"name\":\"g\"}", "{\"id\":1, \"name\":\"u\"}", "{\"id\":2, \"name\":\"c\"}", "{\"id\":3, \"name\":\"h\"}", "{\"id\":4, \"name\":\"a\"}", "{\"id\":5, \"name\":\"0\"}"))
rdd.map(x => getValue(x)).collect