如何在 Spark 中使用 Java 8 Date 类和 Jackson？答案

【问题标题】：How to use Java 8 Date classes and Jackson with Spark?如何在 Spark 中使用 Java 8 Date 类和 Jackson？
【发布时间】：2015-07-16 11:23:52
【问题描述】：

我有一个 Spark 1.4.0 项目，我正在尝试使用 Jackson 和 JSR-310 module 解析多个包含时间戳字段的 JSON 记录并将其存储在 ZonedDateTime 对象中。如果我尝试从 IDE（即 IntelliJ IDEA 14.0）运行驱动程序，它会正常运行，但如果我使用 sbt assembly 和 spark-submit，则会出现以下异常：

15/07/16 14:13:03 ERROR Executor: Exception in task 3.0 in stage 0.0 (TID 3)
java.lang.AbstractMethodError: com.mycompany.input.EventParser$$anonfun$1$$anon$1.com$fasterxml$jackson$module$scala$experimental$ScalaObjectMapper$_setter_$com$fasterxml$jackson$module$scala$experimental$ScalaObjectMapper$$typeCache_$eq(Lorg/spark-project/guava/cache/LoadingCache;)V
    at com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper$class.$init$(ScalaObjectMapper.scala:50)
    at com.mycompany.input.EventParser$$anonfun$1$$anon$1.<init>(EventParser.scala:27)
    at com.mycompany.input.EventParser$$anonfun$1.apply(EventParser.scala:27)
    at com.mycompany.input.EventParser$$anonfun$1.apply(EventParser.scala:24)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:686)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:686)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
    at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:69)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:242)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:70)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
    at org.apache.spark.scheduler.Task.run(Task.scala:70)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我尝试了几个版本的汇编、杰克逊和火花，但没有运气。我想这在某种程度上与 spark 和我的项目（不知何故，使用 Guava 库）之间的依赖冲突有关。有什么想法吗？

谢谢！

编辑：重现问题here的示例项目。

【问题讨论】：

检查 spark 自己的 lib 目录中所有 jar 的版本（例如 jackson）。然后确保您针对这些完全相同的版本进行构建。
@Imm 似乎 Spark 1.4.0 使用的是 Jackson 2.4.4，这与我在项目中使用的版本相同。尽管如此，还是没有运气。同样的问题仍然存在。
你使用的是同一个版本的scala，对吧？
@Imm 是的，我尝试使用相同版本的 Scala (2.10.4)。此外，我自己使用 Scala 2.11 编译了 Spark，我还制作了一个等效的 Maven 项目来尝试遮蔽有问题的库。但仍然没有运气:(
不幸的是，很明显，系统的某些部分在使用与编译使用不同的 Guava 版本时遇到问题。我的另一个建议是尝试升级到 Jackson 2.5（见答案）；这将消除 Guava 依赖项之一，因为 Scala 模块的 2.5 不再依赖 Guava。

标签： java scala apache-spark jackson

【解决方案1】：

我遇到了类似的问题，并通过更改两件事解决了它：

1) 我使用 ObjectMapper 而不是 ScalaObjectMapper，正如对此 SO 问题的评论中所建议的那样：Error in running job on Spark 1.4.0 with Jackson module with ScalaObjectMapper

2) 我需要在映射操作中定义映射器。

val alertsData = sc.textFile(rawlines).map(alertStr => {
      val mapper = new ObjectMapper()
      mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
      mapper.registerModule(DefaultScalaModule)
      broadcastVar.value.readValue(alertStr, classOf[Alert])
    })

如果映射器是在外部定义的，我会得到 NullPointerException。也试过播，还是不行。

此外，不需要像 spark 提供的那样显式添加 jackson 作为依赖项。

希望这会有所帮助。

艾丽莎

【讨论】：

类似于Learning Spark 示例，但不要使用ScalaObjectMapper github.com/databricks/learning-spark/blob/master/src/main/scala/…
@Aliza 但这不会为 RDD 的每个元素创建一个 ObjectMapper 吗？这听起来有点矫枉过正。如果我错了，请纠正我。
@kaktusito 你是对的。请参阅我在这里提出的解决方案：stackoverflow.com/questions/32495891/…
@Aliza 感谢您的解决方案。在看到这个之前，我想通了。我正在将一个 Scalding 程序翻译成 Spark 并遇到了这个问题。似乎 Spark 并没有像 Scalding 对 Job 类那样序列化和分发整个类。手动广播 JSON 映射器仍然感觉很奇怪。

【解决方案2】：

升级到 Jackson 2.5 可能会有所帮助。虽然 Jackson Scala 模块在 2.4 版之前确实依赖于 Guava，但从 2.5 版中删除了此依赖项（测试有 test 依赖项，但运行时没有依赖项）。这至少会消除传递依赖冲突。

【讨论】：

我也按照您的建议尝试过，但也没有帮助。我上传了一个非常简单的项目来重现 github 中的问题。请参阅我的问题中的编辑。感谢您的帮助。