【问题标题】:How to convert a Spark Dataframe to JSONObject如何将 Spark Dataframe 转换为 JSONObject
【发布时间】:2017-12-31 07:21:59
【问题描述】:

我的目标是将 DataFrame 转换为 JSONObject 的有效 JSONArray。

我目前正在使用:

val res = df.toJSON.collect()

但我得到一个 Array[String] - JSON 转义字符串数组,即:

["{\"url\":\"http://www.w3schools.com/html/html_form_action.asp?user=123\",\"subnet\":\"32.2.208.1\",\"country\":\"\",\"status_code\":\"200\"}"]

我正在寻找一种将这些字符串转换为实际 JSONObjects 的方法,我找到了一些建议 find and replace characters 的解决方案,但我正在寻找更清洁的东西。

我尝试使用 org.json 库将每个字符串转换为 JSONObject,但显然它不是可序列化的对象。

有什么建议吗?任何可以运行的快速 Scala JSON 库?

或者一般建议如何使用 toJSON 方法。

更新

这有点浪费,但这个选项对我有用:

 val res = df.toJSON.map(new JSONObject(_).toString).collect()

由于 JSONObject 不可序列化 - 我可以使用它的 toString 来获取有效的 JSON 格式。

如果您对如何改进它仍有任何建议 - 请告诉我。

【问题讨论】:

  • 你能用你的数据框的样本数据更新吗?
  • 当我使用toJSON 时,这些引号没有转义。我正在使用 Spark 1.6。
  • @philantrovert,这很奇怪,我们使用的是 Spark 1.6.1。我在网上看到的关于这个问题的所有问题,都说格式是一样的。
  • @RameshMaharjan,抱歉,我无法提供数据帧结构,这不是我的私人代码。但是,如果有帮助,我可以说它有 2 个嵌套字段。我知道我可以从数据框中提取模式,无论如何我可以使用该模式来编写 JSONObject 吗?
  • @LiranBo org.codehaus.jettison.json.JSONObject 确实是可序列化的。

标签: scala apache-spark


【解决方案1】:

您可以使用spray-json 将字符串解析为案例类:

import spray.json._
import DefaultJsonProtocol._
case class Data(url: String, subnet: String, country: String, status_code: String)
implicit val dataFormat = jsonFormat4(Data)
val source = Array("{\"url\":\"http://www.w3schools.com/html/html_form_action.asp?user=123\",\"subnet\":\"32.2.208.1\",\"country\":\"\",\"status_code\":\"200\"}")
val data = source(0).parseJson.convertTo[Data]

【讨论】:

  • 我也考虑了用例类的选项,但是有超过 80 个嵌套字段。所以创建一个会非常复杂。我希望以某种方式使用数据框架构。
  • 如果您只想从 JSON 中获取一些数据,您只需要为您需要的位定义案例类层次结构。例如,上述内容也适用于case class Data(url: String, subnet: String)。
  • 是的,我知道,但我几乎需要整个结构。我确实在一些内部 UDF 中使用案例类来操作一些数据。但结果应该包括几乎所有内容。
  • 这也是为什么我更喜欢使用 spark toJSON 函数而不是提供自己的 writer。
【解决方案2】:

您可以使用 DataframeWriter 类。

df.write.json(path)

如果输出文件有多个记录/分区,这可能会创建多个部分文件。然后您可以编写一个简单的合并实用程序来合并 hdfs/本地文件系统中的部分文件。

如果输出是一个小文件 - 你可以使用 coalesce()

df.coalesce(1).write.json(path)

然后您可以将其读回 DF。

【讨论】:

  • 感谢您的建议,但性能在这里是一个严重的问题。我无法将数据写入文件并再次读取它们。
【解决方案3】:

你可以:

  1. 收集数据框 - 你会得到Array[Row]
  2. 映射每一行折叠它到Map[String,Any] - 结果将由Array[Map[String,Any]]
  3. 序列化为 JSON
implicit val formats = DefaultFormats

val dataFrame = (1 to 10)
  .map(i => ("value" + i, i))
  .toDF("name", "value")

val maps = dataFrame
  .collect
  .map(
    row => dataFrame
      .columns
      .foldLeft(Map.empty[String, Any])
      (
        (acc, item) => acc + (item -> row.getAs[Any](item))
      )
  )

val json = Serialization.write(maps)

println(json)

【讨论】:

    【解决方案4】:
    I will show how dataframe converted into Json object list in spark.
    I/P: Dataframe
    O/P Json : [{ "id":"111","Loc":"Pune"},{"id":"2222","Loc":"Mumbai"}]
    Sol:-> 
    1] Create  Person POJO having id and loc fields.
    2] Suppose dataframe named 'myDF'
    3] myDF.collect.foreach { record =>
     val recMap = record.getValuesMap(myDF.columns).toMap[Any, Any]
     val person =new Person
     person.setLoc(recMap("LOC"))
     jsonList.append(person) //List of Person obj
    }
    val gson = new Gson //GSON lib
    jsonStr = gson.toJson(jsonList.asJava)
    

    【讨论】:

      猜你喜欢
      • 2017-03-17
      • 2021-12-23
      • 2016-05-16
      • 2017-07-08
      • 2023-03-28
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多