在Scala中使用from_json()来解析DataFrame中的多行

Question

我在Spark DataFrame的一个列中有一个JSON，如下所示：

ID|           Text|           JSON
------------------------------------------------------------------------------
1|             xyz|          [{"Hour": 1, "Total": 10, "Fail": 1}, {"Hour": 2, "Total": 40, "Fail": 4}, {"Hour": 3, "Total": 20, "Fail": 2}]

我正在使用以下架构

val schema = StructType(Array(StructField("Hour", IntegerType),
   StructField("Total", IntegerType), StructField("Fail", IntegerType))

我使用以下代码来解析DataFrame并将其输出为多列

val newDF = DF.withColumn("JSON", from_json(col("JSON"), schema)).select(col("JSON.*"))
newDF.show()

上面的代码只解析JSON中的一条记录。但是，我想让它解析JSON中的所有记录。

输出如下：

Hour|       Total|       Fail|
-------------------------------
   1|          10|          1|
-------------------------------

但是，我希望输出如下所示：

Hour|       Total|       Fail|
-------------------------------
   1|          10|          1|
   2|          40|          4|
   3|          20|          2|
-------------------------------

有没有人，请告诉我。我到底错过了什么！

提前谢谢。

Answer 1

如果我正确地解释了您的示例数据，那么您的JSON列是一系列带有已发布模式的JSON元素。在应用from_json之前，您需要分解列，如下所示：

val df = Seq(
  (1, "xyz", Seq("""{"Hour": 1, "Total": 10, "Fail": 1}""",
                 """{"Hour": 2, "Total": 40, "Fail": 4}""",
                 """{"Hour": 3, "Total": 20, "Fail": 2}""")
  )).toDF("ID", "Text", "JSON")

import org.apache.spark.sql.types._

val jsonSchema = StructType(Array(
  StructField("Hour", IntegerType),
  StructField("Total", IntegerType),
  StructField("Fail", IntegerType)
))

df.
  withColumn("JSON", explode(col("JSON"))).
  withColumn("JSON", from_json(col("JSON"), jsonSchema)).
  select("JSON.*").
  show
// +----+-----+----+
// |Hour|Total|Fail|
// +----+-----+----+
// |   1|   10|   1|
// |   2|   40|   4|
// |   3|   20|   2|
// +----+-----+----+