通过SparkSQL读取嵌套JSON-[Analysis Exception]无法解析列

Question

我有这样的JSON数据：

{  
   "parent":[  
      {  
         "prop1":1.0,
         "prop2":"C",
         "children":[  
            {  
               "child_prop1":[  
                  "3026"
               ]
            }
         ]
      }
   ]
}

从Spark读取数据后，我得到以下模式：

val df = spark.read.json("test.json")

df.printSchema
root
 |-- parent: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- children: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- child_prop1: array (nullable = true)
 |    |    |    |    |    |-- element: string (containsNull = true)
 |    |    |-- prop1: double (nullable = true)
 |    |    |-- prop2: string (nullable = true)

现在，我想从df中选择child_prop1。但是当我尝试选择它时，我得到了org.apache.spark.sql.分析异常。像这样：

df.select("parent.children.child_prop1")

org.apache.spark.sql.AnalysisException: cannot resolve '`parent`.`children`['child_prop1']' due to data type mismatch: argument 2 requires integral type, however, ''child_prop1'' is of string type.;;
'Project [parent#60.children[child_prop1] AS child_prop1#63]
+- Relation[parent#60] json

  at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$2.applyOrElse(CheckAnalysis.scala:82)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$2.applyOrElse(CheckAnalysis.scala:74)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:310)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:310)
  at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:309)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:307)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:307)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$5.apply(TreeNode.scala:331)
  at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:188)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformChildren(TreeNode.scala:329)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:307)
  at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpressionUp$1(QueryPlan.scala:282)
  at org.apache.spark.sql.catalyst.plans.QueryPlan.org$apache$spark$sql$catalyst$plans$QueryPlan$$recursiveTransform$2(QueryPlan.scala:292)
  at org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$org$apache$spark$sql$catalyst$plans$QueryPlan$$recursiveTransform$2$1.apply(QueryPlan.scala:296)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
  at scala.collection.AbstractTraversable.map(Traversable.scala:104)
  at org.apache.spark.sql.catalyst.plans.QueryPlan.org$apache$spark$sql$catalyst$plans$QueryPlan$$recursiveTransform$2(QueryPlan.scala:296)
  at org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$7.apply(QueryPlan.scala:301)
  at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:188)
  at org.apache.spark.sql.catalyst.plans.QueryPlan.transformExpressionsUp(QueryPlan.scala:301)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:74)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:67)
  at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:128)
  at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.checkAnalysis(CheckAnalysis.scala:67)
  at org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyzer.scala:57)
  at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:48)
  at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:63)
  at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$withPlan(Dataset.scala:2822)
  at org.apache.spark.sql.Dataset.select(Dataset.scala:1121)
  at org.apache.spark.sql.Dataset.select(Dataset.scala:1139)
  ... 48 elided

尽管如此，当我从df中仅选择children时，效果很好。

df.select("parent.children").show(false)

+------------------------------------+
|children                            |
+------------------------------------+
|[WrappedArray([WrappedArray(3026)])]|
+------------------------------------+

我无法理解为什么它会出现异常，即使该列存在于数据帧中。

感谢您的帮助！

Answer 1

如果查看模式，则子prop1在根数组父数组的嵌套数组中。所以我们需要能够定义child_prop1的位置，这就是错误建议您定义的
转换json格式应该可以做到这一点
将json更改为

{"parent":{"prop1":1.0,"prop2":"C","children":{"child_prop1":["3026"]}}}

并应用

df.select("parent.children.child_prop1").show(false)

将输出为

+-----------+
|child_prop1|
+-----------+
|[3026]     |
+-----------+

以及将json更改为

{"parent":{"prop1":1.0,"prop2":"C","children":[{"child_prop1":["3026"]}]}}

并应用

df.select("parent.children.child_prop1").show(false)

将导致

+--------------------+
|child_prop1         |
+--------------------+
|[WrappedArray(3026)]|
+--------------------+

我希望答案有帮助

Answer 2

您的Json是一个有效的json，我认为您不需要更改输入数据。

使用explode将数据作为

import org.apache.spark.sql.functions.explode

val data = spark.read.json("src/test/java/data.json")
val child = data.select(explode(data("parent.children"))).toDF("children")

child.select(explode(child("children.child_prop1"))).toDF("child_prop1").show()

如果您可以更改输入数据，您可以遵循@raesh建议