【问题标题】:create a spark dataframe from a nested json file in scala [duplicate]从scala中的嵌套json文件创建一个spark数据框[重复]
【发布时间】:2017-12-24 00:05:32
【问题描述】:

我有一个像这样的 json 文件

{
"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”中提取数据并像这样应用案例类

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

我已阅读此帖子 Reading JSON with Apache Spark - `corrupt_record` 并了解每条记录都需要在换行符上,但就我而言,我无法更改文件结构

【问题讨论】:

    标签: scala apache-spark dataframe nested apache-spark-sql


    【解决方案1】:

    json 格式错误。 sqlContextjson api 将其读取为损坏的记录。正确的形式是

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

    假设你有一个文件(“/home/test.json”),那么你可以使用下面的方法来得到你想要的dataframe

    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 如下所示

    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
    

    它应该给你dataframeschema

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

    【讨论】:

    • Ramesh,我应该更清楚一点,我无法更改我的 json 文件格式,它只包含一个 json 文档,即之前共享的数据,它有换行符,而在你的情况下,它全部在一行中.
    • 它的格式与您在问题中粘贴的格式相同吗?那么如何以该格式添加另一条记录?
    • 是的,格式保持不变,在 .json 文件中,获取新数据的是 "lang" : [ 4, "ruby","interpreted" ]
    • 我想我越来越近了,我现在有一个带有组和语言模式根的 df |-- lang: array (nullable = true) | |-- 元素:数组 (containsNull = true) | | |-- element: string (containsNull = true) 这里是代码val readJSON = sc.wholeTextFiles("'home/data.json").map(x => x._2).map(data => data.replaceAll("""[\n]+""", " ")) val df = sqlContext.read.json(readJSON) df.printSchema
    • 是的,我也在考虑使用 wholeTextFiles。你答对了 。之后,您可以使用我上面的解决方案。 :) 创建 df2 的部分 :)
    【解决方案2】:

    As of Spark 2.2 可以使用multiLine 选项来处理多行 JSON 的情况。

    scala> spark.read.option("multiLine", true).json("jsonFile.json").printSchema
    root
     |-- lang: array (nullable = true)
     |    |-- element: array (containsNull = true)
     |    |    |-- element: string (containsNull = true)
    

    在 Spark 2.2 之前,请参阅 How to access sub-entities in JSON file?Read multiline JSON in Apache Spark

    【讨论】:

      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 2020-12-30
      • 2019-04-27
      • 1970-01-01
      • 2019-11-27
      • 2020-12-22
      • 2020-01-21
      • 2019-05-22
      相关资源
      最近更新 更多