如何在没有 SparkSQL 的情况下使用 fastxml 解析 Spark 中的 JSON？答案

【问题标题】：How to parse JSON in Spark with fasterxml without SparkSQL?如何在没有 SparkSQL 的情况下使用 fastxml 解析 Spark 中的 JSON？
【发布时间】：2016-05-19 12:18:40
【问题描述】：

我已经走到这一步了：

import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.DeserializationFeature

case class Person(name: String, lovesPandas: Boolean)

val mapper = new ObjectMapper()

val input = sc.textFile("files/pandainfo.json")
val result = input.flatMap(record => {
    try{
        Some(mapper.readValue(record, classOf[Person]))
    } catch {
        case e: Exception => None
    }
})
result.collect

但得到Array() 结果（没有错误）。文件是https://github.com/databricks/learning-spark/blob/master/files/pandainfo.json我怎么从这里继续？

咨询Spark: broadcasting jackson ObjectMapper我试过了

import org.apache.spark._
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.DeserializationFeature

case class Person(name: String, lovesPandas: Boolean)

val input = """{"name":"Sparky The Bear", "lovesPandas":true}"""
val result = input.flatMap(record => {
    try{
        val mapper = new ObjectMapper()
        mapper.registerModule(DefaultScalaModule)
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
        Some(mapper.readValue(record, classOf[Person]))
    } catch {
        case e: Exception => None
    }
})
result.collect

得到了

Name: Compile Error
Message: <console>:34: error: overloaded method value readValue with alternatives:
  [T](x$1: Array[Byte], x$2: com.fasterxml.jackson.databind.JavaType)T <and>
  [T](x$1: Array[Byte], x$2: com.fasterxml.jackson.core.type.TypeReference[_])T <and>
  [T](x$1: Array[Byte], x$2: Class[T])T <and>

【问题讨论】：

我只是在谷歌上搜索，但你需要mapper.registerModule(DefaultScalaModule)吗？另外，您是否尝试过从 Spark 之外的文字字符串中解析 Person 只是为了检查该位是否正常工作？
@TheArchetypalPaul 这可以通过很少的谷歌帮助和一些调试来解决..
@YuvalItzchakov，嗯，是的，但我不知道你为什么要对我发表评论！
@TheArchetypalPaul：1）如果我添加这个，我会得到Name: org.apache.spark.SparkException Message: Task not serializable 2）我如何添加文字字符串？ val text = new String('{"name":"Sparky The Bear", "lovesPandas":true}') 给Message: <console>:1: error: unclosed character literal
要在带引号的字符串中包含引号，请阅读 Scala 的任何基本介绍/教程。确实，从这个问题和其他最近的问题来看，你没有花很短的时间来查找一些 Scala 基础知识，从而使你的任务变得更加困难。从简单的东西开始——让 JSON 读取工作，然后添加 Spark 部分。

标签： json scala apache-spark

【解决方案1】：

我看到您尝试了 Learning Spark 示例。这里参考完整代码 https://github.com/holdenk/learning-spark-examples/blob/master/src/main/scala/com/oreilly/learningsparkexamples/scala/BasicParseJsonWithJackson.scala E.

【讨论】：

这在 jupyter notebook 中不起作用。 Jupyter 不知道如何导入 jackson 类。有什么想法吗？
你能解释清楚吗？
您指向我的代码在 jupyter notebook 中不起作用。我开始怀疑这不是因为代码错误，而是因为笔记本。可能学习 spark 的代码和 notebook 配合不好。
也许您需要更改笔记本的元数据添加杰克逊包。

【解决方案2】：

代替sc.textfile("path\to\json")你可以试试这个（我用java写因为我不知道scala但API是一样的）：

SQLContext sqlContext = new SQLContext(sc);
DataFrame dfFromJson = sqlContext.read().json("path\to\json\file.json");

spark 将读取您的 json 文件并将其转换为数据帧。

如果你的 json 文件是嵌套的，你可以使用

org.apache.spark.sql.functions.explode(e: Column): Column

比如看我的回答here

希望对您有所帮助。

【讨论】：