从Scala中的嵌套json文件创建spark数据框 [英] create a spark dataframe from a nested json file in scala

问题描述

我有一个看起来像这样的json文件

I have a json file that looks like this

{
"group" : {},
"lang" : [ 
    [ 1, "scala", "functional" ], 
    [ 2, "java","object" ], 
    [ 3, "py","interpreted" ]
]
}

我尝试使用创建数据框

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

当我运行它时，我得到

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

我们如何基于"lang"键的内容创建df?我不在乎group {}，我只需要从"lang"中提取数据并像这样应用案例类

How do we create a df based on contents of "lang" key? I do not care about group{} all I need is, pull data out of "lang" and apply case class like this

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

我已阅读此帖子使用Apache Spark阅读JSON-`corrupt_record` 并了解每个记录都必须放在换行符上，但就我而言，我无法更改文件结构

I have read this post Reading JSON with Apache Spark - `corrupt_record` and understand that each record needs to be on a newline but in my case I cannot change the file structure

推荐答案

json格式错误. sqlContext的json api正在将其读取为损坏的记录.正确的格式是

The json format is wrong. The the json api of sqlContext is reading it as corrupt record. Correct form is

{"group":{},"lang":[[1,"scala","functional"],[2,"java","object"],[3,"py","interpreted"]]}

并假设您将其保存在文件("/home/test.json")中，则可以使用以下方法获取所需的dataframe

and supposing you have it in a file ("/home/test.json"), then you can use following method to get the dataframe you want

import org.apache.spark.sql.functions._
import sqlContext.implicits._

val df = sqlContext.read.json("/home/test.json")

val df2 = df.withColumn("lang", explode($"lang"))
    .withColumn("id", $"lang"(0))
    .withColumn("langs", $"lang"(1))
    .withColumn("type", $"lang"(2))
    .drop("lang")
    .withColumnRenamed("langs", "lang")
    .show(false)

您应该拥有

+---+-----+-----------+
|id |lang |type       |
+---+-----+-----------+
|1  |scala|functional |
|2  |java |object     |
|3  |py   |interpreted|
+---+-----+-----------+

已更新

如果您不想按照下面的注释中所述更改输入的json格式，则可以使用wholeTextFiles读取json文件，然后使用parse如下所示

If you don't want to change your input json format as mentioned in your comment below, you can use wholeTextFiles to read the json file and parse it as below

import sqlContext.implicits._
import org.apache.spark.sql.functions._

val readJSON = sc.wholeTextFiles("/home/test.json")
  .map(x => x._2)
  .map(data => data.replaceAll("\n", ""))

val df = sqlContext.read.json(readJSON)

val df2 = df.withColumn("lang", explode($"lang"))
  .withColumn("id", $"lang"(0).cast(IntegerType))
  .withColumn("langs", $"lang"(1))
  .withColumn("type", $"lang"(2))
  .drop("lang")
  .withColumnRenamed("langs", "lang")

df2.show(false)
df2.printSchema

应该为您提供dataframe，为schema提供

root
 |-- id: integer (nullable = true)
 |-- lang: string (nullable = true)
 |-- type: string (nullable = true)

这篇关于从Scala中的嵌套json文件创建spark数据框的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！