我需要从 scala 中的嵌套 json 文件创建一个 spark 数据框答案

【问题标题】：I need to create a spark dataframe from a nested json file in scala我需要从 scala 中的嵌套 json 文件创建一个 spark 数据框
【发布时间】：2019-01-23 22:21:23
【问题描述】：

我有一个像这样的 Json 文件

{
  "tags": [
    {
      "1": "NpProgressBarTag",
      "2": "userPath",
      "3": "screen",
      "4": 6,
      "12": 9,
      "13": "buttonName",
      "16": 0,
      "17": 10,
      "18": 5,
      "19": 6,
      "20": 1,
      "35": 1,
      "36": 1,
      "37": 4,
      "38": 0,
      "39": "npChannelGuid",
      "40": "npShowGuid",
      "41": "npCategoryGuid",
      "42": "npEpisodeGuid",
      "43": "npAodEpisodeGuid",
      "44": "npVodEpisodeGuid",
      "45": "npLiveEventGuid",
      "46": "npTeamGuid",
      "47": "npLeagueGuid",
      "48": "npStatus",
      "50": 0,
      "52": "gupId",
      "54": "deviceID",
      "55": 1,
      "56": 0,
      "57": "uiVersion",
      "58": 1,
      "59": "deviceOS",
      "60": 1,
      "61": 0,
      "62": "channelLineupID",
      "63": 2,
      "64": "userProfile",
      "65": "sessionId",
      "66": "hitId",
      "67": "actionTime",
      "68": "seekTo",
      "69": "seekFrom",
      "70": "currentPosition"
    }
  ]
}

我尝试使用创建数据框

val path = "some/path/to/jsonFile.json"
val df = sqlContext.read.json(path)
df.show()

当我运行它时，我得到了

df: org.apache.spark.sql.DataFrame = [_corrupt_record: string]

我们如何根据“tags”键的内容创建一个df？我只需要从“标签”中提取数据并像这样应用案例类

case class ProgLang (id: String, type: String )

我需要将此 json 数据转换为具有两列名称的数据框 .toDF(id, Type) 谁能解释一下这个错误？

【问题讨论】：

嗯，第一个问题是您正在尝试读取 multi-line JSON，正如 docs 解释的那样，您需要在读取 JSON 时设置option(key = "multiLine", value = "true")。然而，第二个问题是这个 JSON 将为每个 "id"... 创建嵌套字段...这不是您想要的并且似乎很难使用。您是否可以控制 JSON 生成？我会 1. 将 JSON 设为一行且没有空格。 2.将结构改为{ "id": "0", "type": "NpProgressBarTag" }
@LuisMiguelMejíaSuárez 不，我无法控制 Json Generation。您是否知道如何将此文件转换为 CSV 文件，然后根据列（id，类型）加载为数据框。
使用 JSON 解析库将其转换为合适的格式。然后你可以使用 spark 来读取这些文件

标签： json scala apache-spark

【解决方案1】：

您可以使用 Circe 修改 JSON。

鉴于您的值有时是 Strings 而有时是 Numbers，这非常复杂。

import io.circe._, io.circe.parser._, io.circe.generic.semiauto._

val json = """ ... """ // your JSON here.
val doc = parse(json).right.get
val mappedDoc = doc.hcursor.downField("tags").withFocus { array =>
  array.mapArray { jsons =>
    jsons.map { json =>
      json.mapObject { o =>
        o.mapValues { v =>
          // Cast numbers to strings.
          if (v.isString) v else Json.fromString(v.asNumber.get.toString)
        }
      }
    }
  }
}

final case class ProgLang(id: String, `type`: String )
final case class Tags(tags: List[Map[String, String]])
implicit val TagsDecoder: Decoder[Tags] = deriveDecoder

val tags = mappedDoc.top.get.as[Tags]
val data = for {
  tag <- res29.tags
  (id, _type) <- tag
} yield ProgLang(id, _type)

现在你有一个ProgLang 的列表，你可以直接从它创建一个DataFrame，将它保存为一个文件，每行每个 JSON，将它保存为 CSV 文件等...
如果文件很大，您可以在转换时使用fs2 流式传输它，integrates 与 Circe 很好地配合。

免责声明：我远不是 Circe 的“专业人士”，这对于做一些看似“简单任务”的事情来说似乎过于复杂，可能有更好的/ 更清洁的方式（也许使用光学？），但是，嘿！有用！ - 无论如何，如果有人知道解决此问题的更好方法，请随时编辑问题或提供您的问题。

【讨论】：

感谢您的回复，当我尝试安装圈子依赖项时出现以下错误。 ... [警告] :::::::::::::::::::::::::::::::::::::::::::: :: [警告] :: 未解决的依赖关系 :: [警告] ::::::::::::::::::::::::::::::::::::: :::::::::: [警告] :: io.circe#circe-parser_2.10;0.10.0: 未找到 [警告] :::::::::::::::: :::::::::::::::::::::::::::::: io.circe:circe-parser_2.10:0.10.0 [警告] +- tag_decode: tag_decode_2.10:1.0 sbt.ResolveException: 未解决的依赖项：io.circe#circe-parser_2.10;0.10.0: not found ...
@R7777777 您正在尝试使用 Circe 版本 0.10.0 和 Scala 2.10 其中 Circe @987654324 @，您可以使用 Circe* 0.9.3 （代码可能会更改） 或升级到 Circe 0.11.1 和 Scala 2.12.8 （这是我用过的）.

【解决方案2】：

val path = "some/path/to/jsonFile.json"
spark.read
  .option("multiLine", true).option("mode", "PERMISSIVE")
  .json(path)

【讨论】：

【解决方案3】：

如果您的 json 文件不是很大，请尝试以下代码

    val spark = SparkSession.builder().getOrCreate()
    val df = spark.read.json(spark.sparkContext.wholeTextFiles("some/path/to/jsonFile.json").values)

【讨论】：