如何使用 Spark 和 Spray Json 读取 json 文件并转换为大小写 class

Question

我有一个包含 json 行的文本文件，其结构如下所示。

{"city": "London","street": null, "place": "Pizzaria", "foo": "Bar"}

我需要用 spark 将它读入 JSON 并用下面的 scala 代码将它转换成一个案例 class。我只需要 json 中 class 情况下给出的定义字段。

 import org.apache.spark.sql.SparkSession
 import spray.json.DefaultJsonProtocol
 import spray.json._


object SimpleExample extends DefaultJsonProtocol {

  case class Row(city: String,
                 street: Option[String],
                 place: String)

  implicit val rowFormat = jsonFormat3(Row)

  def main(args: Array[String]): Unit = {

    val logFile = "example.txt"
    val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
    val logData = spark.read.textFile(logFile).cache()

    import spark.implicits._

    val parsed = logData.map(line => line.parseJson.convertTo[Row])

    println(s"Total Count : ${parsed.count()}")

    spark.stop()
  }
}

然而，当我运行我的 spark 应用程序时，出现以下错误：

Exception in thread "main" java.lang.NoClassDefFoundError: spray/json/JsonFormat
        at java.lang.Class.getDeclaredMethods0(Native Method)
        at java.lang.Class.privateGetDeclaredMethods(Class.java:2701)
        at java.lang.Class.privateGetMethodRecursive(Class.java:3048)
        at java.lang.Class.getMethod0(Class.java:3018)
        at java.lang.Class.getMethod(Class.java:1784)
        at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:42)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:879)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain(SparkSubmit.scala:197)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:227)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:136)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: spray.json.JsonFormat
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)

我猜错误是关于进口的，但无法解决。

Answer 1

可以直接读取数据为json（不用spray-json），然后转换成数据集

import spark.implicits._

val logData = spark.read.json(logFile)
logData.select("city", "street", "place").as[Row]

只要 case class 中的变量名与文件中的变量名匹配，就可以正常工作。

如何使用 Spark 和 Spray Json 读取 json 文件并转换为大小写 class

How to read json file and convert to case class with Spark and Spray Json

json

scala

spray-json

apache-spark